Të nderuar lexues, përshëndetje!
Në këtë artikull, këshilltari kryesor i drejtimit të biznesit për Zgjidhjet Big Data në kompaninë «Neoflex» shpjegon hollësisht mundësitë e ndërtimit të vitrinave me strukturë të ndryshueshme duke përdorur Apache Spark.
Në kuadër të një projekti për analizën e të dhënave, shpesh lind nevoja për të ndërtuar vitrina mbi të dhëna me strukturë të dobët.
Zakonisht, këto janë loge, ose përgjigje nga sisteme të ndryshme, të ruajtura në formën e JSON ose XML. Të dhënat shkarkohen në Hadoop, dhe pastaj nga to duhet të ndërtohet një vitrinë. Organizimi i qasjes ndaj kësaj vetrinës mund të bëhet, për shembull, përmes Impala.
Në këtë rast, skema e vitrinës së synuar nuk dihet paraprakisht. Për më tepër, skema nuk mund të përgatitet paraprakisht, pasi varet nga të dhënat, dhe ne kemi të bëjmë me këto të dhëna me strukturë të dobët.
Për shembull, sot regjistrohet një përgjigje e tillë:
{source: "app1", error_code: ""}dhe nesër nga ky sistem vjen një përgjigje e tillë:
{source: "app1", error_code: "error", description: "Gabim nĂ« rrjet"}Si rezultat, nĂ« vitrinĂ« duhet tĂ« shtohet njĂ« fushĂ« tjetĂ«r â pĂ«rshkrimi, dhe nĂ«se do tĂ« vijĂ« apo jo, askush nuk e di.
Detyra e krijimit të një vitrimi me të dhëna të tilla është mjaft standarde, dhe për këtë, Spark ka një sërë mjetesh. Ekziston mbështetje për JSON dhe XML për analizimin e të dhënave burimore, dhe për një skemë që nuk dihet paraprakisht është parashikuar mbështetje për schemaEvolution.
Me një shikim të parë, zgjidhja duket e thjeshtë. Duhet të merret një dosje me JSON dhe të lexohet në dataframe. Spark do të krijojë skemën, duke e trasformuar të dhënat e thella në struktura. Më pas, gjithçka duhet të ruhet në parquet, që mbështetet gjithashtu në Impala, duke regjistruar vitrimin në Hive metastore.
Duket se gjithçka është e thjeshtë.
Megjithatë, nga shembujt e shkurtër në dokumentacion nuk është e qartë se çfarë duhet bërë me një varg problemesh në praktikë.
Në dokumentacion përshkruhet qasja jo për krijimin e vitrimit, por për leximin e JSON ose XML në dataframe.
Saktësisht, vetëm përmendet se si të lexosh dhe analizosh JSON:
df = spark.read.json(path...)Kjo mjafton për të bërë të dhënat të disponueshme për Spark.
Në praktikë, skenari është shumë më i komplikuar se sa thjesht të lexosh skedarët JSON nga një dosje dhe të krijosh një dataframe. Situata është kështu: tashmë ekziston një vitrimi i caktuar, çdo ditë vijnë të dhëna të reja, dhe ato duhet të shtohen në vitrimin, duke mos harruar se skema mund të ndryshojë.
Skema e zakonshme e ndërtimit të një vitrine është:
Hapi 1. Të dhënat ngarkohen në Hadoop me ngarkimin e përditshëm të mëpasshëm dhe grumbullohen në një parti të re. Kështu krijohet një folder i ndarë sipas ditëve me të dhënat burimore.
Hapi 2. Gjatë ngarkimit fillestar, ky folder lexohen dhe analizohen përmes Spark. Dataframe i marrë ruhet në formatin e duhur për analizë, për shembull, në parquet, i cili pastaj mund të importohet në Impala. Kështu krijohet vitrina e synuar me të gjitha të dhënat që janë mbledhur deri në këtë moment.
Hapi 3. Krijohet një ngarkesë që do të përditësojë vitrinën çdo ditë.
Shfaqet çështja e ngarkimit inkremental, nevoja për ndarjen e vitrinës, dhe çështja e mbështetjes së skemës së përgjithshme të vitrinës.
Le të japim një shembull. Supozoni se hapja e parë e ndërtimit të depozitës është realizuar dhe eksportimi i skedarëve JSON është konfiguruar në folder.
Krijimi i një dataframe nga këta skedarë për t'i ruajtur më pas si vitrinë, nuk paraqet problem. Ky është hapi i parë, i cili lehtë mund të gjendet në dokumentacionin e Spark:
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)Duket sikur gjithçka është mirë.
Ne lexuam dhe parseuam JSON, më pas ruajmë dataframe si parquet, duke e regjistruar në Hive në çdo mënyrë të përshtatshme:
df.write.format("parquet").option('path','').saveAsTable('') Marrim vitrinën.
Por, ditën tjetër u shtuan të dhëna të reja nga burimi. Kemi një dosje me JSON dhe vitrinë, e cila u krijua në bazë të kësaj dosje. Pas ngarkimit të grupeve të reja të të dhënave nga burimi, në vitrinë mungojnë të dhënat për një ditë.
Një zgjidhje logjike do të ishte të ndaheshin vitrinat sipas ditëve, që do të lejonte shtimin e një parti të re çdo ditë. Mekanizmi i këtij procesi gjithashtu njihet mirë, Spark lejon të shkruhen partitë veçmas.
Së pari bëjmë një ngarkim fillestar, duke ruajtur të dhënat siç u përshkrua më lart, duke shtuar vetëm ndarjen. Ky veprim quhet inkuadrimi i vitrinës dhe bëhet vetëm një herë:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)
Ditën tjetër ngarkojmë vetëm partinë e re:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable +"/date_load=" + date_load + "/")
Nuk mbetet veçse të regjistrohet përsëri në Hive, për të përditësuar skemën.
Megjithatë, këtu lindin problemet.
Problemi i parë. Herë pas here, parquet që del nuk do të jetë i lexueshëm. Kjo është për shkak të mënyrave të ndryshme që parquet dhe JSON i qasen fushave të bosh.
Le të shqyrtojmë një situatë tipike. Për shembull, dje merrni JSON-in:
Dita 1: {"a": {"b": 1}},
ndërsa sot ky JSON duket kështu:
Dita 2: {"a": null}
Supozoni se kemi dy particione të ndryshme, ku secila ka një rresht.
Kur lexojmë të dhënat fillestare totalisht, Spark do të jetë në gjendje të përcaktojë tipin dhe do të kuptojë se "a" është një fushë e tipit "strukturë", me një fushë të brendshme "b" që është tip INT. Por, nëse çdo particion ruhet veçmas, atëherë kemi parquet me skema particionesh të papajtueshme:
df1 (a: <struct>) df2 (a: STRING NULLABLE)
Kjo situatë është e njohur mirë, prandaj është shtuar veçanërisht një opsion - gjatë parse-it të të dhënave fillestare për të hequr fushat e bosh:
df = spark.read.json("...", dropFieldIfAllNull=True)
NĂ« kĂ«tĂ« rast, parquet do tĂ« pĂ«rbĂ«het nga particione qĂ« do tĂ« jetĂ« e mundur tâi lexoni sĂ« bashku.
Megjithëse ata që e kanë bërë këtë në praktikë do të qeshin me dhimbje. Pse? Sepse, përkundrazi, do të shfaqen dy situata të tjera. Ose tre. Ose katër. E para që do të ndodhi me shumë gjasa, tipet numerike do të duken ndryshe në skedarët e ndryshëm JSON. Për shembull, {intField: 1} dhe {intField: 1.1}. Nëse këto fusha ndodhen në të njëjtën paketë, atëherë mërgimi i skemave do t'i lexojë gjithçka në mënyrë korrekte, duke i çuar në tipin më të saktë. Por nëse janë në të ndryshme, atëherë në njërën do të jetë intField: int, ndërsa në tjetrën intField: double.
Për të trajtuar këtë situatë, ka flamurin e mëposhtëm:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Tani kemi një dosje ku ndodhen paketat që mund të lexohen në një dataframe të vetme dhe parquet të vlefshëm për të gjithë vitrinën. Po? Jo.
Duhet të kujtojmë se ne e regjistruam tabelën në Hive. Hive nuk është i ndjeshëm ndaj rastit në emrat e fushave, ndërsa parquet është i ndjeshëm. Prandaj, paketat me skema: field1: int, dhe Field1: int janë të njëjta për Hive, por jo për Spark. Duhet të mos harrojmë t'i sjellim emrat e fushave në shkronja të vogla.
Pas kësaj, duket se gjithçka është mirë.
Megjithatë, nuk është kaq e thjeshtë. Lind një problem i dytë, gjithashtu i njohur mirë. Dukë qenë se çdo particion i ri ruhet në veçanti, në dosjen e particionit do të jenë skedarë shërbimi të Spark, si p.sh. flamuri i suksesit të operacionit _SUCCESS. Kjo do të shkaktojë një gabim kur të përpiqesh të përdorësh parquet. Për ta parandaluar këtë, duhet të konfigurosh konfigurimin duke ndaluar Spark që të shkruajë skedarë shërbimi në dosje:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Duket se tani çdo ditë në dosjen e vitrinës së synuar po shtohet një particion i ri parquet, ku ndodhen të dhënat e parse. Ne e kemi marrë parasysh që të mos ketë particione me konflikte të llojeve të të dhënave.
Por, para nesh ndodhet një problem i tretë. Tani skema e përgjithshme nuk është e njohur, madje, në tabelën Hive skema është e gabuar, pasi çdokë particion i ri ka sjellë ndoshta një zhvendosje në skemë.
Duhet të regjistrosh përsëri tabelën. Kjo mund të bëhet lehtë: lexoni përsëri vitrinën parquet, merrni skemën dhe krijoni mbi të DDL, me të cilën regjistrohet përsëri dosja në Hive si një tabelë e jashtme, duke përditësuar skemën e vitrinës së synuar.
Para ne del të katërtin problem. Kur regjistruam tabelën për herë të parë, u mbështetëm në Spark. Tani po e bëjmë këtë vetë, dhe duhet të kujtojmë se fushat parquet mund të fillojnë me simbole që nuk janë të pranuara nga Hive. Për shembull, Spark hedh jashtë rreshtat që nuk arriti t'i shkruajë në fushën "corrupt_record". Një fushë e tillë nuk do të regjistrohet në Hive pa e bërë ato të qarta.
Duke e ditur këtë, marrim skemën:
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)
Kodi ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "`<").replace(",", ",`").replace("array<`", "array<") bën DDL të sigurt, pra në vend të:
create table tname (_field1 string, 1field string)
Me emrat e tillë të fushave si "_field1, 1field", bëhet DDL i sigurt, ku emrat e fushave janë të qarta: create table `tname` (`_field1` string, `1field` string).
Ka ndodh një pyetje: si të marrim saktësisht një dataframe me skemën e plotë (në kodin pf)? Si ta marrim këtë pf? Ky është problemi i pestë. A duhet të rilexojmë skemën e të gjitha partiçioneve nga folderi me skedarët parquet të vitrinës së synuar? Ky është metoda më e sigurt, por e rëndë.
Skema tashmë ekziston në Hive. Një skemë të re mund ta marrim duke bashkuar skemën e gjithë tabelës me partiçionin e ri. Pra, na nevojitet të marrim skemën e tabelës nga Hive dhe ta bashkojmë atë me skemën e partiçionit të ri. Kjo mund të bëhet duke lexuar metadatën testuese nga Hive, duke e ruajtur atë në një dosje përkohshme dhe duke lexuar me Spark të dy partiçionet në të njëjtën kohë.
Në thelb, kemi gjithçka që na nevojitet: skemën origjinale të tabelës në Hive dhe partiçionin e ri. Të dhënat i kemi gjithashtu. Na mbetet vetëm të marrim një skemë të re, ku bashkohet skema e vitrinës dhe fushat e reja nga partiçioni i krijuar:
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/*")
Më pas krijojmë DDL-në për regjistrimin e tabelës, si në fragmentin e mëparshëm.
NĂ«se e gjithĂ« zinxhiri funksionon siç duhet, pĂ«rkatĂ«sisht â ka pasur njĂ« ngarkesĂ« fillestare, dhe tabela Ă«shtĂ« krijuar siç duhet nĂ« Hive, atĂ«herĂ« ne marrim njĂ« skemĂ« tĂ« pĂ«rditĂ«suar tĂ« tabelĂ«s.
Dhe problemi i fundit është se nuk mund ta shtosh thjesht një parti në tabelën Hive, pasi ajo do të prishet. Duhet të bësh që Hive ta riparojë strukturën e partive të saj:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Një detyrë e thjeshtë e leximit të JSON dhe krijimit të një ndihme mbi të buron në përballjen me një sërë vështirësish të paprekura, zgjidhjet për të cilat duhet të kërkohen veç e veç. Dhe ndonëse këto zgjidhje janë të thjeshta, por kërkesa për to merr shumë kohë.
Për të realizuar ndërtimin e ndihmës, duhej:
- Të shtoja parti në ndihmë, duke u liruar nga skedarët shërbues
- Të kuptoja fushat e zbrazëta në të dhënat burimore, të cilat Spark i kishte tipizuar
- Të përmirësoja tipet e thjeshta në varg
- Të sillja emrat e fushave në shkronja të vogla
- Të ndaja ngarkimin e të dhënave dhe regjistrimin e tabelës në Hive (krijimi i DDL)
- Të mos harroja të ekranoja emrat e fushave që mund të ishin të papërshtatshme me Hive
- Të mësoja si të përditësoja regjistrimin e tabelës në Hive
Në përfundim, është e rëndësishme të theksohet se ndërtimi i vitrinave sjell shumë sfida. Prandaj, kur hasni vështirësi në realizim, është më mirë të konsultoheni me një partner të ekspertizës së suksesshme.
Faleminderit për leximin e këtij artikulli, shpresojmë që informacioni të jetë i dobishëm.
Burimi: habr.com
Bli njĂ« hosting tĂ« besueshĂ«m pĂ«r faqet me mbrojtje DDoS, VPS VDS serverĂ« đ„ Bli njĂ« hosting tĂ« besueshĂ«m pĂ«r faqet me mbrojtje DDoS, VPS VDS serverĂ« | ProHoster
