Portare AUTO CDC al livello successivo: risolvere i casi d'uso più complessi del mondo reale

Il Change Data Capture è una delle attività più comuni che i data engineer creano su Spark, e una delle più tediose da realizzare correttamente a mano. Nel nostro post precedente, Basta scrivere a mano le pipeline di change data capture, abbiamo presentato come AUTO CDC in Apache™ Spark Declarative Pipelines (SDP) automatizza SCD Type 1, SCD Type 2 e Snapshot CDC, sostituendo centinaia di righe di fragile logica MERGE con poche e semplici dichiarazioni.

Con l'evolversi dei requisiti delle pipeline, i data engineer si scontrano con situazioni che i pattern CDC standard faticano a risolvere:

• Gestire timeline bitemporali non in ordine
• Elaborare aggiornamenti parziali dei record senza corrompere i dati esistenti
• Mantenere una tracciabilità che superi le finestre di conservazione dello storage

Oggi stiamo portando AUTO CDC a un livello superiore per risolvere proprio queste sfide del mondo reale, ed estendendo queste funzionalità nella versione open-source di Apache Spark 4.2.

Tracciamento della cronologia su due assi con AUTO CDC bitemporale

Le tabelle SCD Type 2 standard possono indicare quando un fatto è cambiato nel mondo reale, ma non possono mostrare ciò che il sistema riteneva vero in un determinato momento.

In base alla norma SEC Rule 17a-4 e alle regole di conservazione dei registri FINRA, le aziende devono essere in grado di ricostruire i record così com'erano in un determinato momento; solo i controlli della SEC sulla conservazione dei registri hanno portato a oltre 2 miliardi di dollari di sanzioni per più di 100 aziende dal 2021. La parte difficile raramente consiste nel memorizzare il valore odierno. La vera sfida è rispondere, a distanza di mesi, a cosa dicessero i dati di riferimento alla data del report e cosa ritenessero vero i nostri sistemi in quel momento.

Lo standard SCD Type 2 traccia una sola timeline: quando un fatto è cambiato. AUTO CDC bitemporale ne traccia due, in modo indipendente:

• Business time (noto anche come event time o valid time): quando il fatto era effettivamente vero nel mondo reale. Ad esempio, un simbolo azionario è diventato segnalabile il lunedì; un codice paese è stato ritirato alla fine del trimestre.
• System time (noto anche come transaction time o processing time): quando il sistema di registrazione ha appreso il dato. La modifica del lunedì potrebbe non arrivare nella pipeline fino a mercoledì.

Ogni tabella di destinazione riceve quattro colonne gestite dal sistema: __START_AT e __END_AT per il business time, __SYSTEM_START_AT e __SYSTEM_END_AT per il system time. Un singolo fatto logico può avere diverse righe fisiche, una per ogni combinazione di versione business/versione sistema, ed è proprio questo che rende possibile la ricostruzione temporale lungo entrambi gli assi. La garanzia di comportamento fondamentale: gli eventi possono arrivare in qualsiasi ordine su entrambe le timeline.

Quando si presenta una correzione con un business time o un system time precedente rispetto a qualcosa già elaborato, l'engine riscrive la cronologia interessata invece di limitarsi ad aggiungerla alla fine. Nessuna logica scritta a mano: basta dichiarare le due colonne di sequenziamento e l'engine manterrà entrambi gli intervalli. Questo funziona altrettanto bene per le tabelle delle dimensioni, come le anagrafiche dei simboli, e per le tabelle dei fatti, come la cronologia delle transazioni o le letture dei sensori, che richiedono una tracciabilità rigorosa. Ecco come si presenta rispetto ai dati di riferimento FINRA CAT:

Nota che la clausola SQL esatta è STORED AS BITEMPORAL, non STORED AS SCD TYPE BITEMPORAL, e richiede sia SEQUENCE BY che SYSTEM SEQUENCE BY. Ipotizziamo che il flag di segnalazione di Acme cambi il 1° gennaio (business time), ma il feed non lo riceva fino al 5 gennaio (system time). Successivamente, l'8 gennaio arriva una correzione retrodatata che indica che la modifica reale è avvenuta il 1° gennaio, ma con un valore diverso. L'AUTO CDC bitemporale può rispondere a entrambe le domande:

Il 3 gennaio, la prima query non restituisce nulla: la risposta corretta e verificabile per ciò che il sistema mostrava in quel momento. La seconda query, eseguita oggi, riflette la verità corretta. Due orologi, due risposte, entrambe corrette. Le colonne di sequenziamento devono essere di tipi ordinabili, senza valori di sequenziamento NULL. La funzionalità è disponibile su SDP serverless o sulle edizioni di prodotto Pro/Advanced, ed è attualmente in Beta, quindi blocca la pipeline sul canale: PREVIEW.

Oltre il time travel: ML riproducibile che sopravvive a VACUUM

Quando un modello viene addestrato su dati di riferimento o feature, riproducibilità significa essere in grado di ricostruire l'esatto dataset utilizzato dal modello, a distanza di mesi, durante una revisione o un audit. L'istinto è quello di ricorrere al time travel di Delta Lake, ma questa è una proprietà della cronologia dei file della tabella, non un record permanente. VACUUM elimina in modo permanente i file di dati non più referenziati dalle versioni recenti; una volta superata la finestra di conservazione predefinita di 7 giorni, un TIMESTAMP AS OF registrato al momento dell'addestramento può smettere silenziosamente di risolversi. Una tabella bitemporale memorizza tale cronologia come dati, non come versioni di file. VACUUM e OPTIMIZE compattano i file ma non toccano mai la cronologia logica, quindi ogni versione passata di business o di sistema rimane una riga interrogabile. Ci sono due modi per ottenere la riproducibilità da questo processo: registrare due istanti temporali (business time e system time) come parametri MLflow e bloccare la query di addestramento a quello stato di consistenza:

Oppure, se la tabella espone una vista corrente, registra un singolo istante di sistema al momento dell'addestramento e ricostruiscilo in seguito con una query sul system-time a quel timestamp:

In entrambi i casi, il contratto di riproducibilità consiste in una coppia di timestamp nell'esecuzione di MLflow e, poiché la cronologia bitemporale è memorizzata come righe, tale contratto rimane valido anche dopo che VACUUM ha ripulito i file sottostanti.

Gli aggiornamenti parziali di AutoCDC sono ora generalmente disponibili

Non tutte le sorgenti di change data capture (CDC) emettono righe complete per gli aggiornamenti. Al contrario, molte inviano solo i campi modificati, rappresentando tutte le altre colonne come NULL. Senza una gestione speciale, questi valori NULL possono sovrascrivere involontariamente i dati esistenti nella tabella di destinazione. Fino ad ora, i clienti dovevano creare una logica personalizzata per aggirare questo comportamento. Con gli aggiornamenti parziali di AutoCDC, questo processo viene ora gestito automaticamente.

Gli aggiornamenti parziali estendono AutoCDC consentendo agli eventi di aggiornamento di modificare solo un sottoinsieme di colonne. Per le colonne selezionate, i valori NULL in un aggiornamento in entrata vengono interpretati come "non aggiornare" anziché sovrascrivere il valore esistente.

Questo è particolarmente utile per le sorgenti CDC che omettono i valori non modificati emettendo NULL. Senza gli aggiornamenti parziali, questi NULL sovrascriverebbero i dati esistenti nella tabella di destinazione.

Ad esempio, supponiamo che la tabella di destinazione contenga: (1, 'A', 20)

Un evento di aggiornamento in entrata contiene: (1, NULL, 30)

Per impostazione predefinita, AutoCDC aggiornerebbe la riga in: (1, NULL, 30).

Con gli aggiornamenti parziali abilitati, il valore NULL in name viene trattato come "lascia invariato il valore esistente", con il seguente risultato: (1, 'A', 30).

L'abilitazione degli aggiornamenti parziali richiede solo l'aggiunta di un parametro alla definizione di AutoCDC. Puoi scegliere tra tre modi per specificare quali colonne devono essere trattate come aggiornamenti parziali:

1. un elenco di colonne che deve ignorare i valori NULL:
   IGNORE NULL UPDATES ON columnList
2. un elenco di colonne che NON deve ignorare i valori NULL:
   IGNORE NULL UPDATES ON * EXCEPT (columnList)
3. un nome di colonna di origine che può essere diverso per ogni riga:
   COLUMNS TO UPDATE

Per la sintassi completa, gli esempi e le linee guida sull'uso, consulta la documentazione sull'applicazione degli aggiornamenti parziali.

Continuiamo a impegnarci per l'open source

Spark Declarative Pipelines è open source, quindi dovrebbe esserlo anche il suo tipo di flusso più diffuso. Iniziamo contribuendo con l'API Python per AUTO CDC Type 1 a Apache Spark 4.2.

Abbiamo contribuito nello stesso modo in cui si evolve il resto di Spark: come una serie di proposte revisionate e pull request, non come un rilascio di codice una tantum (vedi SPIP e SPARK-56249). La correttezza con i dati fuori ordine è integrata: una piccola tabella ausiliaria tiene traccia dello stato degli eventi in arrivo anticipato come i tombstone di eliminazione, i micro-batch riprovati convergono anziché corrompere la destinazione e, poiché si basa sulle astrazioni di streaming e tabella di Spark anziché su un formato di archiviazione, viene eseguito sia su Delta Lake che su Apache Iceberg.

Le prossime novità, in modalità open source:

• Funzionalità della prossima release: abbiamo già unito l'interfaccia SQL (CREATE FLOW ... AS AUTO CDC INTO) nel ramo master, che sarà disponibile nella prossima release di Apache Spark.
• Semantica avanzata delle pipeline: è in corso lo sviluppo per la gestione della cronologia completa di SCD Type 2, input di changelog nativi e supporto per gli aggiornamenti parziali per evitare che i valori NULL sovrascrivano i dati di destinazione.
• Affidabilità e test: stiamo aggiungendo funzionalità apply-as-truncate e al contempo stiamo ampliando le nostre suite di test automatizzati per i dati fuori ordine e i tentativi idempotenti.

Per iniziare

Che tu voglia implementare la conformità bitemporale, configurare gli aggiornamenti parziali o esplorare AutoCDC open source in Apache Spark, dai un'occhiata alle risorse seguenti per iniziare:

• AUTO CDC bitemporale: scopri come configurare il tracciamento della cronologia a doppio asse in SDP
• Guida agli aggiornamenti parziali: visualizza la sintassi completa e gli esempi per gestire i campi di aggiornamento mancanti senza codice personalizzato.
• Guida alla programmazione di AutoCDC open source: esplora le funzionalità CDC dichiarative disponibili in Apache Spark 4.2

(Questo post sul blog è stato tradotto utilizzando strumenti basati sull'intelligenza artificiale) Post originale