स्पार्क स्कीमा व्यवहार में विकास

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

इस लेख में, नियोफ्लेक्स के बिग डेटा सॉल्यूशंस व्यवसाय क्षेत्र के अग्रणी सलाहकार ने अपाचे स्पार्क का उपयोग करके परिवर्तनीय संरचना शोकेस बनाने के विकल्पों का विस्तार से वर्णन किया है।

डेटा विश्लेषण परियोजना के हिस्से के रूप में, ढीले-ढाले संरचित डेटा के आधार पर स्टोरफ्रंट बनाने का कार्य अक्सर सामने आता है।

आमतौर पर ये JSON या XML के रूप में सहेजे गए विभिन्न सिस्टम से लॉग या प्रतिक्रियाएँ होती हैं। डेटा Hadoop पर अपलोड किया गया है, फिर आपको उनसे एक स्टोरफ्रंट बनाने की आवश्यकता है। हम बनाए गए शोकेस तक पहुंच को व्यवस्थित कर सकते हैं, उदाहरण के लिए, इम्पाला के माध्यम से।

इस मामले में, लक्ष्य स्टोरफ्रंट का स्कीमा पहले से ज्ञात नहीं है। इसके अलावा, योजना भी पहले से तैयार नहीं की जा सकती, क्योंकि यह डेटा पर निर्भर करती है, और हम इन बहुत ही ढीले-ढाले संरचित डेटा से निपट रहे हैं।

उदाहरण के लिए, आज निम्नलिखित प्रतिक्रिया लॉग की गई है:

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

और कल उसी प्रणाली से निम्नलिखित उत्तर आएगा:

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

परिणामस्वरूप, शोकेस में एक और फ़ील्ड जोड़ा जाना चाहिए - विवरण, और कोई नहीं जानता कि यह आएगा या नहीं।

ऐसे डेटा पर स्टोरफ्रंट बनाने का कार्य काफी मानक है, और स्पार्क के पास इसके लिए कई उपकरण हैं। स्रोत डेटा को पार्स करने के लिए, JSON और XML दोनों के लिए समर्थन है, और पहले से अज्ञात स्कीमा के लिए, schemaEvolution के लिए समर्थन प्रदान किया गया है।

पहली नज़र में, समाधान सरल लगता है. आपको JSON के साथ एक फ़ोल्डर लेना होगा और उसे डेटाफ़्रेम में पढ़ना होगा। स्पार्क एक स्कीमा बनाएगा, नेस्टेड डेटा को संरचनाओं में बदल देगा। इसके अलावा, हर चीज को लकड़ी की छत में सहेजने की जरूरत है, जो कि हाइव मेटास्टोर में स्टोरफ्रंट को पंजीकृत करके इम्पाला में भी समर्थित है।

सब कुछ सरल प्रतीत होता है.

हालाँकि, दस्तावेज़ में संक्षिप्त उदाहरणों से यह स्पष्ट नहीं है कि व्यवहार में आने वाली कई समस्याओं का क्या किया जाए।

दस्तावेज़ स्टोरफ्रंट बनाने के लिए नहीं, बल्कि 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 को पढ़ते हैं और पार्स करते हैं, फिर हम डेटाफ़्रेम को लकड़ी की छत के रूप में सहेजते हैं, इसे किसी भी सुविधाजनक तरीके से हाइव में पंजीकृत करते हैं:

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 + "/")

स्कीमा को अपडेट करने के लिए बस हाइव में फिर से पंजीकरण करना बाकी है।
हालाँकि, यहीं समस्याएँ उत्पन्न होती हैं।

पहली समस्या. देर-सबेर, परिणामी लकड़ी की छत अपठनीय हो जाएगी। इसका कारण यह है कि कैसे लकड़ी की छत और JSON खाली फ़ील्ड के साथ अलग-अलग व्यवहार करते हैं।

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

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

और आज वही JSON इस तरह दिखता है:

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

मान लीजिए कि हमारे पास दो अलग-अलग विभाजन हैं, प्रत्येक में एक पंक्ति है।
जब हम संपूर्ण स्रोत डेटा पढ़ते हैं, तो स्पार्क प्रकार निर्धारित करने में सक्षम होगा, और समझ जाएगा कि "ए" प्रकार "संरचना" का एक क्षेत्र है, जिसमें आईएनटी प्रकार का नेस्टेड फ़ील्ड "बी" है। लेकिन, यदि प्रत्येक विभाजन को अलग से सहेजा गया था, तो हमें असंगत विभाजन योजनाओं के साथ एक लकड़ी की छत मिलती है:

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)

अब हमारे पास एक फ़ोल्डर है जहां विभाजन हैं जिन्हें एकल डेटाफ्रेम और पूरे शोकेस के वैध पैराकेट में पढ़ा जा सकता है। हाँ? नहीं।

हमें याद रखना चाहिए कि हमने तालिका को हाइव में पंजीकृत किया है। फ़ील्ड नामों में हाइव केस संवेदी नहीं है, जबकि लकड़ी की छत केस संवेदी है। इसलिए, स्कीमा वाले विभाजन: फ़ील्ड1: int, और फ़ील्ड1: int हाइव के लिए समान हैं, लेकिन स्पार्क के लिए नहीं। फ़ील्ड नामों को लोअरकेस में परिवर्तित करना न भूलें।

उसके बाद सब कुछ ठीक होने लगता है.

हालाँकि, सब कुछ इतना सरल नहीं है। एक दूसरी, सर्वविदित समस्या भी है। चूँकि प्रत्येक नया विभाजन अलग से सहेजा जाता है, विभाजन फ़ोल्डर में स्पार्क सेवा फ़ाइलें होंगी, उदाहरण के लिए, _SUCCESS ऑपरेशन सफलता ध्वज। इसके परिणामस्वरूप लकड़ी की छत बनाने का प्रयास करते समय त्रुटि होगी। इससे बचने के लिए, आपको स्पार्क को फ़ोल्डर में सेवा फ़ाइलें जोड़ने से रोकने के लिए कॉन्फ़िगरेशन को कॉन्फ़िगर करने की आवश्यकता है:

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

ऐसा लगता है कि अब हर दिन लक्ष्य शोकेस फ़ोल्डर में एक नया पैराक्वेट विभाजन जोड़ा जाता है, जहां दिन के लिए पार्स किया गया डेटा स्थित होता है। हमने पहले से ही इस बात का ध्यान रखा कि डेटा प्रकार के टकराव के साथ कोई विभाजन न हो।

लेकिन, हमारे सामने एक तीसरी समस्या है. अब सामान्य स्कीमा ज्ञात नहीं है, इसके अलावा, हाइव में तालिका में एक गलत स्कीमा है, क्योंकि प्रत्येक नए विभाजन से स्कीमा में विकृति आने की संभावना है।

आपको तालिका को पुनः पंजीकृत करना होगा. यह सरलता से किया जा सकता है: स्टोरफ्रंट के पैराक्वेट को फिर से पढ़ें, स्कीमा लें और उसके आधार पर एक डीडीएल बनाएं, जिसके साथ लक्ष्य स्टोरफ्रंट के स्कीमा को अपडेट करते हुए, हाइव में फ़ोल्डर को बाहरी तालिका के रूप में फिर से पंजीकृत करें।

हमारे सामने चौथी समस्या है. जब हमने पहली बार तालिका पंजीकृत की, तो हमने स्पार्क पर भरोसा किया। अब हम इसे स्वयं करते हैं, और हमें यह याद रखने की आवश्यकता है कि लकड़ी के क्षेत्र ऐसे पात्रों से शुरू हो सकते हैं जिनकी हाइव के लिए अनुमति नहीं है। उदाहरण के लिए, स्पार्क उन पंक्तियों को फेंक देता है जिन्हें वह "correct_record" फ़ील्ड में पार्स नहीं कर सका। ऐसे फ़ील्ड को भागने के बिना हाइव में पंजीकृत नहीं किया जा सकता है।

यह जानने पर, हमें योजना मिलती है:

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)

कोड ("_करप्ट_रिकॉर्ड", "`_करप्ट_रिकॉर्ड`") + " " + एफ[1].रिप्लेस(":", "`:").रिप्लेस("<", "<`").रिप्लेस(",, ",`").प्रतिस्थापन("सरणी<`", "सरणी<") सुरक्षित DDL बनाता है, अर्थात इसके बजाय:

create table tname (_field1 string, 1field string)

"_फ़ील्ड1, 1फ़ील्ड" जैसे फ़ील्ड नामों के साथ, सुरक्षित डीडीएल बनाया जाता है जहां फ़ील्ड नाम बच जाते हैं: तालिका `tname` (`_फ़ील्ड1` स्ट्रिंग, `1फ़ील्ड` स्ट्रिंग) बनाएं।

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

स्कीमा पहले से ही हाइव में है. आप संपूर्ण तालिका के स्कीमा और नए विभाजन को मिलाकर एक नया स्कीमा प्राप्त कर सकते हैं। तो आपको हाइव से टेबल स्कीमा लेने और इसे नए विभाजन के स्कीमा के साथ संयोजित करने की आवश्यकता है। यह हाइव से परीक्षण मेटाडेटा को पढ़कर, इसे एक अस्थायी फ़ोल्डर में सहेजकर और दोनों विभाजनों को एक साथ पढ़ने के लिए स्पार्क का उपयोग करके किया जा सकता है।

वास्तव में, आपकी ज़रूरत की हर चीज़ मौजूद है: हाइव में मूल तालिका स्कीमा और नया विभाजन। हमारे पास डेटा भी है. यह केवल एक नई स्कीमा प्राप्त करने के लिए बनी हुई है जो स्टोरफ्रंट स्कीमा और बनाए गए विभाजन से नए फ़ील्ड को जोड़ती है:

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/*")

इसके बाद, हम पिछले स्निपेट की तरह, तालिका पंजीकरण डीडीएल बनाते हैं।
यदि पूरी श्रृंखला सही ढंग से काम करती है, अर्थात्, एक प्रारंभिक लोड था, और तालिका हाइव में सही ढंग से बनाई गई थी, तो हमें एक अद्यतन तालिका स्कीमा मिलती है।

और आखिरी समस्या यह है कि आप हाइव टेबल में केवल एक विभाजन नहीं जोड़ सकते, क्योंकि यह टूट जाएगा। आपको हाइव को उसकी विभाजन संरचना को ठीक करने के लिए बाध्य करने की आवश्यकता है:

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

JSON को पढ़ने और उसके आधार पर स्टोरफ्रंट बनाने का सरल कार्य कई अंतर्निहित कठिनाइयों पर काबू पाने में मदद करता है, जिनके समाधान आपको अलग से तलाशने होंगे। और यद्यपि ये समाधान सरल हैं, फिर भी इन्हें ढूंढने में बहुत समय लगता है।

शोकेस के निर्माण को लागू करने के लिए, मुझे यह करना था:

• सेवा फ़ाइलों से छुटकारा पाकर, शोकेस में विभाजन जोड़ें
• स्पार्क द्वारा टाइप किए गए स्रोत डेटा में खाली फ़ील्ड से निपटें
• सरल प्रकारों को एक स्ट्रिंग में कास्ट करें
• फ़ील्ड नामों को लोअरकेस में बदलें
• हाइव में अलग डेटा अपलोड और टेबल पंजीकरण (डीडीएल पीढ़ी)
• उन फ़ील्ड नामों से बचना न भूलें जो हाइव के साथ असंगत हो सकते हैं
• जानें कि हाइव में टेबल पंजीकरण कैसे अपडेट करें

संक्षेप में, हम ध्यान दें कि दुकान की खिड़कियां बनाने का निर्णय कई नुकसानों से भरा है। इसलिए, कार्यान्वयन में कठिनाइयों के मामले में, सफल विशेषज्ञता वाले अनुभवी भागीदार से संपर्क करना बेहतर है।

इस लेख को पढ़ने के लिए धन्यवाद, हमें उम्मीद है कि आपको जानकारी उपयोगी लगेगी।

स्रोत: www.habr.com