વ્યવહારમાં સ્પાર્ક સ્કીમા ઇવોલ્યુશન

પ્રિય વાચકો, શુભ દિવસ!

આ લેખમાં, Neoflex ના બિગ ડેટા સોલ્યુશન્સ બિઝનેસ એરિયાના અગ્રણી સલાહકાર અપાચે સ્પાર્કનો ઉપયોગ કરીને વેરિયેબલ સ્ટ્રક્ચર શોકેસ બનાવવા માટેના વિકલ્પોનું વિગતવાર વર્ણન કરે છે.

ડેટા એનાલિસિસ પ્રોજેક્ટના ભાગ રૂપે, ઢીલી રીતે સ્ટ્રક્ચર્ડ ડેટા પર આધારિત સ્ટોરફ્રન્ટ્સ બનાવવાનું કાર્ય ઘણીવાર ઉદ્ભવે છે.

સામાન્ય રીતે આ લોગ્સ અથવા વિવિધ સિસ્ટમોના પ્રતિસાદો હોય છે, જે JSON અથવા XML તરીકે સાચવવામાં આવે છે. ડેટા Hadoop પર અપલોડ કરવામાં આવે છે, પછી તમારે તેમની પાસેથી સ્ટોરફ્રન્ટ બનાવવાની જરૂર છે. અમે બનાવેલ શોકેસની ઍક્સેસ ગોઠવી શકીએ છીએ, ઉદાહરણ તરીકે, Impala દ્વારા.

આ કિસ્સામાં, લક્ષ્ય સ્ટોરફ્રન્ટની સ્કીમા અગાઉથી જાણીતી નથી. વધુમાં, સ્કીમ પણ અગાઉથી તૈયાર કરી શકાતી નથી, કારણ કે તે ડેટા પર આધારિત છે, અને અમે આ ખૂબ જ ઢીલા માળખાગત ડેટા સાથે કામ કરી રહ્યા છીએ.

ઉદાહરણ તરીકે, આજે નીચેનો પ્રતિસાદ લૉગ થયેલ છે:

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

અને આવતીકાલે એ જ સિસ્ટમમાંથી નીચેનો જવાબ આવશે:

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

પરિણામે, શોકેસમાં વધુ એક ક્ષેત્ર ઉમેરવું જોઈએ - વર્ણન, અને તે આવશે કે નહીં તે કોઈ જાણતું નથી.

આવા ડેટા પર સ્ટોરફ્રન્ટ બનાવવાનું કાર્ય ખૂબ પ્રમાણભૂત છે, અને સ્પાર્ક પાસે આ માટે ઘણા બધા સાધનો છે. સ્ત્રોત ડેટાને પાર્સ કરવા માટે, JSON અને XML બંને માટે સપોર્ટ છે, અને અગાઉની અજાણી સ્કીમા માટે, સ્કીમા ઈવોલ્યુશન માટે સપોર્ટ આપવામાં આવે છે.

પ્રથમ નજરમાં, ઉકેલ સરળ લાગે છે. તમારે 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 વાંચી અને પાર્સ કર્યું, પછી અમે ડેટાફ્રેમને લાકડાં તરીકે સાચવીએ છીએ, તેને કોઈપણ અનુકૂળ રીતે 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)

હવે અમારી પાસે એક ફોલ્ડર છે જ્યાં ત્યાં પાર્ટીશનો છે જે એક જ ડેટાફ્રેમમાં વાંચી શકાય છે અને સમગ્ર શોકેસની માન્ય લાકડાનું પાતળું પડ છે. હા? ના.

આપણે યાદ રાખવું જોઈએ કે અમે મધપૂડોમાં ટેબલ રજીસ્ટર કર્યું છે. ફીલ્ડના નામોમાં મધપૂડો કેસ સેન્સિટિવ નથી, જ્યારે લાકડી કેસ સેન્સિટિવ છે. તેથી, સ્કીમા સાથેના પાર્ટીશનો: 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].બદલો(":", "`:").બદલો("<", "<`").બદલો(",", ",`").બદલો("એરે<`", "એરે<") સુરક્ષિત DDL બનાવે છે, એટલે કે તેના બદલે:

create table tname (_field1 string, 1field string)

"_field1, 1field" જેવા ફીલ્ડ નામો સાથે, જ્યાં ફીલ્ડના નામો છટકી ગયા હોય ત્યાં સુરક્ષિત DDL બનાવવામાં આવે છે: ટેબલ `tname` (`_field1` સ્ટ્રિંગ, `1field` સ્ટ્રિંગ) બનાવો.

પ્રશ્ન ઊભો થાય છે: સંપૂર્ણ સ્કીમા (pf કોડમાં) સાથે ડેટાફ્રેમ કેવી રીતે યોગ્ય રીતે મેળવવી? આ પીએફ કેવી રીતે મેળવશો? આ પાંચમી સમસ્યા છે. ફોલ્ડરમાંથી તમામ પાર્ટીશનોની સ્કીમને ટાર્ગેટ શોકેસની લાકડાની ફાઈલો સાથે ફરીથી વાંચો? આ પદ્ધતિ સૌથી સલામત છે, પરંતુ મુશ્કેલ છે.

સ્કીમા પહેલેથી જ 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 બનાવીએ છીએ.
જો આખી સાંકળ યોગ્ય રીતે કામ કરે છે, એટલે કે, ત્યાં પ્રારંભિક લોડ હતો, અને કોષ્ટક મધપૂડોમાં યોગ્ય રીતે બનાવવામાં આવ્યું હતું, તો પછી અમને અપડેટ કરેલ ટેબલ સ્કીમા મળે છે.

અને છેલ્લી સમસ્યા એ છે કે તમે મધપૂડો કોષ્ટકમાં ફક્ત પાર્ટીશન ઉમેરી શકતા નથી, કારણ કે તે તૂટી જશે. તમારે તેના પાર્ટીશન સ્ટ્રક્ચરને ઠીક કરવા માટે મધપૂડોને દબાણ કરવાની જરૂર છે:

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

JSON વાંચવાનું અને તેના આધારે સ્ટોરફ્રન્ટ બનાવવાનું સરળ કાર્ય અસંખ્ય ગર્ભિત મુશ્કેલીઓને દૂર કરવામાં પરિણમે છે, ઉકેલો જેના માટે તમારે અલગથી શોધવું પડશે. અને આ ઉકેલો સરળ હોવા છતાં, તેમને શોધવામાં ઘણો સમય લાગે છે.

શોકેસના બાંધકામને અમલમાં મૂકવા માટે, મારે આ કરવું પડ્યું:

• શોકેસમાં પાર્ટીશનો ઉમેરો, સેવા ફાઇલોથી છુટકારો મેળવો
• સ્પાર્કે ટાઈપ કરેલ સ્ત્રોત ડેટામાં ખાલી ફીલ્ડ સાથે ડીલ કરો
• સ્ટ્રિંગ પર સરળ પ્રકારો કાસ્ટ કરો
• ફીલ્ડના નામોને લોઅરકેસમાં કન્વર્ટ કરો
• મધપૂડો (DDL જનરેશન) માં અલગ ડેટા અપલોડ અને ટેબલ નોંધણી
• Hive સાથે અસંગત હોઈ શકે તેવા ફીલ્ડ નામોથી બચવાનું ભૂલશો નહીં
• Hive માં કોષ્ટક નોંધણી કેવી રીતે અપડેટ કરવી તે જાણો

સારાંશમાં, અમે નોંધીએ છીએ કે દુકાનની બારીઓ બનાવવાનો નિર્ણય ઘણી મુશ્કેલીઓથી ભરપૂર છે. તેથી, અમલીકરણમાં મુશ્કેલીઓના કિસ્સામાં, સફળ કુશળતા ધરાવતા અનુભવી ભાગીદારનો સંપર્ક કરવો વધુ સારું છે.

આ લેખ વાંચવા બદલ આભાર, અમને આશા છે કે તમને માહિતી ઉપયોગી લાગશે.

સોર્સ: www.habr.com