Spark schemaEvolution i praktiken

Kära läsare, god dag!

I denna artikel beskriver en ledande konsult inom affärsområdet Big Data Solutions på företaget "Neoflex" noggrant alternativ för att bygga variabla strukturvitriner med hjälp av Apache Spark.

Inom ramen för ett dataanalysprojekt uppstår ofta uppgiften att bygga vitriner baserat på svagt strukturerade data.

Oftast handlar det om loggar, eller svar från olika system, som sparas i JSON- eller XML-format. Data exporteras till Hadoop, och därifrån behöver vi bygga en vitrin. Vi kan organisera åtkomsten till den skapade vitrinen, till exempel via Impala.

I det här fallet är schemat för den målinriktade vitrinen preliminärt okänt. Dessutom kan schemat inte fastställas i förväg, eftersom det beror på data och vi hanterar dessa svagt strukturerade data.

Till exempel, idag loggas följande svar:

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

och nästa dag kommer ett sådant svar från samma system:

{source: "app1", error_code: "error", description: "Nätverksfel"}

Som ett resultat ska ett ytterligare fält - description - läggas till vitrinen, och ingen vet om det kommer eller inte.

Uppgiften att skapa en vitrin med sådana data är ganska standard, och Spark har flera verktyg för detta. För att parsa källdata stöds både JSON och XML, och för okänt schema finns stöd för schemaEvolution.

Vid första anblicken verkar lösningen enkel. Man behöver bara ta en mapp med JSON och läsa in i en dataframe. Spark kommer att skapa ett schema, och inbäddade data kommer att omvandlas till strukturer. Därefter behöver allt sparas i parquet, som även stöds av Impala, genom att registrera vitrinen i Hive metastore.

Det verkar vara allt enkelt.

Men av de korta exemplen i dokumentationen framgår det inte hur man ska hantera flera problem i praktiken.

I dokumentationen beskrivs tillvägagångssättet inte för att skapa en vitrin, utan för att läsa JSON eller XML i en dataframe.

Specifikt visas det bara hur man läser och parserar JSON:

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

Detta är tillräckligt för att göra data åtkomliga för Spark.

I praktiken är scenariot mycket mer komplicerat än att bara läsa JSON-filer från en mapp och skapa en dataframe. Situationen ser ut så här: det finns redan en viss vitrin, nya data kommer varje dag, och de måste läggas till i vitrinen utan att glömma att schemat kan variera.

Den vanliga strukturen för att bygga en vitrin är följande:

Step 1. Data laddas upp i Hadoop med en efterföljande daglig uppdatering och placeras i en ny partition. Det blir en partitionerad mapp med källdata per dag.

Step 2. Under den initierande laddningen läses och parseas denna mapp med hjälp av Spark. Den erhållna dataframe sparas i ett format som är tillgängligt för analys, till exempel i parquet, vilket sedan kan importeras till Impala. Så skapas en målvitrine med alla data som har samlats vid detta tillfälle.

Step 3. En laddning skapas som kommer att uppdatera vitrin varje dag.
Frågan om inkrementell laddning uppstår, behovet av att partitionera vitrin och frågan om att stödja den övergripande schemat för vitrin.

Låt oss ta ett exempel. Antag att det första steget av byggandet av lagret har genomförts och export av JSON-filer till mappen har konfigurerats.

Att skapa en dataframe från dem för att sedan spara som vitrin är inget problem. Det är just det första steget som lätt kan hittas 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)

Allt verkar bra.

Vi har läst och parseat JSON, och nu sparar vi dataframe som parquet, registrerar den i Hive på ett valfritt bekvämt sätt:

df.write.format(“parquet”).option('path','').saveAsTable('')
 Vi får vitrin.
Men, nästa dag kom nya data från källan. Vi har en mapp med JSON och en vitrin som skapades baserat på denna mapp. Efter att ha laddat nästa omgång data från källan saknas det information för en dag i vitrin.
En logisk lösning vore att partitionera vitrin efter dagar, vilket skulle tillåta oss att varje nästa dag lägga till en ny partition. Denna mekanism är också välkänd, Spark tillåter att skriva partitioner separat.
Först gör vi en initierande laddning, och sparar data som beskrivits ovan, bara med partitioneringen tillagd. Denna åtgärd kallas initiering av vitrin och görs endast en gång:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable)
 Nästa dag laddar vi bara den nya partitionen:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "/" + db + "/" + destTable +"/date_load=" + date_load + "/")
 Det återstår bara att registrera om i Hive för att uppdatera schemat.
 Men här uppstår problemen.
Det första problemet. Förr eller senare kommer den resulterande parquet-filen att bli oläslig. Detta beror på hur parquet och JSON hanterar tomma fält på olika sätt.
Låt oss överväga en typisk situation. Till exempel, igår kom JSON:
Dag 1: {"a": {"b": 1}},
 och idag ser samma JSON ut så här:
Dag 2: {"a": null}
 Anta att vi har två olika partitioner, där varje partition innehåller en rad.
 När vi läser in de ursprungliga data fullständigt, kommer Spark att kunna avgöra typen och förstå att "a" är ett fält av typ "struktur" med ett underliggande fält "b" av typ INT. Men om varje partition har sparats separat, så blir resultatet en parquet med inkompatibla partitioneringsscheman:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
 Denna situation är välkänd, därför har det lagts till en speciell option – att ta bort tomma fält vid parsering av de ursprungliga data:
df = spark.read.json("...", dropFieldIfAllNull=True)
 I så fall kommer parquet att bestå av partitioner som kan läsas tillsammans. 
 Men de som har gjort detta i praktiken kommer att dra på munnen. Varför? För det kommer sannolikt att dyka upp ytterligare två situationer. Eller tre. Eller fyra. Den första, som nästan garanterat kommer att inträffa, är att numeriska typer kan se olika ut i olika JSON-filer. Till exempel, {intField: 1} och {intField: 1.1}. Om sådana fält hamnar i samma partition, kommer sammanslagningen av scheman att läsa allt korrekt och ge den mest exakta typen. Men om de befinner sig i olika partitioner, kommer den ena att ha intField: int, medan den andra får intField: double.
För att hantera denna situation finns följande flagga:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
 Nu har vi en mapp där partitionerna ligger, som kan läsas in i en enhetlig dataram och en giltig parquet för hela vitrin. Eller?
Vi måste komma ihåg att vi registrerade tabellen i Hive. Hive är inte skiftlägeskänsligt när det gäller fältnamn, medan parquet är det. Därför är partitioner med scheman: field1: int, och Field1: int identiska för Hive, men inte för Spark. Kom ihåg att konvertera fältnamn till gemener.
Efter detta verkar allt bra.
Men, det är inte så enkelt. Det uppstår ett andra, också välkänt problem. Eftersom varje ny partition sparas separat, kommer det att finnas systemfiler för Spark i partitionsmappen, till exempel flaggan för operationens framgång _SUCCESS. Detta kommer att leda till ett fel vid försök att läsa parquet. För att undvika detta måste vi konfigurera så att Spark inte lägger till systemfiler i mappen:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
 Det verkar som att en ny parquet-partition med parsade data för dagen nu läggs till i målmappen varje dag. Vi har redan tänkt på att undvika partitioner med datatypkonflikter.
Men vi står nu inför det tredje problemet. Den övergripande schemat är okänd, dessutom har Hive-tabellen en felaktig schema eftersom varje ny partition förmodligen har orsakat en förvrängning i schemat.
Vi behöver registrera om tabellen. Detta kan göras enkelt: läs om parquet-vitrinen, ta schemat och skapa en DDL baserat på det, för att registrera mappen i Hive som en extern tabell och uppdatera målvinets schema.
Det fjärde problemet uppstår. När vi registrerade tabellen första gången, baserade vi oss på Spark. Nu gör vi det själva, och vi måste komma ihåg att parquet-fält kan börja med ogiltiga tecken för Hive. Till exempel, Spark utesluter strängar som det inte kunde parsa i fältet „corrupt_record”. Ett sådant fält kan inte registreras i Hive utan att maskera.
Med detta i åtanke får vi schemat:
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)
 Kod ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace("array<`", "array<") skapar en säker DDL, det vill säga istället för:
create table tname (_field1 string, 1field string)
 Med sådana fältnamn som „_field1, 1field” skapas en säker DDL där fältnamn maskeras: create table `tname` (`_field1` string, `1field` string).
Frågan uppstår: hur får man korrekt en dataframe med fullständigt schema (i koden pf)? Hur får man denna pf? Det är det femte problemet. Ska man läsa schemat för alla partitioner från mappen med parquet-filer i målvitrinen? Det är den säkraste metoden, men den är tung.
Schemat finns redan i Hive. För att få en ny schemaläggning kan man kombinera schemat för hela tabellen med den nya partitionen. Det innebär att man behöver ta schema från Hive och kombinera det med schemat för den nya partitionen. Detta kan göras genom att läsa in metadata från Hive, spara dem i en tillfällig mapp och läsa in båda partitionerna samtidigt med Spark.
I grund och botten har vi allt som behövs: den ursprungliga tabellens schema i Hive och den nya partitionen. Vi har också data. Det kvarstår bara att få det nya schemat, där schemat för de presenterade och de nya fälten från den skapade partitionen kombineras:
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/*")
 Nästa steg är att skapa DDL-registreringen av tabellen, precis som i det föregående fragmentet.
 Om hela kedjan fungerar korrekt, det vill säga en initial laddning har gjorts och tabellen i Hive är korrekt skapad, får vi den uppdaterade tabellens schema.
Och slutligen, problemet är att man inte så enkelt kan lägga till en partition i en Hive-tabell eftersom den kommer att bli trasig. Det är nödvändigt att få Hive att rätta till sin partitioneringsstruktur:
from pyspark.sql import HiveContext
hc = HiveContext(spark) 
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
 En enkel uppgift att läsa JSON och skapa en presentation av det leder till att man måste övervinna flera otydliga svårigheter, vars lösningar får sökas separat. Även om dessa lösningar är enkla tar det lång tid att hitta dem.
För att genomföra uppbyggnaden av presentationen var man tvungen att:
 
Lägga till partitioner i presentationen, utan de interna filerna
 
Förstå de tomma fälten i de ursprungliga data som Spark typade
 
Konvertera enkla typer till strängar
 
Tidigare fältnamn till gemener
 
Dela upp dataladdningen och registreringen av tabellen i Hive (skapa DDL)
 
Inte glömma att maskera fältnamn som kan vara oförenliga med Hive
 
Lära sig att uppdatera registreringen av tabellen i Hive
 Sammanfattningsvis är det värt att notera att lösningen för att bygga presentationer döljer många fallgropar. Därför är det bättre att vända sig till en erfaren partner med framgångsrik expertis om man stöter på svårigheter vid genomförandet.
Tack för att du läste denna artikel, vi hoppas att informationen är användbar.
 
Källa: habr.com
   
 Vi använder cookies cookie, för att ge den bästa upplevelsen på vår webbplats. Genom att fortsätta använda webbplatsen godkänner du vår integritetspolicy. Acceptera
 Köp pålitlig webbhotell med DDoS-skydd, VPS VDS-servrar 🔥 Köp pålitlig webbhotell med DDoS-skydd, VPS VDS-servrar | ProHoster