പ്രയോഗത്തിൽ സ്പാർക്ക് സ്കീമപരിണാമം

പ്രിയ വായനക്കാരേ, നല്ല ദിവസം!

ഈ ലേഖനത്തിൽ, നിയോഫ്ലെക്‌സിന്റെ ബിഗ് ഡാറ്റ സൊല്യൂഷൻസ് ബിസിനസ് ഏരിയയിലെ പ്രമുഖ കൺസൾട്ടന്റ് അപ്പാച്ചെ സ്പാർക്ക് ഉപയോഗിച്ച് വേരിയബിൾ സ്ട്രക്ചർ ഷോകേസുകൾ നിർമ്മിക്കുന്നതിനുള്ള ഓപ്ഷനുകൾ വിശദമായി വിവരിക്കുന്നു.

ഒരു ഡാറ്റ വിശകലന പ്രോജക്റ്റിന്റെ ഭാഗമായി, അയഞ്ഞ ഘടനാപരമായ ഡാറ്റയെ അടിസ്ഥാനമാക്കി സ്റ്റോർ ഫ്രണ്ടുകൾ നിർമ്മിക്കുന്നതിനുള്ള ചുമതല പലപ്പോഴും ഉയർന്നുവരുന്നു.

സാധാരണയായി ഇവ JSON അല്ലെങ്കിൽ XML ആയി സംരക്ഷിച്ചിരിക്കുന്ന ലോഗുകൾ അല്ലെങ്കിൽ വിവിധ സിസ്റ്റങ്ങളിൽ നിന്നുള്ള പ്രതികരണങ്ങളാണ്. ഡാറ്റ ഹഡൂപ്പിലേക്ക് അപ്‌ലോഡ് ചെയ്‌തു, തുടർന്ന് നിങ്ങൾ അവരിൽ നിന്ന് ഒരു സ്റ്റോർ ഫ്രണ്ട് നിർമ്മിക്കേണ്ടതുണ്ട്. സൃഷ്ടിച്ച ഷോകേസിലേക്കുള്ള ആക്സസ് നമുക്ക് സംഘടിപ്പിക്കാം, ഉദാഹരണത്തിന്, ഇംപാല വഴി.

ഈ സാഹചര്യത്തിൽ, ടാർഗെറ്റ് സ്റ്റോർ ഫ്രണ്ടിന്റെ സ്കീമ മുൻകൂട്ടി അറിയില്ല. മാത്രമല്ല, സ്കീമും മുൻകൂട്ടി തയ്യാറാക്കാൻ കഴിയില്ല, കാരണം ഇത് ഡാറ്റയെ ആശ്രയിച്ചിരിക്കുന്നു, മാത്രമല്ല ഞങ്ങൾ ഈ വളരെ അയഞ്ഞ ഘടനാപരമായ ഡാറ്റയാണ് കൈകാര്യം ചെയ്യുന്നത്.

ഉദാഹരണത്തിന്, ഇന്ന് ഇനിപ്പറയുന്ന പ്രതികരണം ലോഗ് ചെയ്തു:

{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 ഫയലുകൾ വായിച്ച് ഒരു ഡാറ്റാഫ്രെയിം സൃഷ്ടിക്കുന്നതിനേക്കാൾ വളരെ സങ്കീർണ്ണമാണ് സ്ക്രിപ്റ്റ്. സാഹചര്യം ഇതുപോലെ കാണപ്പെടുന്നു: ഇതിനകം ഒരു പ്രത്യേക സ്റ്റോർഫ്രണ്ട് ഉണ്ട്, എല്ലാ ദിവസവും പുതിയ ഡാറ്റ വരുന്നു, അവ സ്റ്റോറിന്റെ മുൻഭാഗത്തേക്ക് ചേർക്കേണ്ടതുണ്ട്, സ്കീം വ്യത്യാസപ്പെടാം എന്നത് മറക്കരുത്.

ഒരു ഷോകേസ് നിർമ്മിക്കുന്നതിനുള്ള സാധാരണ സ്കീം ഇപ്രകാരമാണ്:

ക്സനുമ്ക്സ ഘട്ടം. തുടർന്നുള്ള പ്രതിദിന റീലോഡിംഗിനൊപ്പം ഡാറ്റ ഹഡൂപ്പിലേക്ക് ലോഡ് ചെയ്യുകയും ഒരു പുതിയ പാർട്ടീഷനിലേക്ക് ചേർക്കുകയും ചെയ്യുന്നു. പ്രാരംഭ ഡാറ്റ ദിവസം അനുസരിച്ച് പാർട്ടീഷൻ ചെയ്ത ഒരു ഫോൾഡർ ഇത് മാറുന്നു.

ക്സനുമ്ക്സ ഘട്ടം. പ്രാരംഭ ലോഡ് സമയത്ത്, ഈ ഫോൾഡർ സ്പാർക്ക് വായിക്കുകയും പാഴ്‌സ് ചെയ്യുകയും ചെയ്യുന്നു. തത്ഫലമായുണ്ടാകുന്ന ഡാറ്റാഫ്രെയിം പാർസബിൾ ഫോർമാറ്റിൽ സംരക്ഷിക്കപ്പെടുന്നു, ഉദാഹരണത്തിന്, പാർക്ക്വെറ്റിൽ, അത് പിന്നീട് ഇംപാലയിലേക്ക് ഇറക്കുമതി ചെയ്യാൻ കഴിയും. ഈ ഘട്ടം വരെ ശേഖരിച്ച എല്ലാ ഡാറ്റയും ഉപയോഗിച്ച് ഇത് ഒരു ടാർഗെറ്റ് ഷോകേസ് സൃഷ്ടിക്കുന്നു.

ക്സനുമ്ക്സ ഘട്ടം. എല്ലാ ദിവസവും സ്റ്റോറിന്റെ മുൻഭാഗം അപ്‌ഡേറ്റ് ചെയ്യുന്ന ഒരു ഡൗൺലോഡ് സൃഷ്‌ടിച്ചു.
ഇൻക്രിമെന്റൽ ലോഡിംഗ്, ഷോകേസ് പാർട്ടീഷൻ ചെയ്യേണ്ടതിന്റെ ആവശ്യകത, ഷോകേസിന്റെ പൊതുവായ സ്കീം നിലനിർത്തുന്നതിനുള്ള ചോദ്യം എന്നിവയുണ്ട്.

നമുക്ക് ഒരു ഉദാഹരണം എടുക്കാം. ഒരു റിപ്പോസിറ്ററി നിർമ്മിക്കുന്നതിനുള്ള ആദ്യ ഘട്ടം നടപ്പിലാക്കി, 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 + "/")

സ്‌കീമ അപ്‌ഡേറ്റ് ചെയ്യുന്നതിന് ഹൈവിൽ വീണ്ടും രജിസ്റ്റർ ചെയ്യുക മാത്രമാണ് അവശേഷിക്കുന്നത്.
എന്നിരുന്നാലും, ഇവിടെയാണ് പ്രശ്നങ്ങൾ ഉണ്ടാകുന്നത്.

ആദ്യത്തെ പ്രശ്നം. താമസിയാതെ അല്ലെങ്കിൽ പിന്നീട്, തത്ഫലമായുണ്ടാകുന്ന പാർക്കറ്റ് വായിക്കാൻ കഴിയില്ല. ശൂന്യമായ ഫീൽഡുകളെ 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)

ഒരൊറ്റ ഡാറ്റാഫ്രെയിമിലേക്കും മുഴുവൻ ഷോകേസിന്റെയും സാധുവായ പാർക്കറ്റിലേക്കും വായിക്കാൻ കഴിയുന്ന പാർട്ടീഷനുകൾ ഉള്ള ഒരു ഫോൾഡർ ഇപ്പോൾ നമുക്കുണ്ട്. അതെ? ഇല്ല.

ഞങ്ങൾ മേശ രജിസ്റ്റർ ചെയ്തത് പുഴയിൽ ആണെന്ന് ഓർക്കണം. ഫീൽഡ് നാമങ്ങളിൽ ഹൈവ് കേസ് സെൻസിറ്റീവ് അല്ല, അതേസമയം പാർക്ക്വെറ്റ് കേസ് സെൻസിറ്റീവ് ആണ്. അതിനാൽ, സ്കീമകളുള്ള പാർട്ടീഷനുകൾ: field1: int, Field1: int എന്നിവ ഹൈവിന് സമാനമാണ്, പക്ഷേ സ്പാർക്കിന് അല്ല. ഫീൽഡ് നാമങ്ങൾ ചെറിയക്ഷരത്തിലേക്ക് പരിവർത്തനം ചെയ്യാൻ മറക്കരുത്.

അതിനുശേഷം, എല്ലാം ശരിയാണെന്ന് തോന്നുന്നു.

എന്നിരുന്നാലും, എല്ലാം അത്ര ലളിതമല്ല. രണ്ടാമത്തേതും അറിയപ്പെടുന്നതുമായ ഒരു പ്രശ്നമുണ്ട്. ഓരോ പുതിയ പാർട്ടീഷനും വെവ്വേറെ സംരക്ഷിച്ചിരിക്കുന്നതിനാൽ, പാർട്ടീഷൻ ഫോൾഡറിൽ Spark സർവീസ് ഫയലുകൾ അടങ്ങിയിരിക്കും, ഉദാഹരണത്തിന്, _SUCCESS ഓപ്പറേഷൻ സക്സസ് ഫ്ലാഗ്. പാർക്ക്വെറ്റ് ചെയ്യാൻ ശ്രമിക്കുമ്പോൾ ഇത് ഒരു പിശകിന് കാരണമാകും. ഇത് ഒഴിവാക്കാൻ, ഫോൾഡറിലേക്ക് സേവന ഫയലുകൾ ചേർക്കുന്നതിൽ നിന്ന് സ്പാർക്കിനെ തടയുന്നതിന് നിങ്ങൾ കോൺഫിഗറേഷൻ കോൺഫിഗർ ചെയ്യേണ്ടതുണ്ട്:

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

ടാർഗെറ്റ് ഷോകേസ് ഫോൾഡറിലേക്ക് ഇപ്പോൾ എല്ലാ ദിവസവും ഒരു പുതിയ പാർക്ക്വെറ്റ് പാർട്ടീഷൻ ചേർക്കുന്നതായി തോന്നുന്നു, അവിടെ ആ ദിവസത്തെ പാഴ്‌സ് ചെയ്ത ഡാറ്റ സ്ഥിതിചെയ്യുന്നു. ഒരു ഡാറ്റാ ടൈപ്പ് വൈരുദ്ധ്യമുള്ള പാർട്ടീഷനുകൾ ഇല്ലെന്ന് ഞങ്ങൾ മുൻകൂട്ടി ശ്രദ്ധിച്ചു.

പക്ഷേ, ഞങ്ങൾക്ക് മൂന്നാമത്തെ പ്രശ്നമുണ്ട്. ഇപ്പോൾ പൊതുവായ സ്കീമ അറിയില്ല, കൂടാതെ, ഹൈവിലെ ടേബിളിന് തെറ്റായ ഒരു സ്കീമ ഉണ്ട്, കാരണം ഓരോ പുതിയ പാർട്ടീഷനും സ്കീമയിൽ ഒരു വികലത അവതരിപ്പിച്ചിരിക്കാം.

നിങ്ങൾ പട്ടിക വീണ്ടും രജിസ്റ്റർ ചെയ്യേണ്ടതുണ്ട്. ഇത് ലളിതമായി ചെയ്യാൻ കഴിയും: സ്റ്റോർഫ്രണ്ടിന്റെ പാർക്കറ്റ് വീണ്ടും വായിക്കുക, സ്കീമ എടുത്ത് അതിനെ അടിസ്ഥാനമാക്കി ഒരു DDL സൃഷ്ടിക്കുക, അതുപയോഗിച്ച് ഹൈവിൽ ഫോൾഡർ ഒരു ബാഹ്യ പട്ടികയായി വീണ്ടും രജിസ്റ്റർ ചെയ്യുക, ടാർഗെറ്റ് സ്റ്റോർഫ്രണ്ടിന്റെ സ്കീമ അപ്ഡേറ്റ് ചെയ്യുക.

ഞങ്ങൾക്ക് നാലാമത്തെ പ്രശ്നമുണ്ട്. ഞങ്ങൾ ആദ്യമായി ടേബിൾ രജിസ്റ്റർ ചെയ്യുമ്പോൾ, ഞങ്ങൾ സ്പാർക്കിനെ ആശ്രയിച്ചു. ഇപ്പോൾ ഞങ്ങൾ അത് സ്വയം ചെയ്യുന്നു, കൂടാതെ പാർക്ക്വെറ്റ് ഫീൽഡുകൾ പുഴയ്ക്ക് അനുവദനീയമല്ലാത്ത പ്രതീകങ്ങളിൽ നിന്ന് ആരംഭിക്കാമെന്ന് ഓർമ്മിക്കേണ്ടതുണ്ട്. ഉദാഹരണത്തിന്, "corrupt_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)

കോഡ് ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") സുരക്ഷിതമായ DDL ഉണ്ടാക്കുന്നു, അതായത് പകരം:

create table tname (_field1 string, 1field string)

"_ഫീൽഡ്1, 1ഫീൽഡ്" പോലുള്ള ഫീൽഡ് നാമങ്ങൾ ഉപയോഗിച്ച്, ഫീൽഡ് നാമങ്ങൾ രക്ഷപ്പെടുന്നിടത്ത് സുരക്ഷിതമായ DDL നിർമ്മിക്കപ്പെടുന്നു: പട്ടിക `tname` (`_field1` സ്ട്രിംഗ്, `1ഫീൽഡ്` സ്ട്രിംഗ്) സൃഷ്ടിക്കുക.

ചോദ്യം ഉയർന്നുവരുന്നു: ഒരു സമ്പൂർണ്ണ സ്കീമ (pf കോഡിൽ) ഉള്ള ഒരു ഡാറ്റാഫ്രെയിം എങ്ങനെ ശരിയായി ലഭിക്കും? ഈ പിഎഫ് എങ്ങനെ ലഭിക്കും? ഇതാണ് അഞ്ചാമത്തെ പ്രശ്നം. ടാർഗെറ്റ് ഷോകേസിന്റെ പാർക്കറ്റ് ഫയലുകൾ ഉള്ള ഫോൾഡറിൽ നിന്ന് എല്ലാ പാർട്ടീഷനുകളുടെയും സ്കീം വീണ്ടും വായിക്കണോ? ഈ രീതി ഏറ്റവും സുരക്ഷിതമാണ്, പക്ഷേ ബുദ്ധിമുട്ടാണ്.

സ്കീമ ഇതിനകം തന്നെ ഹൈവിലാണ്. മുഴുവൻ പട്ടികയുടെയും പുതിയ പാർട്ടീഷന്റെയും സ്കീമ സംയോജിപ്പിച്ച് നിങ്ങൾക്ക് ഒരു പുതിയ സ്കീമ ലഭിക്കും. അതിനാൽ നിങ്ങൾ ഹൈവിൽ നിന്ന് ടേബിൾ സ്കീമ എടുത്ത് പുതിയ പാർട്ടീഷന്റെ സ്കീമയുമായി സംയോജിപ്പിക്കേണ്ടതുണ്ട്. 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 സൃഷ്ടിക്കുന്നു.
മുഴുവൻ ശൃംഖലയും ശരിയായി പ്രവർത്തിക്കുന്നുണ്ടെങ്കിൽ, അതായത്, ഒരു ഇനീഷ്യലൈസിംഗ് ലോഡ് ഉണ്ടായിരുന്നു, കൂടാതെ ഹൈവിൽ ടേബിൾ ശരിയായി സൃഷ്ടിച്ചുവെങ്കിൽ, നമുക്ക് ഒരു അപ്ഡേറ്റ് ചെയ്ത ടേബിൾ സ്കീമ ലഭിക്കും.

അവസാനത്തെ പ്രശ്നം, നിങ്ങൾക്ക് ഒരു ഹൈവ് ടേബിളിലേക്ക് ഒരു പാർട്ടീഷൻ ചേർക്കാൻ കഴിയില്ല, കാരണം അത് തകർക്കപ്പെടും. ഹൈവിനെ അതിന്റെ പാർട്ടീഷൻ ഘടന ശരിയാക്കാൻ നിങ്ങൾ നിർബന്ധിക്കേണ്ടതുണ്ട്:

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

JSON വായിക്കുകയും അതിനെ അടിസ്ഥാനമാക്കി ഒരു സ്റ്റോർഫ്രണ്ട് സൃഷ്ടിക്കുകയും ചെയ്യുക എന്ന ലളിതമായ ദൗത്യം, നിങ്ങൾ പ്രത്യേകം നോക്കേണ്ട നിരവധി പരോക്ഷമായ ബുദ്ധിമുട്ടുകൾ തരണം ചെയ്യുന്നതിൽ കലാശിക്കുന്നു. ഈ പരിഹാരങ്ങൾ ലളിതമാണെങ്കിലും, അവ കണ്ടെത്തുന്നതിന് വളരെയധികം സമയമെടുക്കും.

ഷോകേസിന്റെ നിർമ്മാണം നടപ്പിലാക്കാൻ, എനിക്ക് ഇനിപ്പറയുന്നവ ചെയ്യേണ്ടിവന്നു:

• ഷോകേസിലേക്ക് പാർട്ടീഷനുകൾ ചേർക്കുക, സേവന ഫയലുകൾ ഒഴിവാക്കുക
• സ്പാർക്ക് ടൈപ്പ് ചെയ്ത ഉറവിട ഡാറ്റയിലെ ശൂന്യമായ ഫീൽഡുകൾ കൈകാര്യം ചെയ്യുക
• ലളിതമായ തരങ്ങൾ ഒരു സ്ട്രിംഗിലേക്ക് കാസ്‌റ്റ് ചെയ്യുക
• ഫീൽഡ് പേരുകൾ ചെറിയക്ഷരത്തിലേക്ക് പരിവർത്തനം ചെയ്യുക
• ഹൈവിൽ പ്രത്യേക ഡാറ്റ അപ്‌ലോഡും ടേബിൾ രജിസ്‌ട്രേഷനും (DDL ജനറേഷൻ)
• ഹൈവുമായി പൊരുത്തപ്പെടാത്ത ഫീൽഡ് നാമങ്ങളിൽ നിന്ന് രക്ഷപ്പെടാൻ മറക്കരുത്
• Hive-ൽ ടേബിൾ രജിസ്ട്രേഷൻ എങ്ങനെ അപ്ഡേറ്റ് ചെയ്യാമെന്ന് അറിയുക

ചുരുക്കത്തിൽ, ഷോപ്പ് വിൻഡോകൾ നിർമ്മിക്കാനുള്ള തീരുമാനം നിരവധി അപകടങ്ങൾ നിറഞ്ഞതാണെന്ന് ഞങ്ങൾ ശ്രദ്ധിക്കുന്നു. അതിനാൽ, നടപ്പിലാക്കുന്നതിൽ ബുദ്ധിമുട്ടുകൾ ഉണ്ടായാൽ, വിജയകരമായ വൈദഗ്ധ്യമുള്ള ഒരു പരിചയസമ്പന്നനായ പങ്കാളിയുമായി ബന്ധപ്പെടുന്നതാണ് നല്ലത്.

ഈ ലേഖനം വായിച്ചതിന് നന്ദി, വിവരങ്ങൾ നിങ്ങൾക്ക് ഉപയോഗപ്രദമാണെന്ന് ഞങ്ങൾ പ്രതീക്ഷിക്കുന്നു.

അവലംബം: www.habr.com