Introduzione di transformWithState in Apache Spark™ Structured Streaming

Introduzione

L'elaborazione di stream stateful si riferisce all'elaborazione di un flusso continuo di eventi in tempo reale mantenendo uno stato basato sugli eventi visti finora. Ciò consente al sistema di tracciare modifiche e pattern nel tempo nel flusso di eventi e consente di prendere decisioni o intraprendere azioni basate su queste informazioni.

L'elaborazione di stream stateful in Apache Spark Structured Streaming è supportata utilizzando operatori integrati (come aggregazione con finestre, join stream-stream, eliminazione duplicati, ecc.) per logica predefinita e utilizzando flatMapGroupWithState o mapGroupWithState per logica arbitraria. La logica arbitraria consente agli utenti di scrivere il proprio codice personalizzato di manipolazione dello stato nelle loro pipeline. Tuttavia, con la crescente adozione dello streaming in azienda, applicazioni di streaming più complesse e sofisticate richiedono diverse funzionalità aggiuntive per semplificare agli sviluppatori la scrittura di pipeline di streaming stateful.

Per supportare queste nuove e crescenti applicazioni di streaming stateful o casi d'uso operativi, la community di Spark sta introducendo un nuovo operatore Spark chiamato transformWithState. Questo operatore consentirà modellazione flessibile dei dati, tipi compositi, timer, TTL, concatenamento di operatori stateful dopo transformWithState, evoluzione dello schema, riutilizzo dello stato da una query diversa e integrazione con una serie di altre funzionalità Databricks come Unity Catalog, Delta Live Tables e Spark Connect. Utilizzando questo operatore, i clienti possono sviluppare ed eseguire i loro casi d'uso operativi stateful critici, complessi in modo affidabile ed efficiente sulla piattaforma Databricks utilizzando linguaggi popolari come Scala, Java o Python.

Applicazioni/Casi d'uso che utilizzano l'elaborazione di stream stateful

Molte applicazioni event-driven si basano sull'esecuzione di calcoli stateful per attivare azioni o emettere eventi di output che vengono solitamente scritti su un altro log di eventi/message bus come Apache Kafka/Apache Pulsar/Google Pub-Sub ecc. Queste applicazioni implementano solitamente una state machine che convalida regole, rileva anomalie, traccia sessioni, ecc., e genera i risultati derivati, che vengono solitamente utilizzati per attivare azioni sui sistemi downstream, basati su:

• Eventi di input
• Stato
• Tempo (capacità di lavorare con il tempo di elaborazione e il tempo dell'evento)
• Eventi di output

Esempi di tali applicazioni includono il Tracciamento dell'esperienza utente, il Rilevamento di anomalie, il Monitoraggio dei processi aziendali e gli Alberi decisionali.

Introduzione a transformWithState: un'API stateful più potente

Apache Spark introduce ora transformWithState, un operatore di elaborazione stateful di nuova generazione progettato per rendere la creazione di applicazioni di streaming complesse e in tempo reale più flessibile, efficiente e scalabile. Questa nuova API sblocca funzionalità avanzate per la gestione dello stato, l'elaborazione degli eventi, la gestione dei timer e l'evoluzione dello schema, consentendo agli utenti di implementare logiche di streaming sofisticate con facilità.

Progettazione di alto livello

Stiamo introducendo un nuovo approccio API stratificato, flessibile ed estensibile per affrontare le limitazioni sopra menzionate. Un diagramma architetturale di alto livello dell'architettura stratificata e delle funzionalità associate nei vari livelli è mostrato di seguito.

Come mostrato nella figura, continuiamo a utilizzare i backend di stato disponibili oggi. Attualmente, Apache Spark supporta due backend di state store:

• HDFSBackedStateStoreProvider
• RocksDBStateStoreProvider

Il nuovo operatore transformWithState sarà inizialmente supportato solo con il provider di state store RocksDB. Utilizziamo varie funzionalità di RocksDB relative a scansioni di intervalli, operatori di merge, ecc. per garantire prestazioni ottimali per le varie funzionalità utilizzate all'interno di transformWithState. Sopra questo livello, costruiamo un altro livello di astrazione che utilizza StatefulProcessorHandle per lavorare con tipi compositi, timer, metadati di query, ecc. A livello di operatore, abilitiamo l'uso di StatefulProcessor che può incorporare la logica dell'applicazione utilizzata per fornire queste potenti applicazioni di streaming. Infine, è possibile utilizzare StatefulProcessor all'interno delle query Apache Spark basate sulle API DataFrame.

Ecco un esempio di query di streaming Apache Spark che utilizza l'operatore transformWithState:

Funzionalità chiave con transformWithState

Modellazione flessibile dei dati con variabili di stato

Con transformWithState, gli utenti possono ora definire più variabili di stato indipendenti all'interno di uno StatefulProcessor basato sul modello di programmazione orientata agli oggetti. Queste variabili funzionano come membri privati di una classe, consentendo una gestione granulare dello stato senza richiedere una struttura di stato monolitica. Ciò rende facile evolvere la logica dell'applicazione nel tempo aggiungendo o modificando variabili di stato senza riavviare le query da una nuova directory di checkpoint.

Timer e callback per l'elaborazione event-driven

Gli utenti possono ora registrare timer per attivare la logica dell'applicazione event-driven. L'API supporta timer basati sul tempo di elaborazione (basato sull'orologio) e sul tempo dell'evento (basato su colonna). Quando un timer scade, viene emesso un callback, che consente un'efficiente gestione degli eventi, aggiornamenti dello stato e generazione di output. La possibilità di elencare, registrare ed eliminare timer garantisce un controllo preciso sull'elaborazione degli eventi.

Supporto nativo per tipi di dati compositi

La gestione dello stato è ora più intuitiva con il supporto integrato per strutture dati composite:

• ValueState: Memorizza un singolo valore per chiave di raggruppamento.
• ListState: Mantiene un elenco di valori per chiave, supportando operazioni di aggiunta efficienti.
• MapState: Abilita l'archiviazione chiave-valore all'interno di ogni chiave di raggruppamento con ricerche puntuali efficienti

Spark codifica e persiste automaticamente questi tipi di stato, riducendo la necessità di serializzazione manuale e migliorando le prestazioni.

Scadenza automatica dello stato con TTL

Per conformità ed efficienza operativa, transformWithState introduce il supporto nativo time-to-live (TTL) per le variabili di stato. Ciò consente agli utenti di definire policy di scadenza, garantendo che i vecchi dati di stato vengano rimossi automaticamente senza richiedere pulizia manuale.

Concatenamento di operatori dopo transformWithState

Con questa nuova API, gli operatori stateful possono ora essere concatenati dopo transformWithState, anche quando si utilizza il tempo dell'evento come modalità temporale. Facendo riferimento esplicitamente alle colonne del tempo dell'evento nello schema di output, gli operatori downstream possono eseguire il filtraggio dei record tardivi e l'eviction dello stato senza problemi, eliminando la necessità di complessi workaround che coinvolgono pipeline multiple e storage esterno.

Inizializzazione semplificata dello stato

Gli utenti possono inizializzare lo stato da query esistenti, rendendo più facile riavviare o clonare i job di streaming. L'API consente un'integrazione fluida con il lettore di sorgenti dati di stato, consentendo alle nuove query di sfruttare lo stato scritto in precedenza senza complessi processi di migrazione.

Evoluzione dello schema per query stateful

transformWithState supporta l'evoluzione dello schema, consentendo modifiche come:

• Aggiunta o rimozione di campi
• Riordinamento dei campi
• Aggiornamento dei tipi di dati

Apache Spark rileva e applica automaticamente aggiornamenti compatibili dello schema, garantendo che le query possano continuare a essere eseguite all'interno della stessa directory di checkpoint. Ciò elimina la necessità di ricostruzioni complete dello stato e rielaborazioni, riducendo significativamente i tempi di inattività e la complessità operativa.

Integrazione nativa con il lettore di sorgenti dati di stato

Per semplificare il debug e l'osservabilità, transformWithState è integrato nativamente con il lettore di sorgenti dati di stato. Gli utenti possono ispezionare le variabili di stato e interrogare i dati di stato direttamente, semplificando la risoluzione dei problemi e l'analisi, incluse funzionalità avanzate come readChangeFeed ecc.

Disponibilità

L'API transformWithState è ora disponibile con la release Databricks Runtime 16.2 nei cluster No-Isolation e Unity Catalog Dedicated. Il supporto per i cluster Unity Catalog Standard e il calcolo Serverless verrà aggiunto a breve. L'API è anche prevista per essere disponibile in open-source con la release Apache Spark™ 4.0.

Conclusione

Riteniamo che tutti i miglioramenti delle funzionalità inclusi nella nuova API transformWithState consentiranno la creazione di una nuova classe di carichi di lavoro operativi affidabili, scalabili e mission-critical che alimenteranno i casi d'uso più importanti per i nostri clienti e utenti, il tutto nel comfort e nella facilità d'uso delle API Apache Spark DataFrame. È importante notare che queste modifiche pongono anche le basi per futuri miglioramenti agli operatori stateful integrati e nuovi in Apache Spark Structured Streaming. Siamo entusiasti dei miglioramenti della gestione dello stato in Apache Spark™ Structured Streaming negli ultimi anni e attendiamo con impazienza gli sviluppi previsti nella roadmap in quest'area nel prossimo futuro.

Puoi leggere di più sull'elaborazione di flussi stateful e transformWithState su Databricks qui.

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