Gentili lettori, buon pomeriggio!
In questo articolo, il principale consulente della divisione Big Data Solutions di 'Neoflex' descrive in dettaglio le opzioni per costruire vetrine a struttura variabile utilizzando Apache Spark.
Nel contesto di un progetto di analisi dei dati, si presenta spesso la necessità di costruire vetrine basate su dati scarsamente strutturati.
Di solito si tratta di log o risposte di diversi sistemi, salvate in formato JSON o XML. I dati vengono esportati in Hadoop, quindi è necessario costruire una vetrina. Possiamo organizzare l'accesso alla vetrina creata, ad esempio, attraverso Impala.
In questo caso, lo schema della vetrina target non è conosciuto in anticipo. Inoltre, lo schema non può nemmeno essere definito in anticipo, poiché dipende dai dati e stiamo trattando proprio questi dati scarsamente strutturati.
Ad esempio, oggi viene registrata la seguente risposta:
{source: "app1", error_code: ""}e domani dalla stessa sistema arriva questa risposta:
{source: "app1", error_code: "error", description: "Errore di rete"}Di conseguenza, nella vetrina dovrebbe essere aggiunto un ulteriore campo — description, e se verrà o meno, nessuno può saperlo.
Il compito di creare una vetrina basata su questi dati è piuttosto comune, e Spark offre diversi strumenti per questo. Per il parsing dei dati sorgente è supportato sia JSON che XML, e per schemi sconosciuti in anticipo è prevista la funzionalità di schemaEvolution.
A prima vista, la soluzione sembra semplice. Bisogna prendere la cartella contenente JSON e leggerla in un dataframe. Spark creerà uno schema e trasformerà i dati annidati in strutture. Successivamente, bisogna salvare tutto in formato parquet, che è supportato anche da Impala, registrando la vetrina nel metastore di Hive.
In apparenza, è tutto semplice.
Tuttavia, dai brevi esempi nella documentazione non è chiaro cosa fare con alcune problematiche in pratica.
Nella documentazione si descrive un approccio non per la creazione di una vetrina, ma per la lettura di JSON o XML in un 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.
Nella pratica, lo scenario è molto più complesso che semplicemente leggere file JSON da una cartella e creare un dataframe. La situazione è la seguente: esiste già una vetrina definita, ogni giorno arrivano nuovi dati, e devono essere aggiunti alla vetrina, tenendo presente che lo schema potrebbe variare.
Lo schema standard per la creazione della vetrina è il seguente:
Passo 1. I dati vengono caricati in Hadoop con caricamenti giornalieri successivi e vengono salvati in una nuova partizione. Si ottiene una cartella partizionata giornalmente con i dati sorgente.
Passo 2. Durante il caricamento iniziale, questa cartella viene letta e analizzata tramite Spark. Il dataframe ottenuto viene salvato in un formato accessibile per l'analisi, ad esempio in parquet, che può poi essere importato in Impala. Così si crea la vetrina obiettivo con tutti i dati accumulati fino a quel momento.
Passo 3. Viene creata una routine di caricamento che aggiornerà quotidianamente la vetrina.
Sorge la questione del caricamento incrementale, della necessità di partizionare la vetrina e della questione del supporto dello schema generale della vetrina.
Facciamo un esempio. Supponiamo di aver implementato il primo passo per la costruzione del magazzino e di aver configurato l'esportazione dei file JSON in una cartella.
Creare da esse un dataframe, per poi salvarlo come vetrina, non presenta problemi. Questo è il primo passo che è facile trovare 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)Sembra che tutto vada bene.
Abbiamo letto e analizzato il JSON, quindi salviamo il dataframe come parquet, registrando in Hive in qualsiasi modo conveniente:
df.write.format("parquet").option('path','').saveAsTable('') Creiamo la vetrina.
Tuttavia, il giorno dopo sono stati aggiunti nuovi dati dalla fonte. Abbiamo una cartella con JSON e una vetrina creata sulla base di quella cartella. Dopo il caricamento del successivo lotto di dati dalla fonte, mancano dati per un giorno nella vetrina.
Una soluzione logica sarebbe partizionare la vetrina per giorni, consentendo di aggiungere una nuova partizione ogni giorno successivo. Questo meccanismo è ben conosciuto e Spark consente di registrare le partizioni separatamente.
Per prima cosa facciamo un caricamento iniziale, salvando i dati come descritto sopra, aggiungendo solo la partizione. Questa azione è 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 + "/")
Rimane solo da registrare nuovamente in Hive per aggiornare lo schema.
Tuttavia, qui sorgono problemi.
Il primo problema. Prima o poi, il parquet risultante non sarà leggibile. Questo è dovuto al diverso approccio di parquet e JSON ai campi vuoti.
Consideriamo una situazione tipica. Ad esempio, ieri abbiamo ricevuto un JSON:
Giorno 1: {"a": {"b": 1}},
e oggi questo stesso JSON appare così:
Giorno 2: {"a": null}
Supponiamo di avere due partizioni diverse, ciascuna con una riga.
Quando leggiamo i dati originali per intero, Spark riesce a determinare il tipo e comprende che «a» è un campo di tipo «struttura», con un campo annidato «b» di tipo INT. Tuttavia, se ogni partizione è stata salvata separatamente, si ottiene un parquet con schemi di partizioni incompatibili:
df1 (a: <struct>)
df2 (a: STRING NULLABLE)
Questa situazione è ben nota, quindi è stata appositamente aggiunta un'opzione: durante il parsing dei dati sorgente, eliminare i campi vuoti:
df = spark.read.json("...", dropFieldIfAllNull=True)
In questo caso, il parquet consisterà di partizioni leggibili insieme.
Tuttavia, coloro che l'hanno fatto nella pratica sorrideranno amaramente. Perché? È probabile che si verifichino altre due situazioni. O tre. O quattro. La prima, che quasi sicuramente si presenterà, è che i tipi numerici appariranno in modo diverso nei diversi file JSON. Ad esempio, {intField: 1} e {intField: 1.1}. Se tali campi si trovano in una sola partizione, lo schema di fusione li leggerà correttamente, portando al tipo più preciso. Ma se si trovano in partizioni diverse, allora in una ci sarà 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 contenente le partizioni che possono essere lette in un unico dataframe e un parquet valido per l'intero catalogo. Giusto? No.
Ricordiamo che abbiamo registrato la tabella in Hive. Hive non fa distinzione tra maiuscole e minuscole nei nomi dei campi, mentre parquet è sensibile al maiuscolo. Pertanto, le partizioni con schemi: field1: int e Field1: int sono uguali per Hive, ma non per Spark. Non dobbiamo dimenticare di convertire i nomi dei campi in minuscolo.
Dopo ciò, sembra che tutto vada bene.
Tuttavia, non è tutto così semplice. Sorge un secondo problema, anch'esso ben noto. Poiché ogni nuova partizione viene salvata separatamente, nella cartella della partizione si troveranno i file di sistema di Spark, come il flag di successo dell'operazione _SUCCESS. Questo porterà a un errore nel tentativo di utilizzare parquet. Per evitare ciò, è necessario configurare le impostazioni in modo da impedire a Spark di scrivere file di sistema 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 del data mart, contenente i dati analizzati della giornata. Ci siamo già preoccupati di evitare partizioni con conflitti di tipi di dati.
Ma ora abbiamo un terzo problema. Ora il modello complessivo non è noto, inoltre, nella tabella Hive il modello è errato, poiché ogni nuova partizione ha probabilmente distorto il modello.
È necessario registrare nuovamente la tabella. Questo può essere fatto facilmente: leggi nuovamente la partizione parquet, prendi il modello e crea un DDL su quella base, per registrare nuovamente la cartella in Hive come tabella esterna, aggiornando il modello del data mart.
Ci troviamo di fronte a un quarto problema. Quando abbiamo registrato la tabella per la prima volta, ci siamo affidati a Spark. Ora lo facciamo noi stessi e dobbiamo ricordare che i campi parquet possono iniziare con caratteri non validi per Hive. Ad esempio, Spark scarta le righe che non riesce a parserizzare nel campo «corrupt_record». Quest'ultimo non potrà essere registrato in Hive senza un'adeguata 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<") fa un DDL sicuro, cioè invece di:
create table tname (_field1 string, 1field string)
Con nomi di campo come «_field1, 1field», viene creato un DDL sicuro, dove i nomi dei campi sono escapati: create table `tname` (`_field1` string, `1field` string).
Sorge: come ottenere correttamente un dataframe con uno schema completo (nel codice pf)? Come posso ottenere questo pf? Questo è il quinto problema. Dovrei rileggere lo schema di tutte le partizioni dalla cartella con i file parquet della vetrina target? Questo è il metodo più sicuro, ma pesante.
Lo schema è già presente in Hive. Possiamo ottenere un nuovo schema unendo lo schema dell'intera tabella e della nuova partizione. Quindi, dobbiamo prendere lo schema della tabella da Hive e unirlo con quello della nuova partizione. Questo può essere fatto leggendo i metadati di test da Hive, salvandoli in una cartella temporanea e leggendo tramite Spark entrambe le partizioni contemporaneamente.
In sostanza, abbiamo tutto il necessario: lo schema di origine della tabella in Hive e la nuova partizione. Abbiamo anche i dati. Rimane solo da ottenere un nuovo schema che unisca lo schema della vetrina e i nuovi campi della partizione creata:
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 l'intera catena funziona correttamente, ovvero c'è stata un'inizializzazione e la tabella è stata creata correttamente in Hive, otteniamo uno schema della tabella aggiornato.
L'ultima sfida è che non è semplice aggiungere una partizione a una tabella Hive, poiché questa potrebbe rompersi. È necessario costringere Hive a riparare la sua struttura delle partizioni:
from pyspark.sql import HiveContext
hc = HiveContext(spark)
hc.sql("MSCK REPAIR TABLE " + db + "." + destTable)
Un compito semplice come leggere JSON e creare una vista da esso diventa un viaggio attraverso una serie di difficoltà implicite, per le quali è necessario trovare soluzioni separatamente. Anche se queste soluzioni sono semplici, la loro ricerca richiede molto tempo.
Per implementare la costruzione della vista, è stato necessario:
- Aggiungere partizioni alla vista, liberandosi dei file di sistema
- Affrontare i campi vuoti nei dati di origine che Spark ha tipizzato
- Convertire i tipi semplici in stringhe
- Portare i nomi dei campi in minuscolo
- Separare l'estrazione 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 sintesi, è importante notare che la creazione di vetrine presenta molte insidie. Pertanto, in caso di difficoltà nell'implementazione, è meglio rivolgersi a un partner esperto con una comprovata 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
