CDC basato su push in Unity Catalog con Zerobus Ingest e Debezium

I dati nei database operativi guidano le decisioni aziendali e gran parte di tale analisi avviene nel lakehouse. Spostare i dati lì risulta complesso quando il database si trova in una rete aziendale privata, in un ambiente controllato dal cliente o in una sottorete limitata le cui policy di rete bloccano le connessioni in entrata dai servizi esterni.

Molti team necessitano delle modifiche PostgreSQL in Databricks, ma il team che gestisce il database desidera anche occuparsi dell'acquisizione. Quel team esegue il processo che legge il registro di scrittura preventiva (write-ahead log) e decide quali connessioni lasciano la propria rete. Dall'altra parte, le modifiche arrivano in tabelle Delta governate su cui le pipeline e gli analisti possono lavorare.

Debezium è un comune strumento open source di cui i team possono gestire l'intera distribuzione per estrarre un flusso di change data capture da un database. Si posiziona accanto al database, legge il suo registro di scrittura preventiva e invia gli eventi verso l'esterno. Zerobus Ingest è la destinazione, un'interfaccia di push che scrive i record direttamente nelle tabelle Delta di Unity Catalog. Il sink open-source Debezium Server Zerobus unisce i due elementi senza l'uso di un tradizionale cluster Kafka tra l'acquisizione e il lakehouse.

Gli ingegneri di NVIDIA e Databricks hanno unito le forze per realizzare questo sink e lo hanno contribuito al progetto open-source Debezium. Databricks ha sviluppato il sink tipizzato di Zerobus Ingest. Successivamente, NVIDIA ha introdotto la modalità envelope, che inserisce l'intero evento di modifica di Debezium, insieme alle relative garanzie di consegna, come le chiavi di idempotenza e la conferma durevole prima dell'avanzamento degli offset di origine. Il sink di Zerobus Ingest è incluso nella release 3.7 di Debezium Server.

Questo articolo descrive le funzioni di Debezium e Zerobus, l'envelope di modifica grezzo aggiunto da NVIDIA al sink e il modo in cui gli eventi acquisiti diventano tabelle di stato corrente o storiche. Per una guida pratica, inclusa l'esecuzione in tempo reale da una tabella PostgreSQL locale a una tabella di stato corrente, consulta il relativo post del blog sulla Databricks Community. 

La sfida

La replica affidabile delle modifiche del database operativo nel lakehouse deve risolvere contemporaneamente diversi problemi:

• Acquisire inserimenti, aggiornamenti ed eliminazioni in modo affidabile. Una riga creata, modificata e successivamente rimossa deve generare un flusso fedele per ciascun evento. Questo deve includere le eliminazioni, che gli approcci basati su query di tipo "copia la tabella" spesso non colgono.
• Mantenere le informazioni necessarie per l'ordinamento. La logica a valle che ricostruisce lo stato corrente deve applicare le modifiche nell'ordine di origine, così da replicare correttamente la tabella di origine.
• Evitare di sovraccaricare il database di origine. Eseguire il polling del database secondo una pianificazione (SELECT ... WHERE updated_at > ?) aumenta il carico delle query e scala con difficoltà.
• Funzionare con reti private o controllate dal cliente. Quando il database si trova dietro un perimetro protetto, sappiamo che connettersi a questi sistemi e acquisire i dati può risultare complesso.
• Evitare di configurare Kafka unicamente come mezzo di trasporto. Kafka è un potente event bus, ma gestire un cluster solo per spostare gli eventi CDC da un database a un lakehouse rappresenta un sovraccarico considerevole per i team che non ne hanno altrimenti bisogno.
• Memorizzare i dati in modo che possano essere elaborati facilmente. Il flusso di modifiche deve arrivare in un formato che il lakehouse è in grado di elaborare ed interrogare, anziché in un silo che richiede un ulteriore passaggio di esportazione.

Visti questi requisiti, dobbiamo partire dall'architettura che definisce il flusso dei dati dalla sorgente al sink.

Architettura di riferimento complessiva

In questo blog analizzeremo un'architettura di message bus tradizionale confrontandola con un'architettura basata su Zerobus Ingest.

Architettura con message bus:

In molti ambienti aziendali, la connettività in uscita dall'interno di un perimetro è molto più semplice da autorizzare rispetto all'accesso in entrata. Quando il tuo team desidera eseguire una distribuzione dell'agente che estrae il feed delle modifiche, un modello di push locale consente ai dati di fluire verso l'esterno verso una destinazione. In una configurazione tradizionale, si tratterebbe di un message bus come Kafka. L'utilizzo di Kafka richiede risorse per la pianificazione della capacità, al fine di prevedere il numero di broker necessari per gestire il picco di carico per la scalabilità fisica e stimare il numero di partizioni per la scalabilità logica, che i consumer devono seguire per tenere il passo con la velocità di produzione. Ciò richiede un team dedicato con esperienza nella configurazione e gestione di Kafka.

Una volta che i dati arrivano in Kafka, il passo successivo consiste nel replicare tali eventi nel lakehouse. Questo può avvenire tramite la distribuzione di processi di ingestione come Spark, Kafka Connector o altri strumenti. Il risultato è un'architettura a due passaggi da Debezium al lakehouse.

Architettura con Zerobus Ingest:

Quando l'unica destinazione è il lakehouse, Zerobus consente di saltare i passaggi intermedi come il message bus. Ciò semplifica l'architettura end-to-end ed elimina la necessità di distribuire e gestire l'infrastruttura Kafka per il trasferimento dei dati. Quando si utilizza Zerobus come sink con Debezium, il sistema continua a funzionare regolarmente anche durante le ore di punta, senza necessità di tuning.

Che cos'è Debezium?

Debezium è una piattaforma open-source per il change data capture basata su log. Invece di interrogare le tabelle tramite polling, legge il log delle transazioni di un database, ad esempio il registro di scrittura preventiva (WAL) di PostgreSQL, il binlog di MySQL e i change stream di MongoDB. Successivamente, genera ogni modifica sotto forma di evento strutturato. Il progetto è nato nel 2015 e oggi supporta un ampio ecosistema di connettori (PostgreSQL, MySQL, SQL Server, Oracle, MongoDB e altri).

Ogni evento di modifica contiene un envelope coerente:

• before — lo stato della riga prima della modifica (null per gli inserimenti)
• after — lo stato della riga dopo la modifica (null per le eliminazioni)\
• op — l'operazione: c (creazione/inserimento), u (aggiornamento), d (eliminazione), r (lettura/snapshot)
• source — metadati: database, schema, tabella, numero di sequenza del log WAL (LSN), ID transazione, flag di snapshot
• ts_ms — timestamp dell'evento

Debezium esercita un carico minimo sul database primario grazie all'analisi dei log (log tailing) anziché all'uso del polling per catturare le modifiche a bassa latenza, preservando al contempo l'ordine dei commit.

Standalone (Debezium Server) rispetto alla distribuzione con Kafka Connect

Debezium prevede due modelli operativi. Uno viene eseguito su Kafka Connect, mentre l'altro è un "server autonomo" che può essere avviato su risorse di calcolo standard.

Debezium Server, il runtime indipendente, funge da agente lato origine che traccia i log delle transazioni e invia gli eventi di modifica direttamente a Zerobus Ingest di Databricks, senza un cluster Kafka a fare da trasporto.

La soluzione: Debezium + Zerobus Ingest

Abbiamo quindi identificato Debezium come il nostro sistema di tracciamento dei log. Ora dobbiamo inserire gli eventi in Databricks.

Zerobus Ingest è un servizio di inserimento serverless basato sul push che scrive i dati direttamente nelle tabelle Delta gestite da Unity Catalog, senza alcun message broker da configurare. I produttori si collegano a un endpoint Zerobus e trasmettono i record che finiscono in una tabella Delta in formato aperto. È progettato per gestire carichi di lavoro a elevata concorrenza e throughput. Evitando il passaggio di un message broker, riusciamo a semplificare l'architettura e la manutenzione legata alla scalabilità e all'esecuzione di pipeline di inserimento per spostare i dati nel lakehouse.

Insieme a Debezium, le tue pipeline CDC appariranno così:

• Debezium monitora il log del database e converte ogni modifica in un evento CDC.
• Zerobus Ingest trasporta questi eventi direttamente in Databricks e li scrive in una tabella Delta, senza alcun cluster Kafka come livello di trasporto intermedio.
• Databricks riceve il flusso di modifiche in una tabella Delta e può elaborarlo in tabelle curate a valle.

Zerobus Ingest supporta solo l'aggiunta (append-only). Una volta ricevuti gli eventi di modifica, il passaggio successivo consiste nel creare una tabella dimensionale di Tipo 1 eseguendo un merge per ottenere la versione più recente di ciascuna combinazione di chiavi primarie.

Il contributo di NVIDIA: la modalità envelope

Il sink Zerobus in Debezium offre due contratti di tabella. La modalità Typed, predefinita, mappa ogni evento di modifica su una tabella con le colonne originali della sorgente. La modalità Envelope, introdotta grazie al contributo di NVIDIA, inserisce l'intero evento di modifica di Debezium in una tabella fissa a nove colonne:

1. target_table
2. destination
3. partition
4. operation
5. idempotency_key
6. key
7. value
8. source_position
9. headers

I record possono essere codificati come JSON (memorizzati nelle colonne VARIANT) o Protobuf.

La modalità envelope conserva le informazioni necessarie per l'elaborazione a valle. source_position contiene la partizione e l'offset di origine, incluso l'LSN di PostgreSQL. idempotency_key è un'identità SHA-256 di ciascun evento, derivata da destinazione, partizione, chiave, operazione e posizione di origine, in modo che un evento inviato nuovamente mantenga la stessa chiave. La chiave identifica gli eventi, ma non li ordina né esegue autonomamente la deduplicazione della tabella. 

Se un gruppo successivo fallisce, un gruppo persistente precedente può essere inviato di nuovo; la consegna è quindi di tipo at-least-once e i duplicati vengono gestiti a valle con idempotency_key. La modalità envelope filtra inoltre per destinazione e operazione e rifiuta i record troppo grandi prima dell'inserimento, in modo che il relativo offset non venga mai confermato.

Modellazione dei dati in Databricks

Una volta che gli eventi CDC arrivano in Delta, un flusso in stile medallion trasforma il feed di modifiche grezzo in qualcosa che analisti e applicazioni possono utilizzare:

• Bronze: gli eventi di modifica non elaborati, così come ricevuti. Con la modalità envelope, corrisponde a una riga per evento, che include l'operazione, le immagini di riga prima/dopo, la posizione di origine (LSN) e idempotency_key.
• Silver: record di modifica analizzati, deduplicati per idempotency_key e ordinati per posizione di origine. Rappresentazione utilizzabile dei dati di origine, con inserimenti/aggiornamenti/eliminazioni interpretati per ricostruire lo stato corrente.
• Gold: i dati operativi vengono trasformati insieme ai dati aziendali per la generazione di report, come dati finanziari, metriche di performance e così via.

Vale la pena menzionare brevemente due modi per strutturare il flusso di modifiche:

• Tipo 1 (stato corrente): comprime il registro delle modifiche in modo che ciascuna chiave rifletta il suo valore più recente, ovvero una replica dello stato corrente della tabella di origine.
• Tipo 2 (cronologia): conserva ciascuna versione nel tempo, consentendo di interrogare i dati com'erano in un determinato momento.

Il blog pratico associato sulla Databricks Community illustra questo processo end-to-end utilizzando una tabella PostgreSQL locale. Configura Debezium Server con il sink Zerobus, verifica un inserimento, un aggiornamento e un'eliminazione nella tabella envelope e quindi crea la tabella di Tipo 1 (stato corrente) con AUTO CDC, ordinando le modifiche in base all'LSN di origine.

Conclusioni

La scelta dell'architettura CDC più adatta dipende in ultima analisi dai vincoli di rete, dalle preferenze operative e dall'infrastruttura esistente della tua organizzazione. 

Combinando Debezium Server con Zerobus Ingest, puoi distribuire un'architettura push locale che trasmette in streaming le modifiche direttamente nelle tabelle Delta senza richiedere Kafka come trasporto di messaggi intermedio o fare affidamento su aree di stage nell'object storage cloud, semplificando il percorso di acquisizione end-to-end.

In questo modo la cattura di PostgreSQL rimane sotto il controllo del team proprietario del database, mentre i dati arrivano in Unity Catalog. 

Per iniziare

Per creare autonomamente questa pipeline, segui la guida passo passo sulla Databricks Community. Questa invia le modifiche di PostgreSQL tramite Debezium Server e Zerobus Ingest a una tabella di Unity Catalog, quindi crea una tabella dello stato attuale con AUTO CDC. Per la configurazione di Zerobus Ingest e le sue altre interfacce di acquisizione, consulta la documentazione di Zerobus Ingest.

Domande frequenti

Cos'è il sink Debezium Zerobus Ingest?

È un sink in Debezium Server, il runtime Debezium indipendente, che invia gli eventi di modifica direttamente a Zerobus Ingest. Zerobus Ingest li scrive nelle tabelle Delta gestite da Unity Catalog, quindi nessun cluster Kafka si interpone tra la cattura e il lakehouse. NVIDIA e Databricks hanno collaborato al contributo open source.

Cos'è la modalità envelope?

La modalità envelope posiziona ogni evento di modifica Debezium completo in una tabella fissa a nove colonne anziché mapparlo sulle colonne originali della sorgente. NVIDIA ha contribuito a questa funzionalità per il sink. Mantiene l'operazione, la chiave, il valore, la posizione della sorgente e una chiave di idempotenza per l'elaborazione a valle. La modalità tipizzata è quella predefinita.

Il sink consegna ogni modifica esattamente una volta?

No, la consegna è di tipo at-least-once. Se un gruppo successivo di record non va a buon fine, un gruppo persistente precedente potrebbe essere consegnato di nuovo. Ogni evento trasporta un idempotency_key, in modo da poter eseguire la deduplicazione a valle.

Come faccio a trasformare il flusso di modifiche in una tabella dello stato attuale?

Usa AUTO CDC per creare una tabella Type-1 che conserva l'ultima versione di ciascuna chiave. Metti in sequenza le modifiche in base all'LSN di origine in modo che vengano applicate nell'ordine originale. Per conservare invece tutte le versioni nel tempo, crea una tabella Type-2.

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