Spoštovani bralci, dober dan!
V tem članku vodilni svetovalec v poslovnem segmentu Big Data Solutions podjetja „Neofleks“ podrobno opisuje možnosti za gradnjo vitrin spremenljive strukture z uporabo Apache Spark.
V okviru projekta za analizo podatkov se pogosto pojavi naloga gradnje vitrin na osnovi slabo strukturiranih podatkov.
Običajno gre za dnevnike ali odgovore različnih sistemov, shranjenih v obliki JSON ali XML. Podatki se izvožijo v Hadoop, nato pa je treba zgraditi vitrino. Dostop do ustvarjene vitrine lahko organiziramo, na primer, preko Impale.
V tem primeru ciljna shema vitrine vnaprej ni znana. Poleg tega shemo predhodno ni mogoče sestaviti, ker je odvisna od podatkov, mi pa se ukvarjamo s temi slabo strukturiranimi podatki.
Na primer, danes je zabeležen takšen odgovor:
{source: "app1", error_code: ""}in jutri ta ista sistem pošlje tak odgovor:
{source: "app1", error_code: "error", description: "Omrežna napaka"}Zaradi tega bi se v vitrino moralo dodati še eno polje — description, in nihče ne ve, ali bo prišlo ali ne.
Naloga ustvarjanja vitrine na takšnih podatkih je precej standardna, in Spark za to ponuja vrsto orodij. Za razčlenitev izvornih podatkov obstaja podpora tako za JSON kot XML, za shemo, ki ni vnaprej znana, pa je predvidena podpora za schemaEvolution.
Na prvi pogled se rešitev zdi preprosta. Vzeti je treba mapo z JSON in prebrati v dataframe. Spark bo ustvaril shemo, zastavljene podatke pa spremenil v strukture. Nato je vse treba shraniti v parquet, ki ga podpira tudi Impala, registriramo vitrino v Hive metastore.
Zdi se, da je vse preprosto.
Vendar iz kratkih primerov v dokumentaciji ni jasno, kaj storiti s številnimi težavami v praksi.
V dokumentaciji je opisan pristop ne za ustvarjanje vitrine, temveč za branje JSON ali XML v dataframe.
In sicer, preprosto je prikazano, kako prebrati in razčleniti JSON:
df = spark.read.json(path...)To je dovolj, da so podatki dostopni za Spark.
V praksi je scenarij veliko zahtevnejši kot samo branje JSON datotek iz mape in ustvarjanje dataframe. Situacija je takšna: že obstaja določena vitrino, vsak dan prihajajo novi podatki, ki jih je treba dodati v vitrino, ne da bi pozabili, da se shema lahko razlikuje.
Običajna shema za gradnjo vitrine je taka:
Шаг 1. Podatki se nalagajo v Hadoop z dnevnim dodajanjem in so shranjeni v novo particijo. Tako nastane mapa, razdeljena po dnevih, z izvorno vsebino.
Шаг 2. Med začetnim nalaganjem se ta mapa prebere in obdela s pomočjo Sparka. Pridobljen dataframe se shranjuje v obliko, dostopno za analizo, kot je parquet, ki ga lahko nato uvozimo v Impalo. Tako se ustvari ciljni pregled z vsemi podatki, ki so se do tega trenutka nabrali.
Шаг 3. Ustvari se nalaganje, ki bo vsak dan posodabljalo pregled.
Pojavi se vprašanje inkrementalnega nalaganja, potreba po particioniranju pregleda in vprašanje podpore splošni shemi pregleda.
Poglejmo primer. Recimo, da je bil izveden prvi korak pri gradnji skladišča in da je bila nastavljena izvoz JSON datotek v mapo.
Ustvariti dataframe iz njih, da ga nato shranimo kot pregled, ni težko. To je tisti prvi korak, ki ga je mogoče enostavno najti v 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)Zdi se, da je vse v redu.
Prebrali in obdelali smo JSON, nato pa shranimo dataframe kot parquet, registriramo v Hive na kakršenkoli priročen način:
df.write.format(“parquet”).option('path','').saveAsTable('') Dobimo pregled.
Toda naslednji dan so se dodali novi podatki iz vira. Imamo mapo z JSON, in pregled, ustvarjen na podlagi te mape. Po nalaganju naslednje serije podatkov iz vira v pregledu primanjkuje podatkov za en dan.
Logična rešitev bi bila, da se pregled particionira po dnevih, kar omogoča dodajanje nove particije vsakič naslednji dan. Ta mehanizem je prav tako dobro znan, Spark omogoča zapisovanje particij ločeno.
Najprej izvedemo začetno nalaganje, pri čemer shranimo podatke, kot je bilo opisano zgoraj, dodamo le particioniranje. Ta postopek se imenuje inicializacija pregleda 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 težave.
Prva težava. Prej ali slej pridobljenega parquet ne bo mogoče prebrati. To je povezano z različno obravnavo praznih polj v parquet in JSON-u.
Poglejmo tipično situacijo. Včeraj je bil dobavljen JSON:
Dan 1: {"a": {"b": 1}},
danes pa ta isti JSON izgleda takole:
Dan 2: {"a": null}
Recimo, da imamo dva različna particioniranja, vsako s po enim zapisom.
Ko preberemo izvorne podatke v celoti, Spark lahko določi tip in razume, da je "a" polje tipa "struktur", z vgrajenim poljem "b" tipa INT. Ampak, če je bilo vsako particioniranje shranjeno ločeno, dobimo parquet z nezdružljivimi schemami particij:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Ta težava 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 sestavljal particije, ki jih bo mogoče prebrati skupaj.
Čeprav se bodo tisti, ki so to počeli v praksi, ob tem kislo nasmehnili. Zakaj? Zato, ker se verjetno bodo pojavile še dve situaciji. Ali tri. Ali štiri. Prva, ki se bo skoraj zagotovo pojavila, je, da bodo numerične vrste različne v različnih JSON datotekah. Na primer, {intField: 1} in {intField: 1.1}. Če se takšna polja znajdejo v istem particioniranju, bo združitev schem pravilna in jih privede do najbolj točnega tipa. Če pa so v različnih, bo v enem intField: int, v drugem pa intField: double.
Za obravnavo te situacije obstaja naslednji zastavica:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Zdaj imamo mapo, kjer so particije, ki jih je mogoče prebrati v enoten dataframe in veljaven parquet celotne vitrine. Da? Ne.
Treba je zapomniti, da smo registrirali tabelo v Hive. 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. Ne smemo pozabiti prevesti imen polj v male črke.
Po tem se zdi, da je vse v redu.
Vendar pa ni vse tako preprosto. Pojavi se druga, tudi dobro znana težava. Ker se vsaka nova particija shranjuje ločeno, bodo v mapi particije ležali pomožni datoteki Spark, na primer zastavica uspešnosti operacije _SUCCESS. To bo povzročilo napako pri poskusu parquet. Da bi se temu izognili, je treba konfiguracijo nastaviti tako, da Spark ne piše pomožnih datotek v mapo:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Кажется, теперь каждый день в папку целевой витрины добавляется новая parquet партиция, где лежат распарсенные данные за день. Мы заранее озаботились тем, чтобы не было партиций с конфликтом типов данных.
Но перед нами — третья проблема. Теперь общая схема не известна, более того, в Hive таблица с неправильной схемой, так как каждая новая партиция, скорее всего, внесла искажение в схему.
Нужно перерегистрировать таблицу. Это можно сделать просто: снова прочитать parquet витрины, взять схему и создать на ее основе DDL, с которым заново зарегистрировать папку в Hive как внешнюю таблицу, обновив схему целевой витрины.
Перед нами возникает четвертая проблема. Когда мы регистрировали таблицу первый раз, мы опирались на Spark. Теперь делаем это сами, и нужно помнить, что поля parquet могут начинаться с символов, недопустимых для Hive. Например, Spark выкидывает строки, которые не смог распарсить в поле «corrupt_record». Такое поле нельзя будет зарегистрировать в 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].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") делает безопасный DDL, то есть вместо:
create table tname (_field1 string, 1field string)
С такими названиями полей как «_field1, 1field», делается безопасный DDL, где названия полей экранированы: create table `tname` (`_field1` string, `1field` string).
Возникает вопрос: как правильно получить dataframe с полной схемой (в коде pf)? Как получить этот pf? Это пятая проблема. Перечитывать схему всех партиций из папки с parquet файлами целевой витрины? Это метод самый безопасный, но тяжелый.
Shema je že v Hive. Novo shemo lahko pridobimo tako, da združimo shemo celotne tabele in nove particije. To pomeni, da moramo 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: izvorno shemo tabele v Hive in novo particijo. Podatki so prav tako na voljo. Preostane le, da pridobimo novo shemo, v kateri se združi shema 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 odlomku.
Če celotna veriga deluje pravilno, torej je bila inicializacijska nalaganje, in je bila v Hive pravilno ustvarjena tabela, potem dobimo posodobljeno shemo tabele.
In končna težava je, da particije v tabelo Hive ni tako enostavno dodati, saj bo poškodovana. Hive je treba prepričati, da popravi svojo strukturo particij:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Preprosta naloga branja JSON in ustvarjanja vitrine na njeni osnovi se spremeni v premagovanje vrste prikritih težav, rešitve katerih je treba iskati posamično. Čeprav so te rešitve preproste, iskanje zanje vzame veliko časa.
Za izvedbo izgradnje vitrine je bilo potrebno:
- Dodati particije v vitrina, ob tem pa se znebiti pomožnih datotek
- Razjasniti prazna polja v izvornih podatkih, ki jih je Spark tipiziral
- Prilagoditi preproste tipe v nize
- Prilagoditi imena polj v male črke
- Ločiti izvoz podatkov in registracijo tabele v Hive (ustvarjanje DDL)
- Ne pozabiti, da je treba imena polj, ki so morda nezdružljiva s Hive, ustrezno zaščititi
- Naučiti se osvežiti registracijo tabele v Hive
Skupaj vzeto, rešitev za izgradnjo vitrin skriva številne pasti. Zato je ob pojavu težav pri izvajanju bolje, da se obrnete na izkušenega partnerja z uspešnim znanjem.
Hvala 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
