Dimensiona correttamente i tuoi flussi stateful più esigenti, dal rilevamento delle frodi alla sorveglianza in tempo reale, senza mai dover ricostruire lo stato del checkpoint
di Thangam Vaiyapuri, Jay Palaniappan, B. Micheal Okutubo e Zifei Feng
Chiunque esegua query stateful di Apache Spark™ Structured Streaming in produzione prima o poi si scontra con lo stesso, fastidioso ostacolo.
Hai avviato la query mesi fa. All'epoca il volume dei dati era modesto, quindi hai accettato il valore predefinito di 200 partizioni di shuffle e sei andato avanti. La pipeline funzionava senza problemi. Poi l'attività è cresciuta, il traffico è triplicato e lo state store è cresciuto a dismisura. All'improvviso, quelle 200 partizioni non sono più della dimensione corretta. Alcune partizioni sono sbilanciate e sovraccariche, il cluster è sotto sforzo e ogni microbatch richiede più tempo del dovuto.
Quindi fai la cosa più naturale: aumenti spark.sql.shuffle.partitions e riavvii la query. Non cambia nulla.
La query ignora silenziosamente il nuovo valore perché il numero di partizioni è stato integrato nel checkpoint al primo avvio dello stream. Storicamente, l'unico modo per applicare un nuovo numero era abbandonare il checkpoint esistente e ricominciare da capo, il che, per una query stateful, significa perdere tutto lo stato accumulato che avevi conservato con cura. Per un modello antifrode che traccia milioni di account o un job di sessionizzazione che conserva giorni di finestre temporali, "ricominciare da capo" non è una frase che si vorrebbe pronunciare durante l'analisi di un incidente in produzione.
Il ripartizionamento dello stato on-demand (Public Preview), disponibile in Databricks Runtime 18 e versioni successive, abbatte questo muro. Ora puoi ridimensionare il numero di partizioni per una query di streaming stateful e mantenere intatto lo stato del checkpoint.
Questo si applica a qualsiasi query di streaming stateful, sia che si eseguano aggregazioni, join stream-stream, deduplicazione, sessionizzazione o transformWithState, e a qualsiasi carico di lavoro, dal rilevamento delle frodi al monitoraggio in tempo reale.
Per i primi utilizzatori come Coveo, la possibilità di ridimensionare correttamente la propria infrastruttura di streaming on-demand si è tradotta immediatamente in un notevole risparmio operativo.
In Coveo gestiamo pipeline di streaming stateful su larga scala in cui i volumi di dati oscillano in modo significativo nel tempo. Con Databricks e la funzionalità di ripartizionamento dello stato, abbiamo ridotto del 40% i relativi costi delle API di Amazon S3. In precedenza, ogni decisione di scalabilità imponeva un compromesso: sovradimensionare le risorse o ricostruire a partire da nuovi checkpoint, il che portava i costi delle API di archiviazione quasi allo stesso livello dei costi di calcolo. Ora scaliamo liberamente al variare della domanda, senza interrompere lo stato esistente o innescare costose migrazioni dei checkpoint”. —Alexis Chicoine, Senior Software Developer, Coveo
Per capire perché i risultati di Coveo rappresentino un passo avanti significativo per Structured Streaming, dobbiamo esaminare il motivo per cui il numero di partizioni veniva bloccato fin dall'inizio.
Una query di streaming stateful conserva il proprio stato in uno state store, e tale stato è partizionato fisicamente. Ogni chiave nel flusso (un ID utente, un numero di conto, una finestra) viene associata tramite hash a una partizione specifica, e i dati di ciascuna partizione vengono archiviati nella propria istanza RocksDB separata all'interno del checkpoint. Il numero di partizioni definisce il layout dell'intero state store sul disco.
Se si modificasse semplicemente il numero di partizioni tra un riavvio e l'altro, l'hashing non corrisponderebbe più. Una chiave che prima si trovava in una partizione (ad esempio, la partizione 47) ora potrebbe essere associata tramite hash a una diversa (la partizione 12), ma il suo stato accumulato si troverebbe ancora nei file della partizione originale. La query, in effetti, perderebbe traccia della propria memoria. Per evitare proprio questo tipo di corruzione silenziosa, Structured Streaming bloccava il numero di partizioni al momento della creazione del checkpoint e ignorava qualsiasi modifica successiva a spark.sql.shuffle.partitions.
Sicuro, ma rigido. I due prezzi da pagare erano:
Il ripartizionamento dello stato on-demand risolve entrambi i problemi facendo l'unica cosa che il vecchio design non consentiva, ma facendolo in modo sicuro: ridistribuendo fisicamente lo stato per adeguarlo al nuovo numero di partizioni.
I requisiti sono minimi:
Questo è l'intero elenco dei prerequisiti. Se utilizzi DBR 18 con lo state store predefinito, hai già tutto ciò che ti serve.
Il meccanismo è semplice e riutilizza un pattern che ogni sviluppatore di streaming conosce già: arresto, riconfigurazione, riavvio.
Invece di spark.sql.shuffle.partitions, imposta una configurazione dedicata, spark.sql.streaming.stateStore.partitions, e riavvia la query:
Il dettaglio fondamentale è la nuova configurazione stessa. Per le query stateful, spark.sql.streaming.stateStore.partitions ha la precedenza su spark.sql.shuffle.partitions. Questo è ciò che rende la modifica permanente, a differenza del vecchio approccio.
Quando la query si riavvia, non riprende immediatamente l'elaborazione normale. Innanzitutto, completa l'ultimo microbatch pianificato, se ce n'è ancora uno in sospeso. Quindi esegue un'operazione di ripartizionamento una tantum: ridistribuisce fisicamente i dati dello stato sul nuovo numero di partizioni, ricalcolando l'hash delle chiavi nelle loro nuove posizioni corrette in modo che nulla vada perso o fuori posto. Una volta completata la ridistribuzione, la query riprende l'elaborazione come di consueto, utilizzando ora il numero di partizioni richiesto.
Questo passaggio di ripartizionamento è il cuore della funzionalità. È la differenza tra "abbiamo cambiato un numero" e "abbiamo spostato in modo sicuro lo stato in un nuovo layout".
Poiché il ripartizionamento è un'operazione effettiva il cui tempo di esecuzione è proporzionale alla quantità di stato, è importante averne visibilità. Structured Streaming lo evidenzia attraverso i suoi report di avanzamento standard.
Al termine del microbatch successivo, gli eventi StreamingQueryProgress includono la durata dell'operazione di ripartizionamento. Cerca nelle metriche durationMs dell'evento il campo controlBatch.REPARTITION, che riporta la durata del ripartizionamento in millisecondi.
Un footprint dello stato più grande comporta un ripartizionamento più lungo, ma prevediamo che richieda solo pochi secondi per la maggior parte dei carichi di lavoro. Quindi, per i job di grandi dimensioni, vale la pena acquisire questa metrica per comprenderne la durata. Per ulteriori informazioni sulla lettura di questi eventi, vedi Monitorare le query Structured Streaming su Databricks.
Rendiamo questo concetto concreto con una semplice aggregazione: un conteggio degli eventi per ID con tumbling window. Inizieremo con il valore predefinito di 200 partizioni, decideremo che è superiore a quanto richiesto da questo carico di lavoro e lo ridurremo a 100.
Innanzitutto, la query così come viene eseguita oggi, con il numero di partizioni predefinito:
Ora, dopo aver monitorato questo stream per un po', abbiamo concluso che 200 partizioni sono eccessive. Stiamo pagando un overhead di coordinamento per un parallelismo di cui non abbiamo bisogno. Arrestiamo la query, impostiamo il nuovo numero di partizioni e la riavviamo con le stesse opzioni e lo stesso checkpoint:
Al riavvio della query, questa conclude l'ultimo microbatch pianificato, se ancora in sospeso, esegue il ripartizionamento per ridistribuire lo stato da 200 a 100 partizioni, quindi continua il conteggio mantenendo perfettamente intatta ogni finestra e ogni totale parziale. La stessa procedura funziona al contrario: per fare uno scale up in caso di carico maggiore, basterà impostare un numero più elevato.
Lo stesso approccio si applica a Spark Declarative Pipelines (SDP). Consulta l'esempio di SDP nella documentazione per una guida completa.
Il ripartizionamento dello stato on-demand è uno strumento di ottimizzazione e scalabilità piuttosto che un'operazione di routine. Si rivela utile in alcune situazioni chiave:
Poiché ogni modifica richiede un arresto e un riavvio con una pausa di ripartizionamento una tantum, considerala come un'azione di manutenzione intenzionale. Pianifica il ridimensionamento in una finestra temporale in cui una breve pausa di elaborazione sia accettabile, monitora controlBatch.REPARTITION per verificare quanto tempo ha richiesto e lascia che la query torni al suo ritmo normale.
Per anni, il numero di partizioni di una query di streaming con stato è stata una decisione da prendere una sola volta, all'inizio, per poi non essere mai più rivista, a meno di non pagare un prezzo elevato per ricostruire lo stato da zero. Il ripartizionamento dello stato on-demand elimina questi vincoli. La didistribuzione sicura dello stato su un nuovo numero di partizioni trasforma una decisione che prima si prendeva solo all'avvio in una scelta che puoi riconsiderare ogni volta che il tuo carico di lavoro lo richiede.
Il risultato è esattamente ciò che gli operatori di flussi a lungo termine desideravano: la libertà di ridimensionare in modo ottimale una query in base alle sue esigenze di scalabilità, semplicemente con un arresto, una modifica della configurazione e un riavvio, senza perdere lo stato.
Il ripartizionamento dello stato on-demand è disponibile in Databricks Runtime 18 e versioni successive, utilizzando il provider di archiviazione dello stato RocksDB. Per la documentazione completa, consulta Ripartizionamento dello stato on-demand per query di streaming con stato.
(Questo post sul blog è stato tradotto utilizzando strumenti basati sull'intelligenza artificiale) Post originale
Iscriviti al nostro blog e ricevi gli ultimi articoli direttamente nella tua casella di posta.