Inserire dati con feed CDC

  • 5 minuti

Quando i sistemi di origine generano milioni di transazioni ogni giorno, il ricaricamento di intere tabelle diventa poco pratico. È necessario un modo per acquisire solo le modifiche, ovvero gli inserimenti, gli aggiornamenti e le eliminazioni, e applicarle in modo efficiente alle tabelle di destinazione. Questo modello, noto come Change Data Capture (CDC), costituisce la base delle moderne pipeline di dati incrementali.

Informazioni sui modelli di inserimento CDC

L'inserimento di Change Data Capture comporta la lettura di un flusso di record di modifiche da un sistema di origine e l'applicazione di tali modifiche a una tabella di destinazione. A differenza dei ricaricamenti completi delle tabelle, CDC elabora solo gli elementi modificati dall'ultima sincronizzazione, riducendo sia il tempo di elaborazione che il consumo di risorse.

Diagramma che illustra i modelli di ingestione CDC.

Un feed CDC tipico contiene record con la struttura seguente:

  • Colonne di dati: valori effettivi per ogni campo nella tabella di origine
  • Tipo di operazione: indica se il record è un'istruzione INSERT, UPDATE o DELETE
  • Colonna sequenza: timestamp o numero di sequenza che determina l'ordine delle modifiche

Si consideri una tabella dei dipendenti che tiene traccia del nome e della posizione. Quando si aggiorna la città di un dipendente, il feed CDC non invia l'intera tabella, ma invia un singolo record con il valore della nuova città, il tipo di operazione UPDATE e un numero di sequenza. La pipeline di inserimento legge questo record e applica la modifica alla riga corrispondente nella tabella di destinazione.

Questo approccio offre diversi vantaggi per i data engineer:

  • Latenza ridotta: il flusso delle modifiche raggiunge la destinazione in pochi minuti anziché in ore
  • Costi inferiori: l'elaborazione di meno record significa meno tempo di calcolo e I/O di archiviazione
  • Impatto minimo sull'origine: la lettura dei log delle modifiche comporta un carico inferiore nei database di produzione rispetto alle analisi complete delle tabelle

Elaborare CDC con l'API AUTO CDC

Azure Databricks fornisce l'API AUTO CDC nelle pipeline Lakeflow per semplificare l'elaborazione dei dati delle modifiche. Questa API gestisce automaticamente la complessità dei record non ordinati, della deduplicazione e della modifica automatica dell'applicazione.

Per implementare l'elaborazione CDC, creare prima una tabella di streaming come destinazione. Si definisce quindi un flusso che legge i record di modifica dall'origine e li applica usando il tipo di dimensione a modifica lenta (SCD) appropriato.

ScD Type 1 mantiene solo la versione corrente di ogni record:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, expr

@dp.temporary_view
def employees_cdf():
    return spark.readStream.table("bronze.employee_changes")

dp.create_streaming_table("employees")

dp.create_auto_cdc_flow(
    target="employees",
    source="employees_cdf",
    keys=["employee_id"],
    sequence_by=col("sequence_num"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "sequence_num"],
    stored_as_scd_type=1
)

Con scD Type 1, quando un dipendente passa da Seattle a Portland, la tabella di destinazione aggiorna direttamente la colonna city. Il record di Seattle scompare, ma rimane solo Portland.

ScD Type 2 mantiene la cronologia completa delle modifiche:

dp.create_streaming_table("employees_history")

dp.create_auto_cdc_flow(
    target="employees_history",
    source="employees_cdf",
    keys=["employee_id"],
    sequence_by=col("sequence_num"),
    apply_as_deletes=expr("operation = 'DELETE'"),
    except_column_list=["operation", "sequence_num"],
    stored_as_scd_type=2
)

ScD Type 2 aggiunge __START_AT e __END_AT colonne per tenere traccia di ogni versione attiva. Il record Seattle rimane con una data di fine, mentre un nuovo record Portland viene visualizzato senza data di fine. Gli analisti possono eseguire query sulla tabella per vedere esattamente quando e come cambiano i dati dei dipendenti nel tempo.

Alcuni sistemi di origine generano record di aggiornamento parziali che includono solo le colonne modificate anziché gli snapshot di riga completi. In questi casi, l'applicazione diretta dell'aggiornamento sovrascrive le colonne senza modifiche con valori Null. Il parametro ignore_null_updates (Python) o IGNORE NULL UPDATES clausola (SQL) impedisce questa operazione: quando un evento CDC corrisponde a una riga esistente, eventuali valori Null nel record in ingresso lasciano intatti i valori di colonna esistenti anziché sovrascriverli.

Configurare i flussi CDC in SQL

Gli stessi modelli CDC funzionano in SQL usando la sintassi dichiarativa. I flussi SQL si integrano con naturalezza nelle pipeline dell'architettura medallion, in cui si definiscono i livelli bronzo, argento e oro.

CREATE OR REFRESH STREAMING TABLE employees;

CREATE FLOW employees_cdc_flow
AS AUTO CDC INTO employees
FROM stream(bronze.employee_changes)
KEYS (employee_id)
APPLY AS DELETE WHEN operation = 'DELETE'
SEQUENCE BY sequence_num
COLUMNS * EXCEPT (operation, sequence_num)
STORED AS SCD TYPE 1;

La sintassi SQL rispecchia l'API Python. Specificare la tabella di destinazione, il flusso di origine, le colonne chiave, la condizione di eliminazione, la colonna di sequenziazione e il tipo SCD. Le colonne non necessarie nella destinazione, ad esempio le colonne dell'operazione e della sequenza, vengono escluse con la EXCEPT clausola .

Per il rilevamento cronologico con scD Type 2, è anche possibile limitare le colonne che attivano nuovi record di cronologia:

CREATE FLOW employees_history_flow
AS AUTO CDC INTO employees_history
FROM stream(bronze.employee_changes)
KEYS (employee_id)
SEQUENCE BY sequence_num
COLUMNS * EXCEPT (operation, sequence_num)
STORED AS SCD TYPE 2
TRACK HISTORY ON * EXCEPT (last_login);

Con questa configurazione, le modifiche apportate alla last_login colonna aggiornano il record esistente sul posto. Solo le modifiche apportate ad altre colonne, ad esempio il nome o il reparto, creano nuove versioni cronologiche.

Gestire gli eventi non ordinati

I sistemi distribuiti spesso recapitano eventi non in ordine. Un ritardo di rete potrebbe causare l'arrivo di un aggiornamento precedente dopo uno più recente. L'API AUTO CDC gestisce automaticamente questo scenario utilizzando la colonna di sequenza specificata.

Quando arrivano due aggiornamenti per la stessa chiave, l'API confronta i relativi valori di sequenza. Se l'aggiornamento più recente (numero di sequenza superiore) è già stato applicato e un aggiornamento precedente arriva in ritardo, l'API ignora il record in arrivo in ritardo. In questo modo, la tabella di destinazione riflette sempre lo stato finale corretto.

Diagramma che illustra la gestione degli eventi non ordinati.

Senza questa gestione predefinita, è necessario scrivere una logica di unione complessa che controlla i timestamp, confronta le versioni e applica gli aggiornamenti in modo condizionale. L'API AUTO CDC incapsula tutta questa logica, consentendo di concentrarsi sui requisiti aziendali anziché sulla gestione dei casi perimetrali.