Kære læsere, god dag!
I denne artikel beskriver den ledende konsulent for forretningsområdet Big Data Solutions i firmaet „Neoflex“ nøje mulighederne for at opbygge variable strukturvitriner ved hjælp af Apache Spark.
I forbindelse med et projekt om dataanalyse opstår der ofte behov for at opbygge vitrer baseret på svagt strukturerede data.
Det er typisk logs eller svar fra forskellige systemer, gemt i formatet JSON eller XML. Data eksporteres til Hadoop, og derfra skal der opbygges en vitrine. Adgang til den oprettede vitrine kan eksempelvis organiseres via Impala.
I dette tilfælde er skemaet for den målrettede vitrine på forhånd ukendt. Desuden kan skemaet ikke sammensættes på forhånd, da det afhænger af dataene, og vi har med netop disse svagt strukturerede data at gøre.
For eksempel logges et sådant svar i dag:
{source: "app1", error_code: ""}og i morgen kommer der et sådant svar fra samme system:
{source: "app1", error_code: "error", description: "Netværksfejl"}Som et resultat skal der tilføjes et andet felt — description, og om det kommer eller ej, ved ingen.
Opgaven med at skabe en vitrine baseret på sådanne data er ret standard, og Spark har en række værktøjer til dette. Der er støtte til både JSON og XML til parsing af de oprindelige data, samt støtte til schemaEvolution for et skema, der ikke er kendt på forhånd.
Ved første øjekast ser løsningen enkel ud. Man skal tage mappen med JSON og læse den ind i en dataframe. Spark vil skabe skemaet, og indlejrede data vil blive omdannet til strukturer. Derefter skal alt gemmes i parquet, som også understøttes i Impala, ved at registrere vitringen i Hive metastore.
Det virker til at være simpelt.
Men ud fra de korte eksempler i dokumentationen er det ikke klart, hvad man skal gøre med en række problemer i praksis.
Dokumentationen beskriver en tilgang, der ikke handler om at skabe en vitrine, men om at læse JSON eller XML ind i en dataframe.
Specifikt viser det blot, hvordan man læser og parser JSON:
df = spark.read.json(path...)Det er tilstrækkeligt til at gøre data tilgængelige for Spark.
I praksis er scenariet meget mere komplekst end blot at læse JSON-filer fra en mappe og skabe en dataframe. Situationens udfordring er, at der allerede findes en bestemt vitrine, som hver dag får nye data, der skal føjes til vitrinen, uden at glemme at skemaet kan variere.
Den almindelige skemaopbygning af en vitrine ser således ud:
Trin 1. Dataene indlæses i Hadoop med en efterfølgende daglig opdatering og gemmes i et nyt partition. Det resulterer i en dagligt partitioneret mappe med de rå data.
Trin 2. Under den indledende indlæsning læses og parses denne mappe af Spark-værktøjer. Den resulterende dataframe gemmes i et format, der er tilgængeligt for analyse, for eksempel i parquet, som senere kan importeres til Impala. Således skabes et mål-vitrine med alle de data, der er akkumuleret indtil dette tidspunkt.
Trin 3. Der oprettes en indlæsning, der dagligt opdaterer vitringen.
Spørgsmålet om inkrementel indlæsning opstår, behovet for at partitionere vitringen og spørgsmålet om at opretholde en samlet skemamodel for vitringen.
Lad os give et eksempel. Antag, at det første trin i opbygningen af lageret er blevet implementeret, og der er konfigureret en eksport af JSON-filer til mappen.
At oprette en dataframe ud af dem for derefter at gemme som vitring, er ikke et problem. Det er netop det første skridt, som let kan findes i Spark-dokumentationen:
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)Alt ser tilsyneladende godt ud.
Vi har læst og parsed JSON, og næste skridt er at gemme dataframen som parquet, registrere den i Hive på en hvilken som helst bekvem måde:
df.write.format(“parquet”).option('path','').saveAsTable('') Vi får vitringen.
Men næste dag er der tilføjet nye data fra kilde. Vi har en mappe med JSON, og vitringen, som er oprettet på basis af denne mappe. Efter indlæsning af den næste portion data fra kilden mangler der data i vitringen for en dag.
En logisk løsning ville være at partitionere vitringen efter dage, hvilket ville gøre det muligt at tilføje en ny partition hver næste dag. Denne mekanisme er også velkendt, Spark tillader at skrive partitioner separat.
Først laver vi en indledende indlæsning, gemmer dataene som beskrevet ovenfor, men tilføjer blot partitionering. Denne handling kaldes initialization af vitringen og gøres kun én gang:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable)
Næste dag indlæser vi kun den nye partition:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable +"\/date_load=" + date_load + "\/" )
Det eneste, der er tilbage, er at registrere det i Hive for at opdatere skemaet.
Men her opstår problemerne.
Det første problem. På et eller andet tidspunkt vil den genererede parquet ikke kunne læses. Dette skyldes, at parquet og JSON håndterer tomme felter forskelligt.
Lad os overveje en typisk situation. For eksempel, i går kom der en JSON:
Dag 1: {"a": {"b": 1}},
men i dag ser den samme JSON sådan ud:
Dag 2: {"a": null}
Lad os antage, at vi har to forskellige partitioner, hvor der er én række i hver.
Når vi læser de oprindelige data i sin helhed, vil Spark kunne bestemme typen og forstå, at "a" er et felt af typen "struktur", med det indlejrede felt "b" af typen INT. Men hvis hver partition blev gemt hver for sig, ville det resultere i parquet med inkompatible partitionskemaer:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Denne situation er godt kendt, og derfor er der specielt tilføjet en mulighed - ved parsing af de oprindelige data at fjerne tomme felter:
df = spark.read.json("...", dropFieldIfAllNull=True)
I dette tilfælde vil parquet bestå af partitioner, som kan læses sammen.
Skønt dem, der har gjort dette i praksis, vil her bittert smile. Hvorfor? Fordi der sandsynligvis vil opstå endnu to situationer. Eller tre. Eller fire. Den første, der næsten helt sikkert vil opstå, er, at numeriske typer vil fremstå forskelligt i forskellige JSON-filer. For eksempel, {intField: 1} og {intField: 1.1}. Hvis sådanne felter findes i samme partition, vil mergen af skemaet læse alt korrekt og konvertere til den mest nøjagtige type. Men hvis de er i forskellige partitioner, vil den ene have intField: int, mens den anden har intField: double.
Til behandling af denne situation er der følgende flag:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Nu har vi en mappe, hvor partitionerne findes, som kan læses ind i en enkelt dataframe og en gyldig parquet for hele lageret. Ja? Nej.
Man skal huske, at vi registrerede tabellen i Hive. Hive er ikke følsom over for store og små bogstaver i felt-navne, mens parquet er. Derfor vil partitioner med skemaer: field1: int og Field1: int være ens for Hive, men ikke for Spark. Man skal ikke glemme at konvertere felt-navne til små bogstaver.
Efter dette ser det ud til, at alt er i orden.
Dog er ikke alt så simpelt. Der opstår et andet, også velkendt problem. Da hver ny partition gemmes separat, vil der i partitionens mappe være systemfiler fra Spark, for eksempel flaget for operationens succes _SUCCESS. Det vil føre til en fejl ved forsøg på parquet. For at undgå dette skal konfigurationen indstilles til at forbyde Spark at skrive systemfiler til mappen:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Det ser ud til, at der hver dag tilføjes en ny parquet-partition til destinationskataloget, hvor de parsede data for dagen ligger. Vi har på forhånd taget højde for, at der ikke skulle være partitioner med datakonflikter.
Men vi står overfor det tredje problem. Nu er den samlede skema ukendt, desuden er Hive-tabellen med et forkert skema, da hver ny partition sandsynligvis har forvansket skemaet.
Vi skal registrere tabellen igen. Dette kan gøres let: læse parquet-vitrinen igen, tage skemaet og skabe et DDL ud fra det, hvormed vi igen registrerer mappen i Hive som en ekstern tabel, og opdaterer skemaet for destinationsvitrinen.
Vi står over for det fjerde problem. Da vi registrerede tabellen første gang, støttede vi os til Spark. Nu gør vi det selv, og vi skal huske, at parquet-felter kan starte med tegn, der ikke er tilladte for Hive. For eksempel smider Spark de rækker, som den ikke kunne parse i feltet «corrupt_record». Et sådant felt kan ikke registreres i Hive uden at blive escaperet.
Når vi ved dette, får vi skemaet:
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)
Kode ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace(«array<`», «array<» lavede en sikker DDL, hvilket vil sige i stedet for:
create table tname (_field1 string, 1field string)
Med feltnavne som «_field1, 1field» bliver der lavet en sikker DDL, hvor feltnavnene er escaperede: create table `tname` (`_field1` string, `1field` string).
Spørgsmålet opstår: hvordan får man korrekt en dataframe med det fulde skema (i koden pf)? Hvordan får man denne pf? Dette er det femte problem. At læse skemaet for alle partitioner fra mappen med parquet-filerne til destinationsvitrinen? Denne metode er den sikreste, men også den tungeste.
Skemaen findes allerede i Hive. For at få en ny skema kan man kombinere skemaet for hele tabellen og den nye partition. Det betyder, at man skal tage skemaet for tabellen fra Hive og kombinere det med skemaet for den nye partition. Dette kan gøres ved at læse testmetadata fra Hive, gemme dem i en midlertidig mappe og læse begge partitioner samtidig med Spark.
I princippet har vi alt, hvad der skal til: den oprindelige skema for tabellen i Hive og den nye partition. Dataene har vi også. Det resterende er blot at få den nye skema, hvor skemaet for vinduet kombineres med de nye felter fra den oprettede partition:
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/*")
Dernæst opretter vi DDL-en for registrering af tabellen, som i det foregående fragment.
Hvis hele kæden fungerer korrekt, dvs. der har været en initiel indlæsning, og den korrekt oprettede tabel findes i Hive, så får vi den opdaterede tabelskema.
Og det sidste problem er, at man ikke bare kan tilføje en partition til Hive-tabellen, da dette vil bryde den. Det er nødvendigt at få Hive til at reparere sin partitionsstruktur:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
En simpel opgave med at læse JSON og oprette et vindue derudfra, viser sig at være forbundet med at overvinde en række implicitte udfordringer, som man må finde løsninger til hver for sig. Selvom disse løsninger er enkle, kan det tage meget tid at finde dem.
For at implementere opbygningen af vinduet, var man nødt til at:
- Tilføje partitioner til vinduet og slippe af med systemfiler
- Forstå de tomme felter i de oprindelige data, som Spark har typet
- Konvertere enkle typer til strenge
- Sætte feltnavne i små bogstaver
- Adskille udtræk af data og registrering af tabellen i Hive (oprette DDL)
- Ikke glemme at escape feltnavne, som muligvis er uforenelige med Hive
- Lære at opdatere registreringen af tabellen i Hive
Afslutningsvis er det værd at bemærke, at løsningen til opbygning af vinduer gemmer mange faldgruber. Derfor er det bedst at kontakte en erfaren partner med succesfuld ekspertise, hvis der opstår vanskeligheder i implementeringen.
Tak for at læse denne artikel, vi håber, informationen vil være nyttig.
Kilde: habr.com
Køb pålidelig webhosting med DDoS-beskyttelse, VPS VDS-servere 🔥 Køb pålidelig webhosting med DDoS-beskyttelse, VPS VDS-servere | ProHoster
