अभ्यासमा स्पार्क योजना विकास

प्रिय पाठकहरू, शुभ दिउँसो!

यस लेखमा, Neoflex को बिग डाटा समाधान व्यवसाय क्षेत्र को अग्रणी सल्लाहकार Apache Spark प्रयोग गरेर चर संरचना शोकेस निर्माण को लागी विकल्पहरु को विस्तार मा वर्णन गर्दछ।

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

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

यस अवस्थामा, लक्षित स्टोरफ्रन्टको योजना पहिले नै थाहा छैन। यसबाहेक, योजना पनि अग्रिम रूपमा कोर्न सकिँदैन, किनकि यो डाटामा निर्भर गर्दछ, र हामी यी धेरै ढिलो संरचित डाटासँग काम गरिरहेका छौं।

उदाहरण को लागी, आज निम्न प्रतिक्रिया लग छ:

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

र भोलि त्यही प्रणालीबाट निम्न जवाफ आउँछ:

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

नतिजाको रूपमा, शोकेसमा थप एउटा फिल्ड थपिनुपर्छ - विवरण, र कसैलाई थाहा छैन कि यो आउँछ कि हुँदैन।

यस्तो डाटामा स्टोरफ्रन्ट सिर्जना गर्ने कार्य राम्रो मानक हो, र स्पार्कसँग यसका लागि धेरै उपकरणहरू छन्। स्रोत डेटा पार्स गर्न को लागी, JSON र XML दुबै को लागी समर्थन छ, र पहिले अज्ञात स्कीमा को लागी, schemaEvolution को लागी समर्थन प्रदान गरिएको छ।

पहिलो नजर मा, समाधान सरल देखिन्छ। तपाईंले JSON सँग फोल्डर लिनु पर्छ र यसलाई डाटाफ्रेममा पढ्नु पर्छ। स्पार्कले स्किमा सिर्जना गर्नेछ, नेस्टेड डाटालाई संरचनाहरूमा बदल्नेछ। यसबाहेक, सबै चीजहरू पर्केटमा बचत गर्न आवश्यक छ, जुन इम्पालामा पनि समर्थित छ, Hive मेटास्टोरमा स्टोरफ्रन्ट दर्ता गरेर।

सबै कुरा सरल देखिन्छ।

यद्यपि, यो कागजातमा छोटो उदाहरणहरूबाट स्पष्ट छैन कि व्यवहारमा धेरै समस्याहरूको साथ के गर्ने।

कागजातले स्टोरफ्रन्ट सिर्जना गर्न होइन, तर डाटाफ्रेममा JSON वा XML पढ्नको लागि दृष्टिकोणको वर्णन गर्दछ।

अर्थात्, यसले JSON कसरी पढ्ने र पार्स गर्ने भनेर देखाउँछ:

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

यो Spark मा डाटा उपलब्ध गराउन पर्याप्त छ।

अभ्यासमा, स्क्रिप्ट फोल्डरबाट 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 मा पुन: दर्ता गर्न बाँकी रहेको छ।
तर, समस्या यहीँ छ ।

पहिलो समस्या। ढिलो वा पछि, नतिजा पर्केट अपठनीय हुनेछ। यो कसरी parquet र 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)

अब हामीसँग एउटा फोल्डर छ जहाँ विभाजनहरू छन् जुन एकल डाटाफ्रेममा पढ्न सकिन्छ र सम्पूर्ण शोकेसको मान्य पर्केट। हो? छैन।

हामीले याद गर्नुपर्छ कि हामीले Hive मा तालिका दर्ता गर्यौं। फिल्ड नामहरूमा Hive केस सेन्सेटिभ हुँदैन, जबकि पर्केट केस सेन्सेटिभ हुन्छ। त्यसकारण, स्कीमा: फिल्ड १: इन्ट, र फिल्ड १: इन्टको साथ विभाजनहरू Hive का लागि समान छन्, तर स्पार्कको लागि होइन। क्षेत्रको नामलाई सानो अक्षरमा रूपान्तरण गर्न नबिर्सनुहोस्।

त्यस पछि, सबै ठीक छ जस्तो लाग्छ।

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

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

यस्तो देखिन्छ कि अब हरेक दिन नयाँ पर्केट विभाजन लक्षित शोकेस फोल्डरमा थपिएको छ, जहाँ दिनको लागि पार्स गरिएको डाटा अवस्थित छ। हामीले डेटा प्रकारको द्वन्द्वको साथ कुनै पनि विभाजनहरू थिएनन् भनेर पहिले नै ख्याल राख्यौं।

तर, हामीसँग तेस्रो समस्या छ । अब सामान्य स्कीमा थाहा छैन, यसबाहेक, Hive मा तालिकामा गलत स्कीमा छ, किनकि प्रत्येक नयाँ विभाजनले प्राय: स्किमामा विरूपण प्रस्तुत गरेको छ।

तपाईंले तालिका पुन: दर्ता गर्न आवश्यक छ। यो सरल रूपमा गर्न सकिन्छ: स्टोरफ्रन्टको पार्केट फेरि पढ्नुहोस्, स्किमा लिनुहोस् र त्यसमा आधारित DDL सिर्जना गर्नुहोस्, जसको साथ Hive मा फोल्डरलाई बाह्य तालिकाको रूपमा पुन: दर्ता गर्न, लक्षित स्टोरफ्रन्टको स्कीमा अद्यावधिक गर्दै।

हामीसँग चौथो समस्या छ। जब हामीले पहिलो पटक तालिका दर्ता गर्यौं, हामी स्पार्कमा भर पर्यौं। अब हामी यो आफैं गर्छौं, र हामीले याद गर्न आवश्यक छ कि पार्केट क्षेत्रहरू Hive को लागि अनुमति छैन कि क्यारेक्टरहरूसँग सुरु हुन सक्छ। उदाहरणका लागि, स्पार्कले "भ्रष्ट_रेकर्ड" फिल्डमा पार्स गर्न नसक्ने रेखाहरू फ्याँक्छ। यस्तो क्षेत्र भागेर बिना 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]।बदल्नुहोस्(":", "`:")।बदल्नुहोस्("<", "<`")।बदल्नुहोस्(",", ",`") बदल्नुहोस्(" array<`", "array<") सुरक्षित DDL बनाउँछ, अर्थात्:

create table tname (_field1 string, 1field string)

क्षेत्र नामहरू जस्तै "_field1, 1field", सुरक्षित DDL बनाइन्छ जहाँ फिल्ड नामहरू एस्केप हुन्छन्: तालिका `tname` (`_field1` string, `1field` string) सिर्जना गर्नुहोस्।

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

स्किमा पहिले नै 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 तालिकामा विभाजन मात्र थप्न सक्नुहुन्न, किनभने यो भाँचिनेछ। तपाईंले Hive लाई यसको विभाजन संरचना ठीक गर्न बल गर्न आवश्यक छ:

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

JSON पढ्ने र त्यसमा आधारित स्टोरफ्रन्ट सिर्जना गर्ने सरल कार्यले धेरै अव्यवस्थित कठिनाइहरू पार गर्न परिणाम दिन्छ, समाधानहरू जसको लागि तपाईंले छुट्टै खोज्नुपर्छ। र यद्यपि यी समाधानहरू सरल छन्, तिनीहरूलाई फेला पार्न धेरै समय लाग्छ।

शोकेसको निर्माण कार्यान्वयन गर्नको लागि, मैले:

• शोकेसमा विभाजनहरू थप्नुहोस्, सेवा फाइलहरूबाट छुटकारा पाउँदै
• स्पार्कले टाइप गरेको स्रोत डेटामा खाली क्षेत्रहरूसँग व्यवहार गर्नुहोस्
• स्ट्रिङमा साधारण प्रकारहरू कास्ट गर्नुहोस्
• क्षेत्रका नामहरूलाई सानो अक्षरमा रूपान्तरण गर्नुहोस्
• Hive मा अलग डाटा अपलोड र तालिका दर्ता (DDL जेनरेशन)
• Hive सँग असंगत हुन सक्ने क्षेत्र नामहरू एस्केप गर्न नबिर्सनुहोस्
• Hive मा तालिका दर्ता अद्यावधिक गर्न जान्नुहोस्

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

यो लेख पढ्नु भएकोमा धन्यवाद, हामी आशा गर्दछौं कि तपाईंले जानकारी उपयोगी पाउनुभयो।

स्रोत: www.habr.com