Spoštovani bralci, dober dan!
V tem članku vodilni svetovalec poslovne enote Big Data Solutions podjetja «Neoflex» podrobno opisuje možnosti gradnje vitrin spremenljive strukture z uporabo Apache Sparka.
V okviru projekta za analizo podatkov se pogosto pojavi naloga gradnje vitrin na podlagi slabo strukturiranih podatkov.
Običajno gre za dnevnike ali odgovore različnih sistemov, shranjenih v obliki JSON ali XML. Podatki se izvažajo v Hadoop, nato pa je treba z njihovo pomočjo zgraditi vitrino. Dostop do ustvarjene vitrine lahko organiziramo na primer preko Impale.
V tem primeru je shema ciljne vitrine vnaprej neznana. Poleg tega shemo ni mogoče pripraviti vnaprej, saj je odvisna od podatkov, s katerimi imamo opravka in ti so slabo strukturirani.
Na primer, danes je zabeležen takšen odgovor:
{source: "app1", error_code: ""}jutri pa iz istega sistema prispe tak odgovor:
{source: "app1", error_code: "error", description: "Napaka omrežja"}Zaradi tega bi moralo v vitrino priti še eno polje — description, in nihče ne ve, ali pride ali ne.
Naloga ustvarjanja vitrine na takih podatkih je kar standardna, za Spark pa obstaja vrsta orodij. Za analizo izvornih podatkov je podpora tako za JSON kot za XML, za neznano shemo pa je predvidena podpora schemaEvolution.
Na prvi pogled rešitev izgleda preprosto. Treba je vzeti mapo z JSON in jo prebrati v dataframe. Spark bo ustvaril shemo, vdelane podatke pa pretvoril v strukture. Nato je vse treba shraniti v parquet, ki ga podpira tudi Impala, registriramo vitrino v Hive metastore.
Na videz vse enostavno.
Vendar pa iz kratkih primerov v dokumentaciji ni jasno, kako ravnati z vrsto težav v praksi.
V dokumentaciji je opisan pristop ne za ustvarjanje vitrine, ampak za branje JSON ali XML v dataframe.
Namreč, preprosto je prikazano, kako prebrati in razčleniti JSON:
df = spark.read.json(path...)To je dovolj, da podatki postanejo dostopni za Spark.
V praksi pa je scenarij precej bolj zapleten kot le branje JSON datotek iz mape in ustvarjanje dataframe. Situacija je takšna: že obstaja določena vitrina, vsak dan prihajajo novi podatki, te je treba dodati v vitrino, pri čemer ne smemo pozabiti, da se shema lahko razlikuje.
Običajna shema gradnje vitrine je taka:
Шаг 1. Podatki se nalagajo v Hadoop s kasnejšim dnevnim nalaganjem in se shranijo v novo particijo. Tako nastane mapo, ki je particionirana po dnevih z izvorni podatki.
Шаг 2. Med inicialnim nalaganjem se ta mapa prebere in razčleni s sredstvi Spark. Pridobljen dataframe se shrani v obliko, dostopno za analizo, na primer v parquet, ki ga lahko kasneje uvozimo v Impalo. Tako se ustvari ciljna vitrina z vsemi podatki, ki so se nabrali do tega trenutka.
Шаг 3. Ustvari se nalaganje, ki bo vsak dan osveževalo vitrino.
Pojavi se vprašanje inkrementalnega nalaganja, potreba po particioniranju vitrino in vprašanje podpore skupni shemi vitrine.
Poglejmo primer. Recimo, da je bil izveden prvi korak pri postavljanju skladišča in nastavljena izvoz JSON datotek v mapo.
Ustvariti dataframe iz njih, da ga kasneje shranimo kot vitrino, ni težko. To je tisti prvi korak, ki ga lahko enostavno najdemo v dokumentaciji 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)Zdi se, da je vse dobro.
Prebrali smo in razčlenili JSON, nato shranimo dataframe kot parquet, pri čemer registriramo v Hive na kateri koli priročen način:
df.write.format(“parquet”).option('path','').saveAsTable('') Dobimo vitrino.
Toda drugi dan so se dodali novi podatki iz vira. Imamo mapo z JSON in vitrino, ustvarjeno na podlagi te mape. Po nalaganju naslednjega niza podatkov iz vira v vitrini primanjkuje podatkov za en dan.
Logična rešitev bi bila particionirati vitrino po dneh, kar bo omogočilo dodajanje nove particije vsak naslednji dan. Ta mehanizem je prav tako dobro znan, Spark omogoča zapisovanje particij ločeno.
Najprej izvedemo inicialno nalaganje, pri čemer shranimo podatke, kot je bilo opisano zgoraj, in dodamo le še particioniranje. Ta postopek se imenuje inicializacija vitrine in se izvede le enkrat:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)
Naslednji dan nalagamo le novo particijo:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable +"/date_load=" + date_load + "/")
Ostane le še, da ponovno registriramo v Hive, da posodobimo shemo.
Vendar se tukaj pojavijo problemi.
Prva težava. Prej ali slej bo ustvarjeni parquet neberljiv. To je povezano s tem, kako se parquet in JSON različно lotevata praznih polj.
Oglejmo si tipično situacijo. Na primer, včeraj prejmemo JSON:
Dan 1: {"a": {"b": 1}},
danes pa je isti JSON videti tako:
Dan 2: {"a": null}
Recimo, da imamo dva različna particije, pri katerih je vsaka data enako vrsto.
Ko preberemo izvorne podatke v celoti, bo Spark sposoben določiti tip in razumeti, da je »a« polje tipa »struktura«, z vgrajenim poljem »b« tipa INT. Vendar, če je bila vsaka particija shranjena posebej, bo nastal parquet z nezdružljivimi shemami particijami:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Ta situacija je dobro znana, zato je bila posebej dodana možnost — pri obdelavi izvornih podatkov odstraniti prazna polja:
df = spark.read.json("...", dropFieldIfAllNull=True)
V tem primeru bo parquet sestavljen iz particij, ki jih bo mogoče prebrati skupaj.
Čeprav se bodo tisti, ki so to počeli v praksi, tu grenko smejali. Zakaj? No, ker bo najverjetneje prišlo do še dveh situacij. Ali treh. Ali štirih. Prva, ki se bo skoraj zagotovo pojavila, je, da bodo numerični tipi različni v različnih JSON datotekah. Na primer, {intField: 1} in {intField: 1.1}. Če naletimo na taka polja v eni particiji, bo združitev shem vse pravilno prebrala in pripeljala do najnatančnejšega tipa. Če pa so v različnih, bo v eni intField: int, v drugi pa intField: double.
Za obravnavo te situacije obstaja naslednja zastavica:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Zdaj imamo mapo, v kateri so particije, ki jih je mogoče prebrati v enoten dataframe in veljaven parquet celotne vitrine. Ja? Ne.
Treba se je spomniti, da smo registrirali tabelo v Hivu. Hive ni občutljiv na velike in male črke v imenih polj, parquet pa je. Zato so particije s shemami: field1: int, in Field1: int za Hive enake, za Spark pa ne. Treba je upoštevati, da se imena polj spremenijo v male črke.
Po tem se zdi, da je vse v redu.
Vendar pa ni vse tako preprosto. Pojavi se druga, prav tako dobro znana težava. Ker se vsaka nova particija shrani ločeno, bodo v mapi particije prisotni sistemski datoteki Spark, na primer, zastavica uspešnosti operacije _SUCCESS. To bo povzročilo napako pri poskusu parqueta. Da bi se temu izognili, je treba nastaviti konfiguracijo, ki prepoveduje Sparku zapisovanje sistemskih datotek v mapo:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Zdi se, da vsak dan v mapo ciljne izložbe dodajo novo parquet particijo, kjer so shranjeni razčlenjeni podatki za dan. Vnaprej smo poskrbeli, da ne bo prišlo do konfliktov tipov podatkov.
Toda pred nami je tretja težava. Zdaj skupna shema ni znana, bolj kot to, v Hive je tabela z napačno shemo, ker vsaka nova particija verjetno dodaja popravke v shemo.
Treba je ponovno registrirati tabelo. To lahko storimo preprosto: ponovno preberemo parquet izložbe, vzamemo shemo in na njeni podlagi ustvarimo DDL, s katerim ponovno registriramo mapo v Hive kot zunanjo tabelo, ter posodobimo shemo ciljne izložbe.
Pred nami se pojavi četrta težava. Ko smo prvič registrirali tabelo, smo se zanašali na Spark. Zdaj to počnemo sami in moramo upoštevati, da lahko polja parquet začnejo s simboli, ki niso sprejemljivi za Hive. Na primer, Spark zavrne vrstice, ki jih ni mogel razčleniti v polje »corrupt_record«. Takšnega polja ne bo mogoče registrirati v Hive brez, da ga escapiramo.
Vemo to, dobimo shemo:
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)
Koda ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") naredi varno DDL, torej namesto:
create table tname (_field1 string, 1field string)
S takšnimi imeni polj kot so „_field1, 1field“, se naredi varno DDL, kjer so imena polj escapirana: create table `tname` (`_field1` string, `1field` string).
Pojavi se vprašanje: kako pravilno pridobiti dataframe s polno shemo (v kodi pf)? Kako pridobiti ta pf? To je peta težava. Prebrati shemo vseh particij iz mape s parquet datotekami ciljne izložbe? To je najsigurnejša metoda, vendar težka.
Na voljo je shema v Hive. Novo shemo lahko pridobimo s povezovanjem sheme celotne tabele in nove particije. Torej je treba vzeti shemo tabele iz Hive in jo združiti z shemo nove particije. To lahko storimo tako, da preberemo testne metapodatke iz Hive, jih shranimo v začasno mapo in hkrati preberemo obe particiji s pomočjo Sparka.
V bistvu imamo vse, kar potrebujemo: izvirno shemo tabele v Hive in novo particijo. Imamo tudi podatke. Preprosto je le pridobiti novo shemo, ki združuje shemo vitrine in nova polja iz ustvarjene particije:
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/*")
Nato ustvarimo DDL za registracijo tabele, kot v prejšnjem fragmentu.
Če celotna veriga deluje pravilno, to pomeni, da je bila izvedena začetna nalaganje in da je tabela v Hive pravilno ustvarjena, potem dobimo posodobljeno shemo tabele.
In zadnja težava je, da ni tako enostavno dodati particijo v tabelo Hive, ker jo lahko poškoduje. Potrebno je prisiliti Hive, da popravi svojo strukturo particij:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Enostavna naloga branja JSON in ustvarjanja vitrine na njeni osnovi se spremeni v premagovanje številnih skritih težav, rešitve za katere je treba iskati posamično. Čeprav so te rešitve preproste, njihov iskanje vzame veliko časa.
Za uresničitev gradnje vitrine smo morali:
- Dodati particije v vitrina, znebiti se pomožnih datotek
- Izraziti prazna polja v izvornih podatkih, ki jih je Spark tipiziral
- Spremeniti preproste tipe v niz
- Prevesti imena polj v majhno črko
- Ločiti nalaganje podatkov in registracijo tabele v Hive (ustvarjanje DDL)
- Ne pozabiti pobegniti imen polj, ki so lahko nezdružljiva s Hive
- Naučiti se posodabljati registracijo tabele v Hive
Sklepamo, da rešitev za gradnjo vitrin skriva številne pasti. Zato je ob težavah pri implementaciji najbolje poiskati izkušenega partnerja z uspešno strokovnostjo.
Zahvaljujemo se vam za branje tega članka, upamo, da bodo informacije koristne.
Izvor: habr.com
Kupite zanesljivo gostovanje za spletne strani z zaščito pred DDoS, VPS VDS strežniki 🔥 Kupite zanesljivo gostovanje za spletne strani z zaščito pred DDoS, VPS VDS strežniki | ProHoster
