🥇Immersione nel Delta Lake: applicazione forzata e evoluzione dello schema

Ciao, Habr! Vi presento la traduzione dell'articolo «Diving Into Delta Lake: Schema Enforcement & Evolution» degli autori Burak Yavuz, Brenner Heintz e Denny Lee, preparato in vista dell'inizio del corso «Data Engineer» di OTUS.

I dati, così come la nostra esperienza, si accumulano e si evolvono costantemente. Per non rimanere indietro, i nostri modelli mentali del mondo devono adattarsi ai nuovi dati, alcuni dei quali contengono nuove dimensioni — nuovi modi di osservare le cose di cui non avevamo mai avuto idea prima. Questi modelli mentali non differiscono molto dagli schemi delle tabelle, che definiscono come classifichiamo e trattiamo le nuove informazioni.

Questo ci porta alla questione della gestione degli schemi. Man mano che le esigenze e i requisiti aziendali cambiano nel tempo, cambia anche la struttura dei tuoi dati. Delta Lake consente di implementare facilmente nuove dimensioni quando i dati cambiano. Gli utenti hanno accesso a una semantica semplice per gestire gli schemi delle proprie tabelle. Questi strumenti includono l'applicazione forzata dello schema (Schema Enforcement), che protegge gli utenti dall'inserimento involontario di errore o dati non necessari nelle proprie tabelle, e l'evoluzione dello schema (Schema Evolution), che consente di aggiungere automaticamente nuove colonne con dati preziosi nei luoghi appropriati. In questo articolo, approfondiremo l'uso di questi strumenti.

Comprensione degli schemi delle tabelle

Ogni DataFrame in Apache Spark contiene uno schema che definisce la forma dei dati, come tipi di dati, colonne e metadati. Con Delta Lake, lo schema della tabella è salvato in formato JSON all'interno del log delle transazioni.

Cos'è l'applicazione forzata dello schema?

L'applicazione forzata dello schema (Schema Enforcement), nota anche come validazione dello schema (Schema Validation), è un meccanismo di protezione in Delta Lake che garantisce la qualità dei dati, rifiutando le registrazioni che non corrispondono allo schema della tabella. Proprio come un hostess alla reception di un ristorante popolare che accetta solo con prenotazione, verifica che ogni colonna di dati inserita nella tabella sia presente nell'elenco delle colonne previste (in altre parole, che ci sia una “prenotazione” per ciascuna di esse) e rifiuta qualsiasi registrazione con colonne non presenti nell'elenco.

Come funziona l'applicazione forzata dello schema?

Delta Lake utilizza il controllo dello schema durante la scrittura, il che significa che tutte le nuove registrazioni nella tabella vengono verificate per compatibilità con lo schema della tabella di destinazione al momento della scrittura. Se lo schema non è compatibile, Delta Lake annulla completamente la transazione (i dati non vengono scritti) e genera un'eccezione per informare l'utente dell'incongruenza.
Per determinare la compatibilità di una registrazione con la tabella, Delta Lake utilizza le seguenti regole. DataFrame da registrare:

non può contenere colonne aggiuntive non presenti nello schema della tabella di destinazione. Al contrario, va bene se i dati in ingresso non contengono tutte le colonne della tabella: a queste colonne verranno semplicemente assegnati valori nulli.
non può avere tipi di dati delle colonne diversi dai tipi di dati delle colonne nella tabella di destinazione. Se una colonna della tabella di destinazione contiene dati di tipo StringType, ma la colonna corrispondente nel DataFrame contiene dati di tipo IntegerType, l'applicazione forzata dello schema genererà un'eccezione e impedirà l'esecuzione dell'operazione di scrittura.
non può contenere nomi di colonne che differiscono solo per maiuscole e minuscole. Ciò significa che non puoi avere colonne con nomi ‘Foo’ e ‘foo’ definite nella stessa tabella. Anche se Spark può essere utilizzato in modalità sensibile o insensibile (di default) al maiuscolo, Delta Lake conserva il maiuscolo, ma è insensibile nell'ambito della memorizzazione dello schema. Parquet è sensibile al maiuscolo quando memorizza e restituisce informazioni sulla colonna. Per evitare possibili errori, danni ai dati o perdite (con cui ci siamo trovati personalmente in Databricks), abbiamo deciso di aggiungere questa restrizione.

Per illustrare ciò, diamo un'occhiata a cosa accade nel codice sottostante quando si tenta di aggiungere alcune colonne appena generate a una tabella Delta Lake che non è ancora configurata per accettarle.

# Сгенерируем DataFrame ссуд, который мы добавим в нашу таблицу Delta Lake
loans = sql("""
            SELECT addr_state, CAST(rand(10)*count as bigint) AS count,
            CAST(rand(10) * 10000 * count AS double) AS amount
            FROM loan_by_state_delta
            """)

# Вывести исходную схему DataFrame
original_loans.printSchema()

root
  |-- addr_state: string (nullable = true)
  |-- count: integer (nullable = true)
 
# Вывести новую схему DataFrame
loans.printSchema()
 
root
  |-- addr_state: string (nullable = true)
  |-- count: integer (nullable = true)
  |-- amount: double (nullable = true) # new column
 
# Попытка добавить новый DataFrame (с новым столбцом) в существующую таблицу
loans.write.format("delta") 
           .mode("append") 
           .save(DELTALAKE_PATH)

Returns:

A schema mismatch detected when writing to the Delta table.
 
To enable schema migration, please set:
'.option("mergeSchema", "true")'
 
Table schema:
root
-- addr_state: string (nullable = true)
-- count: long (nullable = true)
 
Data schema:
root
-- addr_state: string (nullable = true)
-- count: long (nullable = true)
-- amount: double (nullable = true)
 
If Table ACLs are enabled, these options will be ignored. Please use the ALTER TABLE command for changing the schema.

Invece di aggiungere automaticamente nuove colonne, Delta Lake applica lo schema e interrompe la scrittura. Per aiutare a identificare quale colonna (o più colonne) sta causando l'incongruenza, Spark stampa entrambi gli schemi dallo stack trace per il confronto.

Qual è il vantaggio dell'applicazione forzata dello schema?

Poiché l'applicazione forzata dello schema rappresenta un controllo piuttosto rigoroso, è uno strumento eccellente da utilizzare come portinaio per un set di dati puliti e completamente trasformati, pronto per la produzione o il consumo. In genere, viene applicato a tabelle che forniscono direttamente i dati:

Algoritmi di apprendimento automatico
Dashboard BI
Analisi dei dati e strumenti di visualizzazione
Qualsiasi sistema di produzione che richieda schemi semantici rigorosamente strutturati e tipizzati.

Per preparare i propri dati a questa barriera finale, molti utenti utilizzano una semplice architettura “multi-hop”, che introduce gradualmente struttura nelle loro tabelle. Per saperne di più, puoi consultare l'articolo Machine Learning di livello produttivo con Delta Lake.

Certo, l'applicazione forzata dello schema può essere utilizzata in qualsiasi parte del tuo pipeline, ma ricorda che la registrazione in streaming in tabella può essere frustrante se, ad esempio, hai dimenticato di aver aggiunto un'altra colonna ai dati in ingresso.

Prevenzione della diluizione dei dati

A questo punto potresti chiederti, perché tanto clamore? Dopotutto, a volte un errore inaspettato di "incompatibilità dello schema" può intralciare il tuo flusso di lavoro, specialmente se sei un principiante in Delta Lake. Perché non permettere semplicemente allo schema di cambiare in modo che io possa registrare il mio DataFrame, a prescindere da tutto?

Come dice un vecchio detto, “un’oncia di prevenzione vale una libbra di cura”. A un certo punto, se non ti prendi cura dell'applicazione del tuo schema, sorgono problemi di compatibilità dei tipi di dati — a prima vista, fonti di dati grezzi omogenee possono contenere casi limite, colonne danneggiate, mappature mal formate o altre cose spaventose che ti assillano nei sogni. Il miglior approccio è fermare questi nemici alle porte — tramite l'applicazione forzata dello schema — e affrontarli alla luce del sole, piuttosto che più tardi, quando iniziano a gironzolare nelle oscure profondità del tuo codice di lavoro.

L'applicazione forzata dello schema assicura che la struttura della tua tabella non cambi, a meno che tu non approvi un'eventuale modifica. Questo previene la "diluzione" dei dati, che può avvenire quando nuove colonne vengono aggiunte così frequentemente che le tabelle compresse e preziose perdono il loro valore e utilità a causa dell'inondazione dei dati. Incoraggiandoti a essere deliberato, impostare standard elevati e aspettarti alta qualità, l'applicazione forzata dello schema fa esattamente ciò per cui è stata concepita: aiutarti a mantenere l'integrità e le tue tabelle a essere pulite.

Se, dopo un ulteriore esame, decidi che hai realmente bisogno di aggiungere una nuova colonna, nessun problema, di seguito è riportata una soluzione in una riga. La soluzione è l'evoluzione dello schema!

Cos'è l'evoluzione dello schema?

L'evoluzione dello schema è una funzionalità che consente agli utenti di modificare facilmente l'attuale schema della tabella in base ai dati che cambiano nel tempo. Viene utilizzata principalmente durante l'operazione di aggiunta o sovrascrittura per adattare automaticamente lo schema includendo una o più nuove colonne.

Come funziona l'evoluzione dello schema?

Seguendo l'esempio del paragrafo precedente, gli sviluppatori possono facilmente utilizzare l'evoluzione dello schema per aggiungere nuove colonne che erano precedentemente rifiutate a causa di incompatibilità con lo schema. L'evoluzione dello schema viene attivata aggiungendo .option('mergeSchema', 'true') al tuo comando Spark .write o .writeStream.

# Добавьте параметр mergeSchema
loans.write.format("delta") 
           .option("mergeSchema", "true") 
           .mode("append") 
           .save(DELTALAKE_SILVER_PATH)

Per visualizzare il grafico, esegui la seguente query Spark SQL

# Создайте график с новым столбцом, чтобы подтвердить, что запись прошла успешно
%sql
SELECT addr_state, sum(`amount`) AS amount
FROM loan_by_state_delta
GROUP BY addr_state
ORDER BY sum(`amount`)
DESC LIMIT 10

In alternativa, puoi impostare questa opzione per tutta la sessione Spark, aggiungendo spark.databricks.delta.schema.autoMerge = True nella configurazione di Spark. Ma usala con cautela, poiché l'applicazione forzata dello schema non ti avviserà più di eventuali incompatibilità non intenzionali.

Attivando il parametro mergeSchema, tutte le colonne presenti nel DataFrame ma assenti nella tabella di destinazione vengono automaticamente aggiunte alla fine dello schema all'interno della transazione di scrittura. Possono essere aggiunti anche campi nidificati, e questi saranno anch'essi aggiunti alla fine delle colonne corrispondenti della struttura.

Data ingegneri e scienziati possono utilizzare questa opzione per aggiungere nuove colonne (una metrica recentemente monitorata o una colonna delle vendite di questo mese) alle loro esistenti tabelle di produzione di apprendimento automatico, senza compromettere i modelli esistenti basati sulle vecchie colonne.

I seguenti tipi di modifiche allo schema sono consentiti nell'ambito dell'evoluzione dello schema durante l'aggiunta o la riscrittura di una tabella:

Aggiunta di nuove colonne (questo è lo scenario più comune)
Modifica dei tipi di dati da NullType -> qualsiasi altro tipo o upgrade da ByteType -> ShortType -> IntegerType

Altre modifiche, non consentite nell'ambito dell'evoluzione dello schema, richiedono che schema e dati siano riscritti attraverso l'aggiunta .option("overwriteSchema", "true"). Ad esempio, nel caso in cui la colonna "Foo" fosse inizialmente un intero e il nuovo schema fosse di tipo stringa, allora tutti i file Parquet (dati) dovrebbero essere riscritti. Queste modifiche includono:

rimozione di una colonna
modifica del tipo di dati di una colonna esistente (in loco)
rinominare colonne che differiscono solo per maiuscole e minuscole (ad esempio "Foo" e "foo")

Infine, con la prossima release di Spark 3.0, sarà supportato completamente DDL esplicito (utilizzando ALTER TABLE), consentendo agli utenti di eseguire le seguenti operazioni sugli schemi delle tabelle:

aggiunta di colonne
modifica dei commenti sulle colonne
configurazione delle proprietà della tabella che definiscono il comportamento della tabella, ad esempio impostare la durata della conservazione del registro delle transazioni.

Qual è il vantaggio dell'evoluzione dello schema?

L'evoluzione dello schema può essere utilizzata ogni volta che si intende modificare lo schema della propria tabella (in contrapposizione ai casi in cui si è accidentalmente aggiunto al proprio DataFrame colonne che non dovrebbero esserci). Questo è il modo più semplice per migrare il proprio schema, poiché aggiunge automaticamente i nomi delle colonne e i tipi di dati corretti senza la necessità di dichiararli esplicitamente.

Conclusione

L'applicazione forzata dello schema rifiuta qualsiasi nuova colonna o altra modifica dello schema che non sia compatibile con la tua tabella. Stabilendo e mantenendo questi elevati standard, gli analisti e gli ingegneri possono contare sul fatto che i loro dati hanno il massimo livello di integrità, ragionando su di esso in modo chiaro e chiaro, il che consente loro di prendere decisioni aziendali più efficaci.

D'altra parte, l'evoluzione dello schema si aggiunge all'applicazione forzata, semplificando le modifiche automatiche dello schema. Alla fine, non dovrebbe essere una complessità - aggiungere una colonna.

L'applicazione forzata dello schema è lo yang, mentre l'evoluzione dello schema è lo yin. Quando utilizzate insieme, queste funzioni semplificano più che mai la soppressione del rumore e la regolazione del segnale.

Vorremmo anche ringraziare Mukul Murty e Pranav Anand per il loro contributo a questo articolo.

Altri articoli di questa serie:

Immersione nel Delta Lake: decodificare il log delle transazioni