सराव मध्ये स्पार्क योजना उत्क्रांती

प्रिय वाचकांनो, शुभ दिवस!

या लेखात, Neoflex च्या बिग डेटा सोल्युशन्स व्यवसाय क्षेत्रातील आघाडीचे सल्लागार Apache Spark वापरून व्हेरिएबल स्ट्रक्चर शोकेस तयार करण्याच्या पर्यायांचे तपशीलवार वर्णन करतात.

डेटा विश्लेषण प्रकल्पाचा एक भाग म्हणून, ढिले संरचित डेटावर आधारित स्टोअरफ्रंट तयार करण्याचे कार्य अनेकदा उद्भवते.

सहसा हे लॉग किंवा विविध सिस्टीममधील प्रतिसाद असतात, जेएसओएन किंवा एक्सएमएल म्हणून सेव्ह केले जातात. डेटा Hadoop वर अपलोड केला जातो, त्यानंतर तुम्हाला त्यांच्याकडून स्टोअरफ्रंट तयार करणे आवश्यक आहे. आम्ही तयार केलेल्या शोकेसमध्ये प्रवेश आयोजित करू शकतो, उदाहरणार्थ, Impala द्वारे.

या प्रकरणात, लक्ष्य स्टोअरफ्रंटची योजना आधीपासून ज्ञात नाही. शिवाय, योजना देखील आगाऊ तयार केली जाऊ शकत नाही, कारण ती डेटावर अवलंबून असते, आणि आम्ही या अतिशय ढिले संरचित डेटावर काम करत आहोत.

उदाहरणार्थ, आज खालील प्रतिसाद लॉग केला आहे:

{source: "app1", error_code: ""}

आणि उद्या त्याच प्रणालीतून पुढील उत्तर येईल:

{source: "app1", error_code: "error", description: "Network error"}

परिणामी, शोकेसमध्ये आणखी एक फील्ड जोडले पाहिजे - वर्णन, आणि ते येईल की नाही हे कोणालाही माहिती नाही.

अशा डेटावर स्टोअरफ्रंट तयार करण्याचे कार्य खूपच मानक आहे आणि यासाठी स्पार्ककडे अनेक साधने आहेत. स्त्रोत डेटा पार्स करण्यासाठी, JSON आणि XML दोन्हीसाठी समर्थन आहे आणि पूर्वी अज्ञात स्कीमासाठी, स्कीमाइव्होल्यूशनसाठी समर्थन प्रदान केले आहे.

पहिल्या दृष्टीक्षेपात, उपाय सोपे दिसते. तुम्हाला JSON सह फोल्डर घ्यायचे आहे आणि ते डेटाफ्रेममध्ये वाचणे आवश्यक आहे. स्पार्क स्कीमा तयार करेल, नेस्टेड डेटाला स्ट्रक्चर्समध्ये बदलेल. पुढे, Hive मेटास्टोरमध्ये स्टोअरफ्रंटची नोंदणी करून, सर्व काही पर्केटमध्ये जतन करणे आवश्यक आहे, ज्याला इम्पालामध्ये देखील समर्थन आहे.

सर्व काही सोपे असल्याचे दिसते.

तथापि, सरावातील अनेक समस्यांचे काय करावे हे दस्तऐवजीकरणातील लहान उदाहरणांवरून स्पष्ट होत नाही.

दस्तऐवजीकरण स्टोअरफ्रंट तयार करण्यासाठी नाही, परंतु डेटाफ्रेममध्ये JSON किंवा XML वाचण्याच्या दृष्टिकोनाचे वर्णन करते.

बहुदा, ते JSON कसे वाचायचे आणि पार्स करायचे ते दाखवते:

df = spark.read.json(path...)

स्पार्कला डेटा उपलब्ध करून देण्यासाठी हे पुरेसे आहे.

सराव मध्ये, स्क्रिप्ट फक्त फोल्डरमधून JSON फायली वाचण्यापेक्षा आणि डेटाफ्रेम तयार करण्यापेक्षा खूपच क्लिष्ट आहे. परिस्थिती अशी दिसते: आधीच एक विशिष्ट स्टोअरफ्रंट आहे, दररोज नवीन डेटा येतो, त्यांना स्टोअरफ्रंटमध्ये जोडणे आवश्यक आहे, हे विसरू नका की योजना भिन्न असू शकते.

शोकेस बांधण्याची नेहमीची योजना खालीलप्रमाणे आहे:

चरण 1. त्यानंतरच्या दैनिक रीलोडिंगसह डेटा Hadoop मध्ये लोड केला जातो आणि नवीन विभाजनामध्ये जोडला जातो. दिवसा विभाजित केलेल्या प्रारंभिक डेटासह एक फोल्डर बाहेर वळते.

चरण 2. प्रारंभिक लोड दरम्यान, हे फोल्डर स्पार्कद्वारे वाचले आणि विश्लेषित केले जाते. परिणामी डेटाफ्रेम पार्सेबल फॉरमॅटमध्ये सेव्ह केली जाते, उदाहरणार्थ, पर्केटमध्ये, जी नंतर इम्पालामध्ये आयात केली जाऊ शकते. हे या बिंदूपर्यंत जमा झालेल्या सर्व डेटासह लक्ष्य शोकेस तयार करते.

चरण 3. एक डाउनलोड तयार केले आहे जे दररोज स्टोअरफ्रंट अद्यतनित करेल.
वाढीव लोडिंगचा प्रश्न आहे, शोकेसचे विभाजन करण्याची आवश्यकता आहे आणि शोकेसची सामान्य योजना राखण्याचा प्रश्न आहे.

एक उदाहरण घेऊ. समजा रेपॉजिटरी बनवण्याची पहिली पायरी लागू केली गेली आहे आणि JSON फाइल्स फोल्डरमध्ये अपलोड केल्या आहेत.

त्यांच्याकडून डेटाफ्रेम तयार करणे, नंतर ते शोकेस म्हणून जतन करणे, ही समस्या नाही. ही पहिली पायरी आहे जी स्पार्क दस्तऐवजीकरणात सहजपणे आढळू शकते:

df = spark.read.option("mergeSchema", True).json(".../*") 
df.printSchema()

root 
|-- a: long (nullable = true) 
|-- b: string (nullable = true) 
|-- c: struct (nullable = true) |    
|-- d: long (nullable = true)

सर्व काही ठीक आहे असे दिसते.

आम्ही JSON वाचतो आणि विश्लेषित करतो, त्यानंतर आम्ही डेटाफ्रेमला पर्केट म्हणून सेव्ह करतो, कोणत्याही सोयीस्कर पद्धतीने Hive मध्ये नोंदणी करतो:

df.write.format(“parquet”).option('path','<External Table Path>').saveAsTable('<Table Name>')

आम्हाला एक खिडकी मिळते.

परंतु, दुसऱ्या दिवशी, स्त्रोताकडून नवीन डेटा जोडला गेला. आमच्याकडे JSON सह फोल्डर आहे आणि या फोल्डरमधून शोकेस तयार केला आहे. स्त्रोताकडून डेटाची पुढील बॅच लोड केल्यानंतर, डेटा मार्टमध्ये एक दिवसाचा डेटा गहाळ आहे.

तार्किक उपाय म्हणजे स्टोअरफ्रंटचे दिवसेंदिवस विभाजन करणे, जे प्रत्येक दुसऱ्या दिवशी नवीन विभाजन जोडण्यास अनुमती देईल. यासाठीची यंत्रणा देखील सुप्रसिद्ध आहे, स्पार्क तुम्हाला विभाजने स्वतंत्रपणे लिहू देते.

प्रथम, आम्ही प्रारंभिक लोड करतो, वर वर्णन केल्याप्रमाणे डेटा जतन करतो, फक्त विभाजन जोडतो. या क्रियेला स्टोअरफ्रंट इनिशिएलायझेशन म्हणतात आणि फक्त एकदाच केले जाते:

df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)

दुसऱ्या दिवशी, आम्ही फक्त एक नवीन विभाजन लोड करतो:

df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable +"/date_load=" + date_load + "/")

स्कीमा अपडेट करण्यासाठी Hive मध्ये पुन्हा नोंदणी करणे बाकी आहे.
मात्र, इथेच समस्या निर्माण होतात.

पहिली समस्या. लवकरच किंवा नंतर, परिणामी पार्केट अवाचनीय असेल. हे पार्केट आणि JSON रिकाम्या फील्डला वेगळ्या पद्धतीने कसे वागवतात यामुळे आहे.

चला एका सामान्य परिस्थितीचा विचार करूया. उदाहरणार्थ, काल JSON आला:

День 1: {"a": {"b": 1}},

आणि आज तेच JSON असे दिसते:

День 2: {"a": null}

समजा आपल्याकडे दोन भिन्न विभाजने आहेत, प्रत्येकाची एक ओळ आहे.
जेव्हा आम्ही संपूर्ण स्त्रोत डेटा वाचतो, तेव्हा स्पार्क प्रकार निर्धारित करण्यात सक्षम होईल आणि "a" हे "संरचना" प्रकारचे फील्ड आहे, ज्यामध्ये INT प्रकारातील "b" नेस्टेड फील्ड आहे हे समजेल. परंतु, जर प्रत्येक विभाजन स्वतंत्रपणे जतन केले गेले असेल, तर आम्हाला विसंगत विभाजन योजनांसह एक पार्केट मिळेल:

df1 (a: <struct<"b": INT>>)
df2 (a: STRING NULLABLE)

ही परिस्थिती सर्वज्ञात आहे, म्हणून एक पर्याय विशेष जोडला गेला आहे - स्त्रोत डेटा पार्स करताना, रिक्त फील्ड काढा:

df = spark.read.json("...", dropFieldIfAllNull=True)

या प्रकरणात, पार्केटमध्ये विभाजने असतील जी एकत्र वाचली जाऊ शकतात.
ज्यांनी हे व्यवहारात केले आहे ते येथे कडवटपणे हसतील. का? होय, कारण आणखी दोन परिस्थिती असण्याची शक्यता आहे. किंवा तीन. किंवा चार. प्रथम, जे जवळजवळ निश्चितपणे घडेल, ते म्हणजे भिन्न JSON फायलींमध्ये संख्यात्मक प्रकार भिन्न दिसतील. उदाहरणार्थ, {intField: 1} आणि {intField: 1.1}. जर अशी फील्ड एका विभाजनात आढळली, तर स्कीमा मर्ज सर्वकाही योग्यरित्या वाचेल, ज्यामुळे सर्वात अचूक प्रकार मिळेल. परंतु जर भिन्न असतील तर, एकाकडे intField: int असेल आणि दुसर्‍याकडे intField: दुहेरी असेल.

ही परिस्थिती हाताळण्यासाठी खालील ध्वज आहे:

df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)

आता आमच्याकडे एक फोल्डर आहे जिथे विभाजने आहेत जी एकाच डेटाफ्रेममध्ये वाचली जाऊ शकतात आणि संपूर्ण शोकेसची वैध पार्केट आहे. होय? नाही.

आम्ही हे लक्षात ठेवले पाहिजे की आम्ही पोळ्यामध्ये टेबलची नोंदणी केली आहे. फील्ड नावांमध्ये पोळे केस सेन्सेटिव्ह नसतात, तर पर्केट केस सेन्सेटिव्ह असते. म्हणून, स्कीमासह विभाजने: field1: int, आणि Field1: int हे Hive साठी समान आहेत, परंतु Spark साठी नाहीत. फील्डची नावे लोअरकेसमध्ये बदलण्यास विसरू नका.

त्यानंतर, सर्वकाही ठीक असल्याचे दिसते.

तथापि, सर्व इतके सोपे नाही. दुसरी, सुप्रसिद्ध समस्या आहे. प्रत्येक नवीन विभाजन स्वतंत्रपणे सेव्ह केल्यामुळे, विभाजन फोल्डरमध्ये स्पार्क सेवा फाइल्स असतील, उदाहरणार्थ, _SUCCESS ऑपरेशन सक्सेस फ्लॅग. यामुळे पर्केट करण्याचा प्रयत्न करताना त्रुटी येईल. हे टाळण्यासाठी, स्पार्कला फोल्डरमध्ये सेवा फाइल्स जोडण्यापासून रोखण्यासाठी तुम्हाला कॉन्फिगरेशन कॉन्फिगर करणे आवश्यक आहे:

hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")

असे दिसते की आता दररोज एक नवीन पर्केट विभाजन लक्ष्य शोकेस फोल्डरमध्ये जोडले जाते, जेथे दिवसासाठी पार्स केलेला डेटा स्थित आहे. आम्‍ही आधीच काळजी घेतली होती की डेटा टाईप विरोधाभास असलेले कोणतेही विभाजन नाहीत.

पण, आमची तिसरी समस्या आहे. आता सामान्य स्कीमा माहित नाही, शिवाय, Hive मधील टेबलमध्ये चुकीची स्कीमा आहे, कारण प्रत्येक नवीन विभाजनाने बहुधा स्कीमामध्ये विकृती आणली आहे.

तुम्हाला टेबलची पुन्हा नोंदणी करावी लागेल. हे अगदी सोप्या पद्धतीने करता येते: स्टोअरफ्रंटचे पार्केट पुन्हा वाचा, स्कीमा घ्या आणि त्यावर आधारित DDL तयार करा, ज्याद्वारे Hive मधील फोल्डरची बाह्य सारणी म्हणून पुन्हा नोंदणी करा, लक्ष्य स्टोअरफ्रंटची स्कीमा अद्यतनित करा.

आमची चौथी समस्या आहे. जेव्हा आम्ही पहिल्यांदा टेबलची नोंदणी केली तेव्हा आम्ही स्पार्कवर अवलंबून होतो. आता आम्ही ते स्वतः करतो आणि आम्हाला हे लक्षात ठेवण्याची गरज आहे की पोळ्यासाठी परवानगी नसलेल्या वर्णांसह पोर्केट फील्ड सुरू होऊ शकतात. उदाहरणार्थ, स्पार्क "भ्रष्ट_रेकॉर्ड" फील्डमध्‍ये विश्लेषित करू शकत नाही अशा ओळी बाहेर टाकतो. निसटल्याशिवाय अशा फील्डची पोळ्यामध्ये नोंदणी केली जाऊ शकत नाही.

हे जाणून घेतल्यावर, आम्हाला योजना मिळते:

f_def = ""
for f in pf.dtypes:
  if f[0] != "date_load":
    f_def = f_def + "," + f[0].replace("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") 
table_define = "CREATE EXTERNAL TABLE jsonevolvtable (" + f_def[1:] + " ) "
table_define = table_define + "PARTITIONED BY (date_load string) STORED AS PARQUET LOCATION '/user/admin/testJson/testSchemaEvolution/pq/'"
hc.sql("drop table if exists jsonevolvtable")
hc.sql(table_define)

कोड ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").बदला("अॅरे<`", "अॅरे<") सुरक्षित DDL बनवते, म्हणजे त्याऐवजी:

create table tname (_field1 string, 1field string)

"_field1, 1field" सारख्या फील्ड नावांसह सुरक्षित DDL तयार केले जाते जेथे फील्डची नावे एस्केप केली जातात: टेबल `tname` (`_field1` स्ट्रिंग, `1field` स्ट्रिंग) तयार करा.

प्रश्न उद्भवतो: संपूर्ण स्कीमा (पीएफ कोडमध्ये) योग्यरित्या डेटाफ्रेम कशी मिळवायची? हा पीएफ कसा मिळवायचा? ही पाचवी समस्या आहे. लक्ष्य शोकेसच्या पार्केट फाइल्ससह फोल्डरमधील सर्व विभाजनांची योजना पुन्हा वाचायची? ही पद्धत सर्वात सुरक्षित आहे, परंतु कठीण आहे.

स्कीमा आधीच Hive मध्ये आहे. संपूर्ण टेबलचा स्कीमा आणि नवीन विभाजन एकत्र करून तुम्ही नवीन स्कीमा मिळवू शकता. म्हणून तुम्हाला Hive वरून टेबल स्कीमा घ्यावा लागेल आणि नवीन विभाजनाच्या स्कीमासह एकत्र करा. हे Hive मधील चाचणी मेटाडेटा वाचून, तात्पुरत्या फोल्डरमध्ये सेव्ह करून आणि एकाच वेळी दोन्ही विभाजने वाचण्यासाठी स्पार्क वापरून केले जाऊ शकते.

खरं तर, आपल्याला आवश्यक असलेली प्रत्येक गोष्ट आहे: Hive मधील मूळ टेबल स्कीमा आणि नवीन विभाजन. आमच्याकडे डेटा देखील आहे. स्टोअरफ्रंट स्कीमा आणि तयार केलेल्या विभाजनामधून नवीन फील्ड एकत्रित करणारी नवीन स्कीमा मिळविण्यासाठी हे फक्त राहते:

from pyspark.sql import HiveContext
from pyspark.sql.functions import lit
hc = HiveContext(spark)
df = spark.read.json("...", dropFieldIfAllNull=True)
df.write.mode("overwrite").parquet(".../date_load=12-12-2019")
pe = hc.sql("select * from jsonevolvtable limit 1")
pe.write.mode("overwrite").parquet(".../fakePartiton/")
pf = spark.read.option("mergeSchema", True).parquet(".../date_load=12-12-2019/*", ".../fakePartiton/*")

पुढे, मागील स्निपेट प्रमाणे आम्ही टेबल नोंदणी DDL तयार करतो.
जर संपूर्ण शृंखला योग्यरित्या कार्य करत असेल, म्हणजे, आरंभिक लोड होता आणि Hive मध्ये टेबल योग्यरित्या तयार केले गेले असेल, तर आम्हाला एक अद्यतनित टेबल स्कीमा मिळेल.

आणि शेवटची अडचण अशी आहे की तुम्ही फक्त हायव्ह टेबलमध्ये विभाजन जोडू शकत नाही, कारण ते तुटले जाईल. विभाजन रचना निश्चित करण्यासाठी तुम्हाला Hive ला सक्ती करणे आवश्यक आहे:

from pyspark.sql import HiveContext
hc = HiveContext(spark) 
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)

JSON वाचणे आणि त्यावर आधारित स्टोअरफ्रंट तयार करणे या सोप्या कार्याचा परिणाम अनेक अव्यक्त अडचणींवर मात करण्यासाठी होतो, ज्यासाठी तुम्ही स्वतंत्रपणे शोधले पाहिजेत. आणि जरी हे उपाय सोपे असले तरी ते शोधण्यासाठी खूप वेळ लागतो.

शोकेसच्या बांधकामाची अंमलबजावणी करण्यासाठी, मला हे करणे आवश्यक होते:

• शोकेसमध्ये विभाजने जोडा, सेवा फाइल्सपासून मुक्त होणे
• स्पार्कने टाइप केलेल्या स्त्रोत डेटामधील रिकाम्या फील्डसह व्यवहार करा
• स्ट्रिंगवर साधे प्रकार कास्ट करा
• फील्डची नावे लोअरकेसमध्ये रूपांतरित करा
• Hive मध्ये स्वतंत्र डेटा अपलोड आणि टेबल नोंदणी (DDL जनरेशन)
• Hive शी विसंगत असू शकणार्‍या फील्डची नावे टाळण्यास विसरू नका
• Hive मध्ये टेबल नोंदणी कशी अपडेट करायची ते जाणून घ्या

सारांश, आम्ही लक्षात घेतो की दुकानाच्या खिडक्या बांधण्याचा निर्णय अनेक अडचणींनी भरलेला आहे. म्हणून, अंमलबजावणीमध्ये अडचणी आल्यास, यशस्वी तज्ञ असलेल्या अनुभवी भागीदाराशी संपर्क साधणे चांगले.

हा लेख वाचल्याबद्दल धन्यवाद, आम्हाला आशा आहे की तुम्हाला माहिती उपयुक्त वाटली.

स्त्रोत: www.habr.com