Introduzione alla DLT Sink API: Scrivi Pipeline su Kafka e Tabelle Delta Esterne

Se sei nuovo a Delta Live Tables, prima di leggere questo post ti consigliamo di leggere Introduzione a Delta Live Tables, che spiega come creare pipeline scalabili e affidabili utilizzando definizioni e istruzioni ETL dichiarative di Delta Live Tables (DLT).

Introduzione

Delta Live Tables (DLT) offre una piattaforma robusta per la creazione di pipeline di elaborazione dati affidabili, manutenibili e testabili all'interno di Databricks. Sfruttando il suo framework dichiarativo e il provisioning automatico di risorse di calcolo serverless ottimali, DLT semplifica le complessità dello streaming, della trasformazione e della gestione dei dati, garantendo scalabilità ed efficienza per i moderni flussi di lavoro dei dati.

Tradizionalmente, le pipeline DLT hanno offerto un modo efficiente per ingerire ed elaborare dati come Tabelle di Streaming o Viste Materializzate governate da Unity Catalog. Sebbene questo approccio soddisfi la maggior parte delle esigenze di elaborazione dati, ci sono casi in cui le pipeline dati devono connettersi con sistemi esterni o utilizzare sink di Structured Streaming anziché scrivere su Tabelle di Streaming o Viste Materializzate.

L'introduzione della nuova API Sinks in DLT affronta questo problema consentendo agli utenti di scrivere dati elaborati su flussi di eventi esterni, come Apache Kafka, Azure Event Hubs, oltre a scrivere su una Tabella Delta. Questa nuova funzionalità amplia l'ambito delle pipeline DLT, consentendo un'integrazione fluida con piattaforme esterne.

Queste funzionalità sono ora in anteprima pubblica e continueremo ad aggiungere altri sink da Databricks Runtime a DLT nel tempo, supportandoli tutti. Il prossimo su cui stiamo lavorando è foreachBatch, che consente ai clienti di scrivere su sink di dati arbitrari ed eseguire merge personalizzati in tabelle Delta.

L'API Sink è disponibile nel pacchetto Python dlt e può essere utilizzata con create_sink() come mostrato di seguito:

L'API accetta tre argomenti chiave nella definizione del sink:

• Nome Sink: Una stringa che identifica univocamente il sink all'interno della tua pipeline. Questo nome ti consente di fare riferimento e gestire il sink.
• Specifiche del Formato: Una stringa che determina il formato di output, con supporto per "kafka" o "delta".
• Opzioni Sink: Un dizionario di coppie chiave-valore, in cui sia le chiavi che i valori sono stringhe. Per i sink Kafka, tutte le opzioni di configurazione disponibili in Structured Streaming possono essere sfruttate, incluse le impostazioni per l'autenticazione, le strategie di partizionamento e altro ancora. Si prega di fare riferimento alla documentazione per un elenco completo delle opzioni di configurazione supportate da Kafka. I sink Delta offrono una configurazione più semplice consentendo di definire un percorso di archiviazione utilizzando l'attributo path o di scrivere direttamente su una tabella in Unity Catalog utilizzando l'attributo tableName.

Scrittura su un Sink

L'API @append_flow è stata migliorata per consentire la scrittura di dati in sink di destinazione identificati dai loro nomi. Tradizionalmente, questa API consentiva agli utenti di caricare dati senza problemi da più origini in un'unica tabella di streaming. Con il nuovo miglioramento, gli utenti possono ora aggiungere dati a sink specifici. Di seguito è riportato un esempio che dimostra come configurarlo:

Costruzione della pipeline

Costruiamo ora una pipeline DLT che elabora dati di clickstream, pacchettizzati all'interno dei dataset Databricks. Questa pipeline analizzerà i dati per identificare gli eventi che rimandano a una pagina Apache Spark e successivamente scriverà questi dati sia su Event Hubs che su sink Delta. Struttureremo la pipeline utilizzando la Architettura Medallion, che organizza i dati in diversi livelli per migliorarne la qualità e l'efficienza di elaborazione.

Iniziamo caricando i dati JSON grezzi nel livello Bronze utilizzando Auto Loader. Quindi, puliamo i dati e applichiamo standard di qualità nel livello Silver per garantirne l'integrità. Infine, nel livello Gold, filtriamo le voci con un titolo della pagina corrente di Apache_Spark e le memorizziamo in una tabella denominata spark_referrers, che servirà da origine per i nostri sink. Si prega di fare riferimento all'Appendice per il codice completo.

Configurazione del Sink Azure Event Hubs

In questa sezione, utilizzeremo l'API create_sink per stabilire un sink Event Hubs. Questo presuppone che tu disponga di uno stream Kafka o Event Hubs operativo. La nostra pipeline trasmetterà dati in streaming su Event Hubs abilitati per Kafka utilizzando una policy di accesso condiviso, con la stringa di connessione archiviata in modo sicuro in Databricks Secrets. In alternativa, è possibile utilizzare un service principal per l'integrazione invece di una policy SAS. Assicurati di aggiornare le proprietà di connessione e i segreti di conseguenza. Ecco il codice per configurare il sink Event Hubs:

Configurazione del Sink Delta

Oltre al sink Event Hubs, possiamo utilizzare l'API create_sink per configurare un sink Delta. Questo sink scrive dati in una posizione specificata nel Databricks File System (DBFS), ma può anche essere configurato per scrivere su una posizione di object storage come Amazon S3 o ADLS.

Di seguito è riportato un esempio di come configurare un sink Delta:

Creazione di Flussi per popolare i sink Kafka e Delta

Con i sink Event Hubs e Delta configurati, il passo successivo è popolare questi sink utilizzando il decoratore append_flow. Questo processo prevede lo streaming dei dati nei sink, garantendo che siano continuamente aggiornati con le informazioni più recenti.

Per il sink Event Hubs, il parametro value è obbligatorio, mentre parametri aggiuntivi come key, partition, headers e topic possono essere specificati facoltativamente. Di seguito sono riportati esempi di come configurare i flussi sia per i sink Kafka che per quelli Delta:

La funzione applyInPandasWithState è ora supportata anche in DLT, consentendo agli utenti di sfruttare la potenza di Pandas per l'elaborazione stateful all'interno delle loro pipeline DLT. Questo miglioramento consente trasformazioni e aggregazioni di dati più complesse utilizzando la familiare API Pandas. Con l'API DLT Sink, gli utenti possono facilmente trasmettere questi dati elaborati stateful ai topic Kafka. Questa integrazione è particolarmente utile per l'analisi in tempo reale e le architetture event-driven, garantendo che le pipeline dati possano gestire e distribuire in modo efficiente i dati in streaming ai sistemi esterni.

Riassumendo

L'approccio dimostrato sopra mostra come costruire una pipeline DLT che elabora efficientemente i dati utilizzando la nuova API Sink per fornire senza problemi i risultati a tabelle Delta esterne e Event Hubs abilitati per Kafka.

Questa funzionalità è particolarmente preziosa per le pipeline di analisi in tempo reale, consentendo ai dati di essere trasmessi in streaming su flussi Kafka per applicazioni come il rilevamento di anomalie, la manutenzione predittiva e altri casi d'uso sensibili al tempo. Abilita anche architetture event-driven, in cui i processi downstream possono essere attivati istantaneamente trasmettendo eventi ai topic Kafka, consentendo un'elaborazione rapida dei dati appena arrivati.

Call to Action

La funzionalità DLT Sinks è ora disponibile in anteprima pubblica per tutti i clienti Databricks! Questa potente nuova capacità ti consente di estendere senza problemi le tue pipeline DLT a sistemi esterni come Kafka e tabelle Delta, garantendo il flusso di dati in tempo reale e integrazioni semplificate. Per ulteriori informazioni, consulta le seguenti risorse:

• Documentazione Databricks sui DLT Sinks: Databricks docs
• Guarda la demo sui Kafka Sinks in DLT utilizzando Confluent Cloud: Watch the demo
• Riferimento al linguaggio Python per DLT: Python Language reference

Appendice:

Codice della pipeline:

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