Spark schemaEvolution praktikas

Lugupidamine, head päeva!

Selles artiklis kirjeldab Big Data Solutionsi ärisuuna juhtiv konsultant ettevõttes «Neoflex» üksikasjalikult muutuva struktuuriga vitriinide loomise võimalusi, kasutades Apache Spark'i.

Andmeanalüüsi projekti raames tekib sageli vajadus vitriinide loomiseks halvasti struktureeritud andmete põhjal.

Tavaliselt on need logid või erinevate süsteemide vastused, mis salvestatakse JSON-i või XML-vormingus. Andmed laaditakse Hadoopi, sealt tuleb seejärel vitriin luua. Juurdepääsu loodud vitriinile saab korraldada näiteks läbi Impala.

Sellisel juhul ei ole sihtvitriini skeem eelnevalt teada. Veelgi enam, skeemi ei ole võimalik koostada ka ette, kuna see sõltub andmetest, millega me tegeleme — need on samad halvasti struktureeritud andmed.

Näiteks tänapäeval logitakse järgmine vastus:

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

ja homme tuleb samalt süsteemilt järgmine vastus:

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

Tulemuseks peaks vitriinile lisanduma veel üks väli — description, ja kas see tuleb või mitte, ei tea keegi.

Andmete esitlemine on üsna tavaline ülesanne ning Sparkil on selleks mitmeid tööriistu. Algandmete parsimiseks toetatakse nii JSON-i kui ka XML-i ning ettearvamatute skeemide jaoks on olemas schemaEvolution tugi.

Esmapilgul tundub lahendus lihtne. Tuleb võtta JSON-iga kaust ja lugeda see dataframe'i. Spark loob skeemi, muutes sisemised andmed struktuurideks. Edasi tuleb kõik salvestada parquet formaati, mida toetatakse ka Impalas, registreerides vitriini Hive metastore'is.

Tundub ju kõik lihtne.

Kuid lühikestest näidetest dokumentatsioonis ei ole selge, kuidas tegeleda mitmete praktiliste probleemidega.

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

Konkreetsemalt tuuakse lihtsalt välja, kuidas lugeda ja parsida JSON-i:

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

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

Praktikas on stsenaarium palju keerulisem kui lihtsalt lugeda JSON-faile kaustast ja luua dataframe. Situatsioon näeb välja selline: olemas on kindel vitriin, iga päev saabub uusi andmeid, mida tuleb vitriini lisada, unustamata, et skeem võib erineda.

Tüüpiline vitriini ehitamise skeem on järgmine:

Samm 1. Andmed laaditakse Hadoopisse koos järgneva igapäevase laadimisega ning need paigutatakse uude partitsiooni. Tulemuseks on päevade kaupa partitsioneeritud kaust algandmetega.

Samm 2. Alglaadimise käigus loetakse ja analüüsitakse seda kausta Spark'i vahendite abil. Saadud dataframe salvestatakse analüüsimiseks sobivasse formaati, näiteks parquet, mida saab hiljem importida Impalasse. Nii luuakse eesmärgipärane vitriin kõigi nende andmetega, mis selleks ajaks on kogunenud.

Samm 3. Luakse laadimise ülesanne, mis igal päeval vitriini uuendab.
Küsimuseks on inkrementaalne laadimine, vitriini partitsioneerimise vajadus ja üldise vitriini skeemi toetamise küsimus.

Toome näite. Oletame, et esimene samm andmehoidla loomisel on ellu viidud ja JSON-failide väljavõte on määratud kausta.

Nendest dataframe'i loomine, et seejärel säilitada vitriinina, ei ole probleem. See on see esimene samm, mille leidmine Spark'i dokumentatsioonist on lihtne:

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)

Näib, et kõik on hästi.

Oleme lugenud ja analüüsinud JSON-i, seejärel salvestame andmeraami parquet-vormingus, registreerides Hive'is ükskõik millise mugava meetodi abil:

df.write.format("parquet").option('path','').saveAsTable('')
 Saame vitriini.
Kuid järgmisel päeval lisandusid uued andmed allikast. Meil on kaust, kus on JSON, ja vitriin, mis on loodud selle kausta põhjal. Pärast järgmise andmepartii laadimist ei piisa vitriinis ühe päeva andmetest.
Loogiliseks lahenduseks oleks vitriini jagamine päevade kaupa, mis võimaldaks iga järgmise päeva lisada uue partitsiooni. Selle mehhanism on samuti hästi tuntud, Spark võimaldab partitsioone eraldi salvestada.
Esiteks teeme initsialiseeriva laadimise, säilitades andmed nagu eespool kirjeldatud, lisades ainult partitsioneerimise. Seda tegevust nimetatakse vitriini initsialiseerimiseks ja seda tehakse vaid ü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äänud on vaid Hive'is uuesti registreerida, et skeemi värskendada.
 Siin aga tekivad probleemid.
Esimene probleem. Varsti või hiljem ei saa loodud parquet'it lugeda. See on seotud sellega, kuidas parquet ja JSON käsitlevad tühje välju erinevalt.
Vaatame tüüpilist olukorda. Näiteks eile saabus JSON:
Päev 1: {"a": {"b": 1}},
 aga täna näeb sama JSON välja nii:
Päev 2: {"a": null}
 Oletame, et meil on kaks erinevat partitsiooni, milles kummaski on üks rida.
 Kui loeme algandmeid koos, suudab Spark määrata tüübi ning mõista, et „a” on struktuuriväli, mille sees on välisvälja „b” tüüp INT. Kuid kui iga partitsioon salvestati eraldi, saame parquet'iga, millel on kokkusobimatud partitsiooniskeemid:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
 See olukord on hästi tuntud, seetõttu on spetsiaalselt lisatud valik — algandmete töötlemisel eemaldada tühjad väljad:
df = spark.read.json("...", dropFieldIfAllNull=True)
 Sellisel juhul koosneb parquet partitsioonidest, mida saab ühiselt lugeda. 
 Kuigi need, kes seda praktikas tegid, naeratavad siin kibedalt. Miks? Sest tõenäoliselt tekib veel kaks situatsiooni. Või kolm. Või neli. Esimene, mis peaaegu kindlasti juhtub, on see, et numbritüübid näevad erinevalt välja erinevates JSON-failides. Näiteks, {intField: 1} ja {intField: 1.1}. Kui sellised väljad satuvad samasse partitsiooni, loeb skeemi fusioon kõik õigesti, muutes selle kõige täpsemaks tüübiks. Ent kui need on erinevates partitsioonides, on ühes 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 on partitsioonid, mida saab lugeda ühte andmeraamistikku ja valideeritud parquet kogu vitriini jaoks. Jah? Ei.
Peame meeles pidama, et registreerisime tabeli Hive'is. Hive ei ole tundlik väljade nimede suuruse osas, kuid parquet on tundlik. Seetõttu on partitsioonid skeemidega: field1: int ja Field1: int Hive'i jaoks samad, kuid Spark'i jaoks mitte. Peame meeles pidama, et väljade nimed tuleks muuta väiketähtedeks.
Pärast seda tundub kõik hästi.
Kuid mitte kõik pole nii lihtne. Tõuseb teine, samuti hästi tuntud probleem. Kuna iga uus partitsioon salvestatakse eraldi, siis on partitsiooni kaustas Spark'i teenindusfailid, näiteks operatsiooni edukuse lipp _SUCCESS. See toob kaasa vea, kui proovida parquet'i. Selle vältimiseks tuleb seadistada konfigureerimine, keelates Spark'il teenindusfailide kirjutamise kausta:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
 Paistab, et nüüd lisandub sihtvitriini kausta igapäevaselt uus parquet partitsioon, kus on päevased parsitud andmed. Oleme eelnevalt mures, et andmetüüpide konflikte ei tekiks.
Aga meie ees on kolmas probleem. Nüüd ei ole üldine skeem teada, lisaks on Hive'is vale skeem, kuna iga uus partitsioon on tõenäoliselt skeemi moonutanud.
Tabel tuleb uuesti registreerida. Seda saab teha lihtsalt: lugedes uuesti parquet vitriini, võtame skeemi ja loome selle põhjal DDL, millega uuesti registreerime kausta Hive'is välise tabelina, värskendades sihtvitriini skeemi.
Meie ees seisab neljas probleem. Kui me tabelit esmakordselt registreerisime, toetusime Sparkile. Nüüd teeme seda ise, ja peame meeles pidama, et parquet-failid võivad alata märkidega, mis Hive'is pole lubatud. Näiteks eemaldab Spark read, mida ta ei suutnud "corrupt_record" välja lugeda. Sellist välja ei saa Hive'is registreerida ilma, et see peaks olema ekraanituks.
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 ohutu DDL, mis tähendab, et selle asemel:
create table tname (_field1 string, 1field string)
 Selliste väljade nagu «_field1, 1field» korral luuakse ohutu DDL, kus välja nimed on ekraanitud: create table `tname` (`_field1` string, `1field` string).
Küsimus on, kuidas saada andmeraam, millel on täielik skeem (pf koodis)? Kuidas saada see pf? See on viies probleem. Kas lugeda kõigi partitsioonide skeemi sihivitrini parquet faili kaustast? See meetod on kõige turvalisem, kuid töömahukam.
Skeem on juba Hive'is. Uue skeemi saab hankida, kombineerides terve tabeli skeemi uue partitsiooniga. See tähendab, et tabeli skeem tuleb võtta Hive'ist ja ühendada see uue partitsiooni skeemiga. Seda saab teha, lugedes testmetainformatsiooni Hive'ist, salvestades need ajutisse kausta ja lugedes Sparkiga mõlemad partitsioonid korraga.
Põhimõtteliselt on kõik vajalik olemas: algne tabeli skeem Hive'is ja uus partitsioon. Andmed on meil samuti olemas. Jääb üle vaid saada uus skeem, milles kombineeritakse vitrini skeem 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 registreerimise, nagu eelnevas fragmendis.
 Kui kogu ahel töötab õigesti, st on toimunud alglaadimine ja Hive'is on õigesti loodud tabel, siis saame värskendatud tabeli skeemi.
Ja viimane probleem on see, et ei saa lihtsalt lisada partitsiooni Hive'i tabelisse, kuna see puruneb. On vajalik panna Hive 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 põhjal vitriin osutub mitmete varjatud raskuste ületamiseks, mille lahendusi tuleb otsida eraldi. Ja kuigi need lahendused on lihtsad, võtab nende leidmine palju aega.
Vitriini koostamiseks tuli:
 
Lisa vitriinile partitsioone, vabanedes abifailidest.
 
Selgitada välja tühjad väljad algandmetes, mille Spark tüübis.
 
Viia lihtsad tüübid stringideks.
 
Viia väljade nimed väikestesse tähtedesse.
 
Eraldada andmete väljaviimine ja tabeli registreerimine Hive'is (DDL loomine).
 
Ära unusta escape'ida väljade nimed, mis võivad Hive'iga ühilduvad olla.
 
Õppida, kuidas uuendada tabeli registreerimist Hive'is.
 Kokkuvõttes tasub märkida, et vitriinide ehitamise lahendusel on palju varjatud ohte. Seetõttu on keeruliste olukordade tekkimisel parem pöörduda kogenud partneri poole, kellel on edukas kogemus.
Aitäh, et lugesite seda artiklit, loodame, et teave osutub kasulikuks.
 
Allikas: habr.com
   
 Me kasutame küpsiseid, et tagada parim kogemus meie veebisaidil. Jätkates saidi kasutamist, nõustute meie privaatsuspoliitikaga. Nõustuda
 Osta usaldusväärne veebihosting DDoS kaitsega, VPS VDS serverid 🔥 Osta usaldusväärne veebihosting DDoS kaitsega, VPS VDS serverid | ProHoster