Poštovani čitatelji, dobar dan!
U ovom članku vodeći konzultant poslovnog sektora Big Data Solutions kompanije „Neoflex“ detaljno opisuje opcije za izgradnju vitrina promjenljive strukture koristeći Apache Spark.
U okviru projekta analize podataka često se javlja zadatak izgradnje vitrina na temelju slabo strukturiranih podataka.
Obično su to logovi ili odgovori različitih sustava, pohranjeni u obliku JSON ili XML. Podaci se prebacuju u Hadoop, a zatim treba iz njih izgraditi vitrinu. Organizirati pristup stvorenoj vitrini možemo, na primjer, preko Impale.
U tom slučaju, shema ciljne vitrine prethodno nije poznata. Štoviše, shemu nije moguće unaprijed izraditi jer ovisi o podacima, a mi se bavimo upravo tim slabo strukturiranim podacima.
Na primjer, danas se bilježi ovakav odgovor:
{source: "app1", error_code: ""}a sutra dolazi ovakav odgovor iz istog sustava:
{source: "app1", error_code: "error", description: "Greška u mreži"}Stoga, u vitrinu treba dodati još jedno polje — description, a hoće li to doći ili ne, nitko ne zna.
Zadatak izrade vitrine na takvim podacima prilično je standardan, i Spark za to ima niz alata. Za parsiranje izvornih podataka podržani su i JSON i XML, a za unaprijed nepoznatu shemu predviđena je podrška za schemaEvolution.
Na prvi pogled rješenje izgleda jednostavno. Samo treba uzeti mapu s JSON i pročitati u dataframe. Spark će stvoriti shemu, a ugniježđeni podaci pretvoriti u strukture. Zatim se sve treba sačuvati u parquet, koji je također podržan u Impali, registrirajući vitrinu u Hive metastore.
Izgleda da je sve jednostavno.
Međutim, iz kratkih primjera u dokumentaciji nije jasno što učiniti u slučaju niza problema u praksi.
U dokumentaciji se opisuje pristup koji nije za izradu vitrine, već za čitanje JSON ili XML u dataframe.
Točnije, samo se navodi kako pročitati i parsirati JSON:
df = spark.read.json(path...)To je dovoljno da se podaci učine dostupnima za Spark.
U praksi je scenarij daleko složeniji od jednostavnog čitanja JSON datoteka iz mape i stvaranja dataframea. Situacija izgleda ovako: već postoji određena vitrina, svakodnevno dolaze novi podaci, treba ih dodati u vitrinu, ne zaboravljajući da shema može biti različita.
Uobičajena shema izgradnje vitrine je sljedeća:
Korak 1. Podaci se učitavaju u Hadoop s dnevnim dozvolama, a pohranjuju se u novu particiju. Na taj način nastaje particionirana mapa po danima s izvornih podacima.
Korak 2. Tijekom inicijalnog učitavanja, ova mapa se čita i analizira pomoću Sparka. Dobiveni dataframe se sprema u format koji je dostupan za analizu, na primjer, u parquet, koji se kasnije može uvesti u Impalu. Tako se stvara ciljni izložak sa svim podacima koji su prikupljeni do tog trenutka.
Korak 3. Stvara se učitavanje koje će svaki dan ažurirati izložak.
Postavlja se pitanje inkrementalnog učitavanja, potreba za particioniranjem izložka i pitanje podrške općoj shemi izložka.
Uzet ćemo primjer. Pretpostavimo da je prvi korak izgradnje skladišta realiziran i da je postavljen izvoz JSON datoteka u mapu.
Stvoriti dataframe od njih kako bi se kasnije spremio kao izložak nije problem. To je onaj prvi korak koji se lako može naći u dokumentaciji Sparka:
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)Izgleda da je sve u redu.
Pročitali smo i analizirali JSON, a zatim spremamo dataframe kao parquet, registrirajući u Hive na bilo koji prikladan način:
df.write.format("parquet").option('path','').saveAsTable('') Dobivamo izložak.
No, sljedećeg dana su dodani novi podaci iz izvora. Imamo mapu s JSON-om i izložak stvoren na temelju te mape. Nakon učitavanja sljedeće serije podataka iz izvora, u izložku nedostaju podaci za jedan dan.
Logično rješenje bi bilo particionirati izložak po danima, što će omogućiti dodavanje nove particije svakog sljedećeg dana. Ovaj mehanizam je također dobro poznat, Spark omogućava odvojeno zapisivanje particija.
Prvo radimo inicijalno učitavanje, spremajući podatke kao što je opisano ranije, dodajući samo particioniranje. Ova akcija se naziva inicijalizacija izložka i provodi se samo jednom:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)
Sljedeći dan učitavamo samo novu particiju:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable + "/date_load=" + date_load + "/")
Ostaje samo ponovno registrirati u Hive kako bi se ažurirala shema.
Međutim, tu se javljaju problemi.
Prvi problem. Rano ili kasno, generirani parquet neće se moći pročitati. To je povezano s različitim pristupima koje parquet i JSON koriste za prazna polja.
Razmotrimo tipičnu situaciju. Na primjer, jučer dolazi JSON:
Dan 1: {"a": {"b": 1}},
a danas ovaj isti JSON izgleda ovako:
Dan 2: {"a": null}
Pretpostavimo da imamo dvije različite particije, od kojih svaka ima po jedan redak.
Kada čitamo izvorne podatke u cjelini, Spark će moći odrediti tip i shvatiti da je 'a' polje tipa 'struktura' s ugniježdenim poljem 'b' tipa INT. No, ako je svaka particija spremljena odvojeno, dobijamo parquet s nespojivim shemama particija:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Ova situacija je dobro poznata, stoga je posebno dodana opcija — prilikom parsiranja izvornih podataka ukloniti prazna polja:
df = spark.read.json("...", dropFieldIfAllNull=True)
U tom slučaju parquet će se sastojati od particija koje će se moći pročitati zajedno.
Iako će se oni koji su to radili u praksi tu gorko nasmiješiti. Zašto? Zato što će vjerojatno nastati još dvije situacije. Ili tri. Ili četiri. Prva koja će se gotovo sigurno dogoditi je da će numerički tipovi izgledati drugačije u različitim JSON datotekama. Na primjer, {intField: 1} i {intField: 1.1}. Ako takva polja dospiju u jednu particiju, spajanje shema će pročitati sve ispravno, dovodeći do najpreciznijeg tipa. No, ako budu u različitim, u jednoj će biti intField: int, a u drugoj intField: double.
Za obradu ove situacije postoji sljedeća zastavica:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Sada imamo mapu u kojoj se nalaze particije koje se mogu pročitati u jedinstveni dataframe i valjani parquet cijelog skladišta. Da? Ne.
Treba se sjetiti da smo registrirali tablicu u Hiveu. Hive nije osjetljiv na velika i mala slova u nazivima polja, dok parquet jest. Dakle, particije sa shemama: field1: int, i Field1: int za Hive su jednake, ali za Spark nisu. Ne smijemo zaboraviti prevesti nazive polja u mala slova.
Nakon toga, čini se da je sve u redu.
Međutim, nije sve tako jednostavno. Nastaje drugi, također dobro poznati problem. Budući da se svaka nova particija pohranjuje odvojeno, u mapi particije nalazit će se pomoćne datoteke Sparka, poput oznake uspješnosti operacije _SUCCESS. To će dovesti do pogreške pri pokušaju čitanja parqueta. Da bismo to izbjegli, potrebno je postaviti konfiguraciju koja će zabraniti Sparku da dopisuje pomoćne datoteke u mapu:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Čini se da svaki dan nova parquet particija dodaje podatke u ciljnu mapu, gdje se nalaze analizirani podaci za taj dan. Unaprijed smo se pobrinuli da ne bude particija sa sukobom tipova podataka.
Međutim, imamo treći problem. Sada je opća shema nepoznata, a čak je i u Hive-u tabela sa pogrešnom shemom, jer je svaka nova particija vjerojatno izvršila izobličenje u shemi.
Potrebno je ponovo registrirati tabelu. To se može jednostavno učiniti: ponovno pročitati parquet izložbu, uzeti shemu i stvoriti na njenoj osnovi DDL, kojom ćemo ponovo registrirati mapu u Hive-u kao vanjsku tabelu, ažurirajući shemu ciljne izložbe.
Javlja se četvrti problem. Kada smo prvi put registrirali tabelu, oslanjali smo se na Spark. Sada to radimo sami, i trebamo imati na umu da polja parquet mogu početi sa simbolima koji nisu dozvoljeni za Hive. Na primjer, Spark odbacuje redove koje nije mogao analizirati u polju "corrupt_record". Takvo polje neće moći biti registrirano u Hive-u bez da se ekranuje.
Znajući ovo, dobivamo shemu:
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)
Kod ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") izvodi sigurni DDL, odnosno umjesto:
create table tname (_field1 string, 1field string)
S takvim nazivima polja kao što su "_field1, 1field", izvodi se sigurni DDL, gdje su nazivi polja ekranujući: create table `tname` (`_field1` string, `1field` string).
Pitanje se postavlja: kako ispravno dobiti dataframe sa punom shemom (u kodu pf)? Kako dobiti ovaj pf? To je peti problem. Ponovno čitanje sheme svih particija iz mape sa parquet datotekama ciljne izložbe? Ova metoda je najsigurnija, ali teška.
Shema već postoji u Hiveu. Novi shemu možete dobiti kombiniranjem sheme cijele tablice i nove particije. To znači da trebate uzeti shemu tablice iz Hivea i spojiti je sa shemom nove particije. To se može učiniti čitanjem testnih metapodataka iz Hivea, spremanjem u privremenu mapu i čitanjem s pomoću Sparka obje particije odjednom.
U suštini, imamo sve što je potrebno: izvornu shemu tablice u Hiveu i novu particiju. Podaci su nam također dostupni. Trebamo samo dobiti novu shemu koja kombinira shemu izloga i nova polja iz stvorene 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/*")
Zatim kreiramo DDL za registraciju tablice, kao u prethodnom isječku.
Ako cijeli lanac radi ispravno, odnosno, ako je bila inicijalna učitavanja, a tablica u Hiveu je ispravno stvorena, tada dobivamo ažuriranu shemu tablice.
I posljednji problem je u tome što se ne može jednostavno dodati particija u tablicu Hive, jer će biti pokvarena. Potrebno je natjerati Hive da popravi svoju strukturu particija:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Jednostavan zadatak čitanja JSON-a i stvaranja na temelju njega izloga postaje prevladavanje niza nejasnih poteškoća, za čija se rješenja mora tražiti odvojeno. Iako su ta rješenja jednostavna, za njihovo pronalaženje potrebno je puno vremena.
Da bismo implementirali izradu izloga, morali smo:
- Dodavati particije u izlog, oslobađajući se pomoćnih datoteka
- Razumjeti prazna polja u izvornih podacima koje je Spark tipizirao
- Pretvoriti osnovne tipove u stringove
- Pretvoriti imena polja u mala slova
- Razdvojiti izvoz podataka i registraciju tablice u Hiveu (stvaranje DDL)
- Ne zaboraviti escape-ati imena polja koja mogu biti nespojiva s Hiveom
- Naučiti ažurirati registraciju tablice u Hiveu
Zaključno, napominjemo da rješenje za izradu izloga skriva mnogo podvodnih stjenovitih područja. Stoga, u slučaju poteškoća u implementaciji, bolje je obratiti se iskusnom partneru s uspješnim iskustvom.
Zahvaljujemo na čitanju ovog članka, nadamo se da će informacije biti korisne.
Izvor: habr.com
Kupite pouzdan hosting za web stranice s DDoS zaštitom, VPS VDS servere 🔥 Kupite pouzdan hosting za web stranice s DDoS zaštitom, VPS VDS servere | ProHoster
