Gentili lettori, buon giorno!
In questo articolo, il principale consulente della divisione Big Data Solutions di "Neoflex" descrive dettagliatamente le opzioni per la costruzione di vetrine a struttura variabile utilizzando Apache Spark.
Nell'ambito di un progetto di analisi dei dati, spesso si presenta la necessità di costruire vetrine basate su dati poco strutturati.
Di solito si tratta di log o delle risposte di vari sistemi, salvati in formato JSON o XML. I dati vengono scaricati in Hadoop, da cui è necessario costruire una vetrina. Possiamo organizzare l'accesso alla vetrina creata, ad esempio, tramite Impala.
In questo caso, lo schema della vetrina target è preliminarmente sconosciuto. Inoltre, lo schema non può nemmeno essere redatto a priori, poiché dipende dai dati, e noi abbiamo a che fare con questi dati poco strutturati.
Ad esempio, oggi viene registrata la seguente risposta:
{source: "app1", error_code: ""}e domani da questo stesso sistema arriva la seguente risposta:
{source: "app1", error_code: "error", description: "Errore di rete"}Di conseguenza, nella vetrina dovrebbe aggiungersi un ulteriore campo — description, e se arriverà o no, nessuno lo sa.
Il compito di creare una vetrina su tali dati è piuttosto standard, e Spark ha a disposizione diversi strumenti per questo. Per il parsing dei dati sorgente, è supportato sia JSON che XML, e per schemi non noti a priori è prevista la supporto dell'evoluzione dello schema (schemaEvolution).
A prima vista, la soluzione sembra semplice. Bisogna prendere la cartella con i file JSON e leggerli in un dataframe. Spark genererà uno schema e trasformerà i dati annidati in strutture. Successivamente, è necessario salvare tutto in parquet, che è supportato anche in Impala, registrando la vetrina nel metastore di Hive.
Sembra tutto semplice.
Tuttavia, dagli esempi brevi nella documentazione non è chiaro come affrontare vari problemi nella pratica.
La documentazione descrive un approccio non per la creazione di vetrine, ma per la lettura di JSON o XML in dataframe.
In particolare, viene semplicemente mostrato come leggere e analizzare un JSON:
df = spark.read.json(path...)Questo è sufficiente per rendere i dati accessibili a Spark.
In pratica, lo scenario è molto più complesso che semplicemente leggere file JSON da una cartella e creare un dataframe. La situazione è la seguente: c'è già una vetrina definita, ogni giorno arrivano nuovi dati, e questi devono essere aggiunti alla vetrina, tenendo presente che lo schema potrebbe differire.
Uno schema comune per la costruzione di vetrine è il seguente:
Passo 1. I dati vengono caricati in Hadoop con successivi caricamenti giornalieri e vengono collocati in una nuova partizione. Si ottiene una cartella partizionata per giorni con i dati sorgente.
Passo 2. Durante il caricamento iniziale, questa cartella viene letta e analizzata tramite i mezzi di Spark. Il dataframe ottenuto viene salvato in un formato analizzabile, ad esempio in parquet, che può essere poi importato in Impala. Si crea così una vetrina obiettivo con tutti i dati accumulati fino a quel momento.
Passo 3. Si crea un caricamento che aggiornerà la vetrina ogni giorno.
Sorge la questione del caricamento incrementale, della necessità di partizionare la vetrina e della questione del supporto a uno schema generale della vetrina.
Facciamo un esempio. Supponiamo che sia stato implementato il primo passo nella costruzione dell'archiviazione e sia stata configurata l'esportazione di file JSON in una cartella.
Creare da essi un dataframe per poi salvare come vetrina non comporta problemi. Questo è il primo passo che può essere facilmente trovato nella documentazione di Spark:
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)Sembrerebbe tutto a posto.
Abbiamo letto e analizzato il JSON, quindi salviamo il dataframe come parquet, registrandolo in Hive in qualsiasi modo comodo:
df.write.format(“parquet”).option('path','').saveAsTable('') Otteniamo la vetrina.
Ma, il giorno successivo sono stati aggiunti nuovi dati dalla sorgente. Abbiamo una cartella con JSON e una vetrina, creata sulla base di questa cartella. Dopo il caricamento della successiva porzione di dati dalla sorgente, mancano i dati per un giorno nella vetrina.
Una soluzione logica sarà quella di partizionare la vetrina per giorni, il che permetterà di aggiungere una nuova partizione ogni giorno successivo. Questo meccanismo è ben noto, Spark consente di registrare partizioni separatamente.
Per prima cosa effettuiamo un caricamento iniziale, salvando i dati come descritto sopra, aggiungendo solo la partizione. Questa azione viene chiamata inizializzazione della vetrina e viene eseguita solo una volta:
df.write.partitionBy("date_load").mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable)
Il giorno successivo carichiamo solo la nuova partizione:
df.coalesce(1).write.mode("overwrite").parquet(dbpath + "\/" + db + "\/" + destTable +"\/date_load=" + date_load + "\/" )
Resta solo da registrare nuovamente in Hive per aggiornare lo schema.
Tuttavia, qui sorgono i problemi.
Il primo problema. Prima o poi, il parquet risultante non sarà leggibile. Ciò è dovuto al modo in cui parquet e JSON gestiscono i campi vuoti in modo diverso.
Consideriamo una situazione tipica. Ad esempio, ieri riceviamo JSON:
Giorno 1: {"a": {"b": 1}},
e oggi lo stesso JSON appare così:
Giorno 2: {"a": null}
Supponiamo di avere due partizioni diverse, ciascuna con una riga.
Quando leggiamo i dati sorgente in modo completo, Spark sarà in grado di determinare il tipo e capirà che "a" è un campo di tipo "struttura", con un campo annidato "b" di tipo INT. Ma, se ogni partizione è stata salvata separatamente, si ottiene un parquet con schemi di partizione incompatibili:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Questa situazione è ben nota, quindi è stata aggiunta appositamente un'opzione: durante il parsing dei dati sorgente, rimuovere i campi vuoti:
df = spark.read.json("...", dropFieldIfAllNull=True)
In questo caso, il parquet sarà composto da partizioni che possono essere lette insieme.
Tuttavia, coloro che lo hanno fatto nella pratica sorrideranno amaramente. Perché? Perché molto probabilmente si presenteranno altre due situazioni. O tre. O quattro. La prima, che si presenterà quasi sicuramente, è che i tipi numerici appariranno in modo diverso nei vari file JSON. Ad esempio, {intField: 1} e {intField: 1.1}. Se tali campi si trovano nella stessa partizione, lo schema di merge li leggerà tutti correttamente, portandoli al tipo più preciso. Ma se sono in partizioni diverse, allora in una avremo intField: int e nell'altra intField: double.
Per gestire questa situazione c'è il seguente flag:
df = spark.read.json("...", dropFieldIfAllNull=True, primitivesAsString=True)
Ora abbiamo una cartella in cui si trovano le partizioni che possono essere lette in un unico dataframe e un parquet valido per l'intera vista. Giusto? No.
Devi ricordare che abbiamo registrato la tabella in Hive. Hive non distingue maiuscole e minuscole nei nomi dei campi, mentre parquet sì. Quindi le partizioni con schemi: field1: int e Field1: int sono identiche per Hive, ma non per Spark. Non dobbiamo dimenticare di convertire i nomi dei campi in minuscolo.
Dopo di ciò, sembra che vada tutto bene.
Tuttavia, non tutto è così semplice. Si presenta un secondo problema, anch'esso ben noto. Poiché ogni nuova partizione viene salvata separatamente, nella cartella della partizione si trovano file di servizio di Spark, come per esempio il flag di successo dell'operazione _SUCCESS. Questo porterà a un errore durante il tentativo di leggere il parquet. Per evitarlo, è necessario configurare l'impostazione in modo da vietare a Spark di scrivere file di servizio nella cartella:
hadoopConf = sc._jsc.hadoopConfiguration()
hadoopConf.set("parquet.enable.summary-metadata", "false")
hadoopConf.set("mapreduce.fileoutputcommitter.marksuccessfuljobs", "false")
Sembra che ora ogni giorno venga aggiunta una nuova partizione parquet alla cartella della vetrina di destinazione, dove si trovano i dati parse per il giorno. Ci siamo preoccupati in anticipo di evitare partizioni con conflitti di tipi di dati.
Ma davanti a noi c'è un terzo problema. Ora la schema generale non è nota, inoltre, nella tabella Hive è presente uno schema errato, poiché ogni nuova partizione ha probabilmente introdotto delle distorsioni nello schema.
È necessario registrare nuovamente la tabella. Questo può essere fatto semplicemente: leggere di nuovo la vetrina parquet, prendere lo schema e crearne un DDL, con il quale registrare nuovamente la cartella in Hive come tabella esterna, aggiornando lo schema della vetrina di destinazione.
Si presenta così un quarto problema. Quando abbiamo registrato la tabella per la prima volta, ci siamo basati su Spark. Ora lo facciamo noi stessi e dobbiamo ricordare che i campi parquet possono iniziare con caratteri non consentiti per Hive. Ad esempio, Spark scarta le righe che non è riuscito a decodificare nel campo «corrupt_record». Tale campo non potrà essere registrato in Hive senza eseguire l'escape.
Sapendo questo, otteniamo lo schema:
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)
Codice ("_corrupt_record", "`_corrupt_record`") + " " + f[1].replace(":", "`:").replace("<", "<`").replace(",", ",`").replace(«array<`», «array<» genera un DDL sicuro, cioè invece di:
create table tname (_field1 string, 1field string)
Con nomi di campi come «_field1, 1field», viene generato un DDL sicuro, dove i nomi dei campi sono escutati: create table `tname` (`_field1` string, `1field` string).
Sorge la questione: come ottenere correttamente un dataframe con lo schema completo (nel codice pf)? Come ottenere questo pf? Questo è il quinto problema. Rileggere lo schema di tutte le partizioni dalla cartella con i file parquet della vetrina di destinazione? Questo metodo è il più sicuro, ma anche il più pesante.
Lo schema è già presente in Hive. È possibile ottenere un nuovo schema combinando lo schema dell'intera tabella con la nuova partizione. Quindi, è necessario prendere lo schema della tabella da Hive e unirlo con lo schema della nuova partizione. Questo può essere fatto leggendo i metadati di prova da Hive, salvandoli in una cartella temporanea e leggendo entrambe le partizioni contemporaneamente con Spark.
In sostanza, abbiamo tutto il necessario: lo schema originale della tabella in Hive e la nuova partizione. Abbiamo anche i dati. Rimane solo da ottenere il nuovo schema, in cui viene combinato lo schema della vetrina e i nuovi campi creati dalla partizione:
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/*")
Successivamente, creiamo il DDL per la registrazione della tabella, come nel frammento precedente.
Se tutta la catena funziona correttamente, vale a dire — è stata effettuata un'inizializzazione e la tabella è stata creata correttamente in Hive, allora otteniamo lo schema aggiornato della tabella.
E infine, il problema è che non è così semplice aggiungere una partizione alla tabella di Hive, poiché essa risulterebbe rotta. È necessario fare in modo che Hive ripari la propria struttura delle partizioni:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Un compito semplice di lettura di JSON e creazione di una vetrina sulla sua base si trasforma nel superamento di una serie di difficoltà implicite, le cui soluzioni devono essere cercate separatamente. E sebbene queste soluzioni siano semplici, ci vuole molto tempo per trovarle.
Per implementare la costruzione della vetrina, è stato necessario:
- Aggiungere partizioni alla vetrina, liberandosi dei file di servizio
- Occuparsi dei campi vuoti nei dati originali, che Spark ha tipizzato
- Convertire i tipi semplici in stringhe
- Convertire i nomi dei campi in minuscolo
- Separare l'esportazione dei dati dalla registrazione della tabella in Hive (creazione DDL)
- Non dimenticare di eseguire l'escape dei nomi dei campi che potrebbero essere incompatibili con Hive
- Imparare ad aggiornare la registrazione della tabella in Hive
In conclusione, va notato che la soluzione per la costruzione delle vetrine presenta molte insidie. Pertanto, in caso di difficoltà nella realizzazione, è meglio consultare un partner esperto con esperienza di successo.
Grazie per aver letto questo articolo, speriamo che le informazioni siano utili.
Fonte: habr.com
Acquista hosting affidabile per siti web con protezione DDoS, VPS VDS server 🔥 Acquista hosting affidabile per siti web con protezione DDoS, VPS VDS server | ProHoster
