Tisztelt olvasók, kellemes napot kívánok!
Ebben a cikkben a "Neoflex" cég Big Data Solutions üzletágának vezető tanácsadója részletesen bemutatja az Apache Spark használatával történő, változó struktúrájú kirakatok kialakításának lehetőségeit.
Adat-analízis projekteknél gyakran felmerül a feladat, hogy gyenge struktúrával rendelkező adatok alapján kirakatokat építsünk.
Ezek rendszerint logok vagy különböző rendszerek válaszai, amelyeket JSON vagy XML formátumban tárolnak. Az adatokat betöltjük a Hadoop-ba, majd a kirakatot ebből kell létrehozni. A létrehozott kirakathoz például Impalán keresztül férhetünk hozzá.
Ebben az esetben a célzott kirakat sémája előre nem ismert. Sőt, a séma nem is állítható össze előre, mivel az az adatoktól függ, és mi ugyebár épp ezekkel a gyenge struktúrával rendelkező adatokkal foglalkozunk.
Például, ma egy ilyen válasz kerül naplózásra:
{source: "app1", error_code: ""}és holnap ugyanattól a rendszertől egy ilyen válasz érkezik:
{source: "app1", error_code: "error", description: "Hálózati hiba"}Ennek következtében a kirakathoz hozzá kell adni egy új mezőt — a description-t, és hogy ez megérkezik-e, senki sem tudja.
Az ilyen adatok alapján történő kirakat létrehozásának feladata meglehetősen általános, és a Spark számára számos eszköz áll rendelkezésre. Az eredeti adatok elemzésére támogatja a JSON-t és az XML-t is, míg az előre nem ismert séma számára a schemaEvolution támogatás is biztosított.
Első pillantásra a megoldás egyszerűnek tűnik. Csak annyit kell tennünk, hogy elvesszük a JSON könyvtárat és beolvassuk a dataframe-be. A Spark létrehozza a sémát, a beágyazott adatokat struktúrákká alakítja. Ezután mindent parquet-ba kell menteni, ami Impalában is támogatott, regisztrálva a kirakatot a Hive metastore-ban.
Úgy tűnik, hogy minden egyszerű.
Azonban a dokumentáció rövid példáiból nem világos, hogy a gyakorlatban mit kell tenni számos problémával kapcsolatban.
A dokumentáció olyan megközelítést ír le, amely nem a kirakat létrehozására, hanem a JSON vagy XML dataframe-be való olvasására vonatkozik.
Nevezetesen, egyszerűen bemutatja, hogyan lehet olvasni és elemezni a JSON-t:
df = spark.read.json(path...)Ez elegendő ahhoz, hogy az adatok elérhetővé váljanak a Spark számára.
A gyakorlatban azonban a forgatókönyv sokkal bonyolultabb, mint egyszerűen csak beolvasni a JSON fájlokat a könyvtárból és létrehozni egy dataframe-et. A helyzet úgy néz ki: már van egy meghatározott kirakat, minden nap új adatok érkeznek, amelyeket hozzá kell adni a kirakathoz, figyelembe véve, hogy a séma eltérhet.
A kirakat általános felépítése a következő:
1. lépés. Az adatok a Hadoopba töltődnek be, majd naponta újra feltöltik őket, és egy új partícióba helyezik. Így egy, napok szerint particionált mappa jön létre az alapadatokkal.
2. lépés. A kezdeti betöltés során ezt a mappát a Spark olvassa és elemzi. A kapott dataframet egy elemzésre alkalmas formátumban, például parquet formátumban mentjük el, amelyet később importálhatunk az Impalába. Így létrejön a célzott adatfelület, amely tartalmazza az összes eddig felhalmozott adatot.
3. lépés. Létrehozunk egy betöltést, amely minden nap frissíti az adatfelületet.
Felmerül az inkrementális betöltés kérdése, az adatfelület particionálásának szükségessége és az adatfelület általános sémájának támogatása.
Tekintsünk egy példát. Tegyük fel, hogy megvalósítottuk az adattár felépítésének első lépését, és beállítottuk a JSON fájlok exportálását a mappába.
Ezekből dataframet kell létrehozni, hogy később vitrinként menthessük őket, nincs nehézség. Ez az a bizonyos első lépés, amelyet könnyen megtalálhatunk a Spark dokumentációjában:
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)Úgy tűnik, minden rendben van.
Elolvastuk és elemeztük a JSON-t, ezután a dataframet parquet formátumban mentjük el, bármilyen kényelmes módon regisztrálva a Hive-ba:
df.write.format(“parquet”).option('path','').saveAsTable('') Megkapjuk az adatfelületet.
De másnap új adatok érkeznek a forrásból. Van egy mappánk a JSON fájlokkal, és egy adatfelület, amely ezen mappa alapján készült. A következő adatcsomag betöltése után hiányzik egy nap adata az adatfelületből.
Logikus megoldás lenne particionálni az adatfelületet napokra, ami lehetővé teszi, hogy minden következő nap új partíciót adjunk hozzá. E mechanizmusról is jól ismertek a Spark általi külön írás partíciókra.
Először elvégezzük a kezdeti betöltést, a fentiek szerint eltárolva az adatokat, mindössze a particionálást hozzáadva. Ezt a lépést inicializálásként nevezzük, és csak egyszer hajtjuk végre:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable)
Másnap csak az új partíciót töltjük be:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable +"\/date_load=" + date_load + "\/" )
Csak újra kell regisztrálni a Hive-ban, hogy frissítsük a sémát.
Ezzel azonban problémák merülnek fel.
Az első probléma. Előbb-utóbb a létrehozott parquet fájl olvashatatlanná válik. Ennek oka a parquet és a JSON eltérő megközelítése az üres mezők kezelésében.
Nézzük meg a tipikus helyzetet. Például tegnap érkezik egy JSON:
1. nap: {"a": {"b": 1}},
de ma ez a JSON így néz ki:
2. nap: {"a": null}
Tegyük fel, hogy van két különböző partíció, mindegyikben egy sorral.
Amikor az eredeti adatokat teljes terjedelmükben olvassuk, a Spark képes meghatározni a típust, és megérti, hogy az „a” mező egy „strukturált” mező, amelynek van egy „b” mezője, INT típussal. De ha minden partíciót külön-külön mentettek el, akkor a partíciók inkompatibilis sémákat hoznak létre:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Ez a helyzet jól ismert, ezért egy opciót is hozzáadtak – az eredeti adatok elemzésekor eltávolítani az üres mezőket:
df = spark.read.json("...", dropFieldIfAllNull=True)
Ebben az esetben a parquet olyan partíciókból fog állni, amelyeket együtt olvashatunk.
Bár akik ezt a gyakorlatban csinálták, itt keserűen elmosolyodnak. Miért? Mert valószínűleg még két, három vagy négy helyzet merül fel. Az első, amely valószínűleg majdnem biztosan meg fog történni, hogy a numerikus típusok eltérően fognak megjelenni különböző JSON fájlokban. Például, {intField: 1} és {intField: 1.1}. Ha ezek a mezők egy partícióban találkoznak, akkor a sémák egyesítése mindent helyesen olvas be, a legpontosabb típusra konvertálva. De ha különböző partíciókban vannak, akkor az egyikben intField: int, a másikban intField: double lesz.
Ennek a helyzetnek a kezelésére van egy következő zászló:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Most van egy mappánk, ahol azok a partíciók találhatók, amelyeket egyetlen dataframe-be olvashatunk és érvényes parquet-t nyerhetünk. Igen? Nem.
Ne felejtsük el, hogy regisztráltuk a táblát a Hive-ban. A Hive nem érzékeny a mezőnevek kis- és nagybetűire, míg a parquet érzékeny. Ezért a partíciók sémái: field1: int és Field1: int a Hive számára ugyanazok, a Spark számára viszont nem. Nem szabad elfelejteni a mezőneveket kisbetűs formára hozni.
Itt aztán úgy tűnik, hogy minden rendben van.
De nem minden olyan egyszerű. Egy második, szintén jól ismert probléma merül fel. Mivel minden új partíció külön-külön van elmentve, a partíció mappájában Spark szolgáltató fájlok lesznek, például a művelet sikerességét jelző _SUCCESS zászló. Ez hibát fog okozni a parquet próbálkozásakor. Ennek elkerülése érdekében be kell állítani a konfigurációt, hogy megakadályozzuk a Spark-ot a szolgáltató fájlok mappába való írásában:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Úgy tűnik, most már minden nap új parquet partició kerül a célvitrin mappájába, ahol a napi feldolgozott adatok találhatók. Előre aggódtunk amiatt, hogy ne legyenek particiók, amelyek adat típuskonfliktusokkal rendelkeznek.
De előttünk áll a harmadik probléma. Most a teljes séma nem ismert, sőt, a Hive táblázat hibás sémával rendelkezik, mivel valószínűleg minden új partició torzítást hozott a sémába.
Újra kell regisztrálni a táblázatot. Ezt egyszerűen megtehetjük: újra beolvassuk a parquet vitrin adatokat, átveszünk egy sémát és létrehozunk egy DDL-t, amellyel újra regisztráljuk a mappát a Hive-ban külső táblázatként, frissítve a célvitrin sémáját.
Elénk áll a negyedik probléma. Amikor először regisztráltuk a táblázatot, a Sparkra támaszkodtunk. Most ezt magunk végezzük, és emlékeznünk kell arra, hogy a parquet mezők olyan karakterekkel kezdődhetnek, amelyek nem megengedettek a Hive-ban. Például a Spark eltávolítja azokat a sorokat, amelyeket nem tudott elemezni a "corrupt_record" mezőbe. Ilyen mezőt nem lehet regisztrálni a Hive-ban anélkül, hogy ne lennének escape-elve.
Ezt tudva, kapjuk a sémát:
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)
Kód ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`, "array<") biztonságos DDL-t készít, azaz a következő helyett:
create table tname (_field1 string, 1field string)
Olyan mezőnevekkel, mint az «_field1, 1field», biztonságos DDL-t készítünk, ahol a mezőnevek escape-elve vannak: create table `tname` (`_field1` string, `1field` string).
Felmerül a kérdés: hogyan lehet helyesen megszerezni a dataframe-t a teljes sémával (a kódban pf)? Hogyan lehet megszerezni ezt a pft? Ez az ötödik probléma. Az összes partició séma újraolvasása a parquet fájlokkal rendelkező célvitrin mappájából? Ez a legbiztonságosabb módszer, de nehéz.
A séma már létezik a Hive-ban. Új sémát úgy kaphatunk, hogy egyesítjük a teljes táblázat séma és az új partíció séma. Ezért a táblázat sémáját a Hive-ból kell venni, és össze kell vonni az új partíció sémájával. Ezt úgy lehet elérni, hogy a Hive-tól elolvassuk a teszt metadatákat, elmentjük azokat egy ideiglenes mappába, és a Sparkkal egyszerre olvassuk be mindkét partíciót.
Lényegében minden rendelkezésre áll: az alap táblázat séma a Hive-ban és az új partíció. Az adataink is megvannak. Csak annyit kell tenni, hogy megszerezzük az új sémát, amely egyesíti a bemutató séma és az új mezők a létrehozott partícióból:
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/*")
Ezután létrehozunk egy DDL-t a táblázat regisztrálásához, mint az előző részletben.
Ha az egész lánc helyesen működik, nevezetesen, hogy volt egy inicializáló betöltés, és a Hive-ban megfelelően létrehozott táblázat van, akkor megkapjuk a frissített táblázat sémát.
Az utolsó probléma az, hogy nem lehet csak úgy egyszerűen hozzáadni egy partíciót a Hive táblázathoz, mert az megsérülhet. Kényszeríteni kell a Hive-ot, hogy javítsa a partíciók szerkezetét:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Egy egyszerű JSON-olvasási és a bemutató létrehozásának feladata számos rejtett nehézség lekyüzdésévé válik, amelyek megoldásait külön-külön kell keresni. Noha ezek a megoldások egyszerűek, a keresésükre sok időt kell fordítani.
A bemutató felépítése érdekében:
- Partíciók hozzáadása a bemutatóhoz, megszabadulva a rendszerfájloktól
- Megérteni az üres mezőket az alapadatokban, amelyeket a Spark típusmagasra állított
- Az egyszerű típusokat karakterlánccá kell alakítani
- A mezők nevét kisbetűsre kell hozni
- Szét kell választani az adatok letöltését és a táblázat regisztrálását a Hive-ban (DDL létrehozása)
- Ne feledje, hogy el kell rejteni az olyan mezőneveket, amelyek esetleg nem kompatibilisek a Hive-val
- Meg kell tanulni a táblázat regisztrálásának frissítését a Hive-ban
Összegzésül elmondható, hogy a bemutatók létrehozása sok csapdát rejt magában. Ezért, ha nehézségek merülnek fel a megvalósítás során, érdemes tapasztalt partnert keresni sikeres szakértelemmel.
Köszönjük, hogy elolvasta ezt a cikket, reméljük, hogy az információ hasznos lesz.
Forrás: habr.com
Megbízható hosting vásárlása weboldalak számára, DDoS védelemmel, VPS VDS szerverek. 🔥 Megbízható webtárhely vásárlása DDoS védelemmel, VPS VDS szerverek | ProHoster
