Spark schemaEvolution praktikas

Lugupidajad, tere päevast!

Käesolevas artiklis käsitleb Big Data Solutions äriüksuse juhtiv nõustaja ettevõttest "Neoflex" üksikasjalikult võimalusi muuta muutuva struktuuriga vitriine Apache Sparki abil.

Andmeanalüüsiprojekti raames tekib sageli vajadus luua vitriine halvasti struktureeritud andmete põhjal.

Need on tavaliselt logid või erinevate süsteemide vastused, mis salvestatakse JSON-i või XML-i kujul. Andmed laaditakse Hadoopisse ja seejärel tuleb nendest vitriin luua. Juurdepääsu loodud vitriinile saame korraldada näiteks läbi Impala.

Sel juhul ei ole sihtvitriini skeem eelnevalt teada. Veelgi enam, skeemi ei saa ka eelnevalt koostada, kuna see sõltub andmetest ja meil on tegemist nende halvasti struktureeritud andmetega.

Näiteks tänasel päeval logitakse selline vastus:

{source: "app1", error_code: ""}

ja homme tuleb samalt süsteemilt selline vastus:

{source: "app1", error_code: "error", description: "Network error"}

Seetõttu peaks vitriini lisanduma veel üks väli — description, ja kas see tuleku või mitte, ei tea keegi.

Vitriini loomise ülesanne selliste andmete põhjal on üsna tavaline, ja Sparkil on selleks olemas mitmeid tööriistu. Algandmete analüüsimiseks on toetatud nii JSON kui ka XML, ning eelnevalt tundmatu skeemi puhul on olemas ka schemaEvolution toe.

Esmapilgul näeb lahendus lihtne välja. Tuleb võtta kaust JSON-ide ja lugeda see dataframe'i. Spark loob skeemi, sisemised andmed muudab struktuuri. Edasi tuleb kõik salvestada parquet'isse, mis on samuti toetatud Impalas, registreerides vitriini Hive metastore'is.

Tundub, et kõik on lihtne.

Kuid lühikestest näidistest dokumendis ei ole selge, mida praktikas mitmete probleemidega ette võtta.

Dokumentatsioonis kirjeldatakse lähenemist mitte vitriini loomiseks, vaid JSON-i või XML-i lugemiseks dataframe'i.

Konkreetselt, kirjeldatakse lihtsalt, kuidas lugeda ja analüüsida JSON-i:

df = spark.read.json(path...)

Seda on piisavalt, et andmed oleksid Sparki jaoks kergesti kättesaadavad.

Kuid praktikas on stsenaarium palju keerulisem kui lihtsalt lugeda JSON-failid kaustast ja luua dataframe. Olukord on selline: juba on olemas kindel vitriin, igapäev tulevad uued andmed, need tuleb vitriini lisada, meeles pidades, et skeem võib varieeruda.

Tavaline vitriini ülesehitamise skeem on järgmine:

Samm 1. Andmed laaditakse Hadoopisse, millele järgneb igapäevane täiendamine ja need koondatakse uude partitsiooni. Saame päevade kaupa partitsioneeritud kausta, kus on algandmed.

Samm 2. Initsialiseerimise käigus loetakse ja analüüsitakse seda kausta Sparkiga. Saadud dataframe salvestatakse analüüsiks sobivasse formaati, näiteks parquet, mille saab hiljem Impalasse importida. Nii luuakse sihtkapp, kus on kõik senikogutud andmed.

Samm 3. Luakse laadimine, mis igapäevaselt uuendab võti.
Küsimus on inkrementaalses laadimises, vitriini partitsioneerimises ja üldise vitriini skeemi toetamise vajaduses.

Toome näite. Oletame, et hoidlahoone esimene etapp on teostatud ja JSON-failide eksportimine kausta on seadistatud.

Nendest dataframe'i loomine, et hiljem salvestada see vitriinina, ei valmista probleeme. See on just see esimene samm, mida on lihtne leida Spark'i dokumentatsioonist:

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)

Tundub, et kõik on korras.

Me oleme JSON'i lugenud ja analüüsinud ning nüüd salvestame dataframe'i parquet'ina, registreerides Hive'is igal sobival viisil:

df.write.format(“parquet”).option('path','').saveAsTable('')
 Saame vitriini.
Kuid järgmisel päeval lisandusid andmed allikast. Meil on kaust JSON-idega ja vitriin, mis on loodud selle kausta põhjal. Pärast järgmise andmepartii laadimist allikast puuduvad vitriinis ühe päeva andmed.
Loogiliseks lahenduseks on partitsioneerida vitriin päevade kaupa, mis võimaldab igapäevaselt lisada uue partitsiooni. Selle mehhanism on samuti hästi tuntud, Spark võimaldab partitsioone eraldi salvestada.
Esmalt teeme initsialiseerimise laadimise, salvestades andmed nagu eespool kirjeldatud, lisades ainult partitsioneerimise. See toiming on vitriini initsialiseerimine ja see tehakse ainult üks kord:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)
 Järgmisel päeval laadime ainult uue partitsiooni:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable + "/date_load=" + date_load + "/")
 Jääb ainult Hive'is uuesti registreerida, et skeemi värskendada.
 Kuid siin tekivad probleemid.
Esimene probleem. A varem või hiljem ei saa loodud parquet'i enam lugeda. See on seotud sellega, kuidas parquet ja JSON käsitlevad tühje välju erinevalt.
Vaadakem tüüpilist olukorda. Näiteks, eile tuleb JSON:
Päev 1: {"a": {"b": 1}},
 aga täna näeb see sama JSON välja nii:
Päev 2: {"a": null}
 Oletame, et meil on kaks erinevat partiid, kus kummaski on üks rida.
 Kui me loeme algandmed täielikult, suudab Spark tuvastada tüübi ja mõista, et "a" on struktuuri tüüpi väli, mille sees on INT tüüpi väli "b". Kuid kui iga partii on salvestatud eraldi, tekib parquet, millel on mitteühtuvad partii skeemid:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
 See olukord on hästi teada, seetõttu on spetsiaalselt lisatud valik - algandmete parsimisel tühjad väljad eemaldada:
df = spark.read.json("...", dropFieldIfAllNull=True)
 Sel juhul koosneb parquet partiiidest, mida saab koos lugeda. 
 Kuigi need, kes on seda praktikas teinud, naeratavad sellele kibedalt. Miks? Sest ilmselt tekib veel kaks olukorda. Või kolm. Või neli. Esimene, mis peaaegu kindlasti tekib, on see, et numbrilised tüübid näevad erinevates JSON failides erinevad välja. Näiteks, {intField: 1} ja {intField: 1.1}. Kui sellised väljad satuvad ühte partiisse, loeb skeemi mergenõuet kõik õigesti, tuues välja kõige täpsema tüübi. Kuid kui need on erinevates partii, siis ühes on intField: int ja teises intField: double.
Selle olukorra töötlemiseks on järgmine lipp:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
 Nüüd on meil kaust, kus asuvad partiiid, mida saab lugeda üheks dataframe'iks ja kehtivaks parquet'iks kogu vitriini jaoks. Jah? Ei.
Peame meenutama, et registreerisime tabeli Hive'is. Hive ei ole tundlik suuruse suhtes väljade nimedes, kuid parquet on tundlik. Seetõttu on partii skeemid: field1: int ja Field1: int Hive jaoks samad, kuid Sparkile mitte. Tuleb meeles pidada, et muudame väljade nimed väikeste tähtedega.
Pärast seda tundub, et kõik on korras.
Kuid asi ei ole nii lihtne. Tekib teine, samuti hästi tuntud probleem. Kuna iga uus partii salvestatakse eraldi, on partii kaustas Spark'i teenindusfailid, näiteks operatsiooni edukuse lipp _SUCCESS. See põhjustab viga, kui proovida parquet'it. Selle vältimiseks tuleb seadistada konfiguratsioon, keelates Spark'il teenindusfailide kirjutamise kausta:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
 Tundub, nüüd lisatakse igapäevaselt sihtkausta uusi parquet partiisid, kus on päeva jooksul analüüsitud andmed. Oleme eelnevalt hoolitsenud selle eest, et andmetüüpide konfliktidega partiisid ei esineks.
Kuid nüüd seisame silmitsi kolmanda probleemiga. Nüüd ei ole üldine skeem teada, veel enam, Hive'is on vale skeem, kuna iga uus partiia on tõenäoliselt skeemi moonutanud.
Tabel tuleb uuesti registreerida. Selle saab teha lihtsalt: loeme uuesti parquet vitriine, võtame skeemi ja loome selle põhjal DDL, millega registreerime kausta Hive'is uuena välise tabelina, värskendades sihtvitriini skeemi.
Nüüd tekib neljas probleem. Kui registreerisime tabeli esmakordselt, tuginesime Sparkile. Nüüd teeme seda ise ja tuleb meeles pidada, et parquet'i väljad võivad alata märkidest, mis ei ole Hive'is lubatud. Näiteks viskab Spark välja read, mida ei suudeta "corrupt_record" välja analüüsida. Sellist välja ei saa Hive'is registreerida ilma, et see oleks escapesitud.
Teades seda, saame skeemi:
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)
 Kood ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") teeb turvalise DDL, st selle asemel:
create table tname (_field1 string, 1field string)
 Selliste väljanimega nagu „_field1, 1field“ tehakse turvaline DDL, kus välja nimed on escapesitud: create table `tname` (`_field1` string, `1field` string).
Küsimus on: kuidas õigesti saada dataframe täieliku skeemiga (koodis pf)? Kuidas saada see pf? See on viies probleem. Kas lugeda skeemi kõikidest partiidest sihtkaustas olevatest parquet failidest? See meetod on kõige turvalisem, kuid keeruline.
Schema on juba Hive'is. Uue skeemi saamiseks tuleb ühendada kogu tabeli skeem ja uus partitsioon. See tähendab, et tuleb võtta tabeli skeem Hive'ist ja ühendada see uue partitsiooni skeemiga. Seda saab teha, lugedes testmetainfot Hive'ist, salvestades selle ajutisse kausta ja lugedes Spark'i abil mõlemat partitsiooni korraga.
Sisuliselt on kõik vajalik olemas: algne tabeli skeem Hive'is ja uus partitsioon. Andmed on meil samuti olemas. Jääb vaid saada uus skeem, milles ühendatakse vitriiniskeem ja uued väljad loodud partitsioonist:
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/*")
 Seejärel loome DDL tabeli registreerimiseks, nagu eelmises fragmentis.
 Kui kogu ahel töötab õigesti, nimelt — oli algne laadimine ja Hive'is on õigesti loodud tabel, siis saame uuendatud tabeli skeemi.
Ja viimane probleem seisneb selles, et ei saa lihtsalt lisada partitsiooni Hive'i tabelisse, kuna see rikki läheb. Oluline on sundida Hive’i parandama oma partitsioonide struktuuri:
from pyspark.sql import HiveContext
hc = HiveContext(spark) 
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
 Lihtne ülesanne lugeda JSON ja luua selle alusel vitriin toob esile mitmeid varjatud raskusi, mille lahendused tuleb eraldi välja otsida. Ja kuigi need lahendused on lihtsad, võtab nende leidmine palju aega.
Vitriini loomise elluviimiseks tuli:
 
Lisada partitsioonid vitriini, vabastudes abifailidest
 
Selgitada välja tühjad väljad algandmetes, mille Spark tüüpides
 
Muuta lihtsad tüübid stringiks
 
Viia väljade nimed väikeste tähtedega
 
Eraldada andmete eksport ja tabeli registreerimine Hive'is (DDL loomine)
 
Ärge unustage ekraanida väljade nimesid, mis võivad olla Hive'iga mitteühilduvad
 
Õppida, kuidas tabeli registreerimist Hive'is uuendada
 Kokkuvõtteks tuleks märkida, et vitriinide loomise lahendusel on palju varjatud takistusi. Seetõttu on keerukate rakenduste puhul parem pöörduda kogenud partneri poole, kellel on edukas ekspertiis.
Tänan teid selle artikli lugemise eest, loodame, et teave osutub kasulikuks.
 
Allikas: habr.com
   
 Kasutame küpsiseid küpsised, et tagada parim kogemus meie saidil. Jätkates saidi kasutamist, nõustute meie privaatsuspoliitikaga. Nõustuda
 Osta usaldusväärne hostimine veebilehtede jaoks DDoS-i kaitsega, VPS VDS serverid 🔥 Osta usaldusväärne hostimine veebilehtede jaoks DDoS-i kaitsega, VPS VDS serverid | ProHoster