Considerazioni sulla produzione per Structured Streaming

Esegui carichi di lavoro di produzione di Structured Streaming come processi pianificati di Lakeflow in Azure Databricks. Consulta Attività di Lakeflow.

Databricks consiglia di configurare sempre quanto segue:

  • Rimuovere il codice non necessario dai notebook che potrebbero restituire risultati, ad esempio display e count.
  • Non eseguire carichi di lavoro di Structured Streaming su calcolo per utilizzo generico. Pianifica sempre i flussi come Lakeflow Jobs usando il calcolo per i processi.
  • Pianifica i processi Lakeflow usando la modalità Continuousmode. Questo si riferisce alla funzionalità di pianificazione dei processi di Azure Databricks, non al trigger interval di Structured Streaming.
  • Non abilitare la scalabilità automatica delle risorse di calcolo per i job di Structured Streaming.

Alcuni carichi di lavoro traggono vantaggio da quanto segue:

Databricks ha introdotto le pipeline Lakeflow per ridurre la complessità della gestione dell'infrastruttura di produzione per i workload di Structured Streaming. Databricks consiglia di usare le pipeline Lakeflow per le nuove pipeline di structured streaming. Vedi pipeline dichiarative di Spark.

Nota

La scalabilità automatica del calcolo ha dei limiti quando si riduce la dimensione del cluster per carichi di lavoro di streaming strutturati. Databricks consiglia di usare pipeline dichiarative Spark in Lakeflow con scalabilità automatica avanzata per i carichi di lavoro di streaming. Vedi Ottimizzare l'utilizzo del cluster della pipeline Lakeflow con la scalabilità automatica.

:::nota Calcolo serverless

Nell'ambiente di calcolo serverless sono supportati solo Trigger.AvailableNow() e Trigger.Once() . Databricks consiglia Trigger.AvailableNow().

Per lo streaming continuo in un ambiente di calcolo serverless, usare la modalità pipeline Attivata oppure Continua in modalità continua.

Vedere Limitazioni dello streaming.

:::

Ridurre la latenza per lo streaming operativo

I carichi di lavoro operativi in streaming assorbono, trasformano e agiscono sui dati quasi in tempo reale. Esempi comuni includono il rilevamento di frodi, il rilevamento di anomalie, la personalizzazione e il monitoraggio e allerta in tempo reale, dove l'elaborazione ritardata influisce direttamente sugli esiti aziendali. La bassa latenza per questi carichi di lavoro tipicamente significa decine o centinaia di millisecondi, anche se molti team impostano accordi di livello di servizio (SLA) nell'intervallo dei secondi per tenere conto della variabilità ai percentili più alti.

Per ottenere la latenza end-to-end più bassa, usa la modalità in tempo reale, che raggiunge una latenza end-to-end inferiore a un secondo nel caso peggiore e di circa 300 millisecondi nei casi più comuni. Vedi concetti di modalità in tempo reale.

Quando la modalità in tempo reale non si adatta al tuo carico di lavoro, le seguenti best practice riducono la latenza per lo streaming strutturato micro-batch:

  • Modalità di output: usa la modalità di aggiornamento se gli operatori della query e il sink la supportano. La modalità di aggiornamento emette righe aggiornate dopo ogni trigger e continua ad aggiornarle finché il watermark non scade, quindi assicurati che il tuo sink downstream sia idempotente per gestire i risultati aggiornati. Usa la modalità append per i workload che la modalità update non supporta, come i join stream-stream, o quando puoi scartare i dati che arrivano in ritardo. Non usare la modalità completa per la bassa latenza. Vedere Selezionare una modalità di output per Structured Streaming.
  • Trigger: Usa un processingTime trigger con intervallo 0 , che avvia il micro-batch successivo non appena il precedente termina e sono disponibili nuovi dati. Questo offre la latenza micro-batch più bassa, ma aumenta i costi dell'API di archiviazione cloud. Non usare AvailableNow, Once, o Continuous per carichi di lavoro operativi. Vedere Configurare gli intervalli di trigger del flusso strutturato.
  • Watermark: imposta il watermark su un valore sufficientemente elevato da includere i dati che arrivano in ritardo e che il tuo carico di lavoro non deve scartare. Il watermark controlla per quanto tempo la query accetta dati fuori ordine basati sul tempo dell’evento prima di scartarli e rimuovere lo stato, quindi un watermark troppo breve scarta silenziosamente record tardivi validi. All'interno di tale vincolo, una filigrana più corta riduce la latenza e mantiene meno stato, mentre una filigrana più lunga tollera dati più tardi a scapito di latenza e stato. Un piccolo multiplo del tuo SLA di latenza, come 2x, è un punto di partenza ragionevole per la messa a punto. Consulta Applicare filigrane per controllare le soglie di elaborazione dei dati.
  • Sorgenti e sink: leggi da fonti a bassa latenza come bus di messaggi (Apache Kafka, Amazon Kinesis, Apache Pulsar o Google Cloud Pub/Sub) oppure modifica i flussi di dati dalle tabelle Delta Lake e Apache Iceberg. Scrive in destinazioni a bassa latenza e ad elevata velocità effettiva, come bus di messaggi, database operativi o destinazioni foreach. Progettare le operazioni di sink per essere idempotenti in modo che i consumatori a valle gestiscano duplicati e dati in ritardo.
  • Stato e salvataggio dei checkpoint: Per le query con stato, usa lo state store di RocksDB, necessario sia per il salvataggio dei checkpoint del changelog sia per il salvataggio asincrono dello stato. Abilita il checkpoint del changelog per far persistere solo cambiamenti di stato incrementali. Quando il checkpointing dello stato rappresenta il collo di bottiglia nella durata dell’elaborazione batch, abilita il checkpointing asincrono dello stato per sovrapporre la scrittura dei checkpoint al micro-batch successivo, dopo aver esaminato le limitazioni relative al ripristino in caso di errore e al ridimensionamento del cluster. Assegna a ogni query la propria directory di checkpoint in un archivio cloud durevole. Vedi Configurare lo store di stato RocksDB su Azure Databricks, Creazione asincrona di checkpoint dello stato per query con stato, e checkpoint di Structured Streaming.
  • Gestione degli offset: Per ridurre la latenza dovuta al checkpoint degli offset nei flussi continui, abilita il monitoraggio asincrono dell'avanzamento, che aggiorna i log degli offset e dei commit senza bloccare l'elaborazione dei dati. Non è compatibile con i trigger AvailableNow o Once. Vedere Rilevamento asincrono dello stato di avanzamento.
  • Storage hops: Mantieni il calcolo all'interno di un'unica pipeline di streaming quando possibile. Distribuire la logica tra più job o pipeline aggiunge passaggi di archiviazione che aumentano la latenza.

Progettare carichi di lavoro di streaming per aspettarsi un errore

Databricks consiglia di configurare sempre i processi di streaming per il riavvio automatico in caso di errore. Alcune funzionalità, inclusa l'evoluzione dello schema, richiedono che i carichi di lavoro di Structured Streaming vengano ritentati automaticamente. Consultare Configurare i processi Structured Streaming per riavviare le query di streaming in caso di errore.

Alcune operazioni, ad esempio foreachBatch, forniscono garanzie di tipo almeno una volta anziché esattamente una volta. Per queste operazioni, assicurarsi che la pipeline di elaborazione sia idempotente. Vedere Usare foreachBatch per scrivere sink di dati arbitrari.

Nota

Quando una query viene riavviata, viene processato il micro-batch pianificato durante esecuzione precedente. Se il processo non è riuscito a causa di un errore di memoria insufficiente o se è stato annullato manualmente un processo a causa di un micro batch sovradimensionato, potrebbe essere necessario aumentare le prestazioni del calcolo per elaborare correttamente il micro batch.

Se si modificano le configurazioni tra le esecuzioni, queste configurazioni si applicano al primo nuovo batch pianificato. Consulta Ripristinare dopo le modifiche in una query di Structured Streaming.

Quando un processo esegue nuovi tentativi

È possibile pianificare più attività come parte di un processo di Azure Databricks. Quando si configura un processo usando il trigger continuo, non è possibile impostare dipendenze tra le attività.

È possibile scegliere di pianificare più flussi in un singolo processo usando uno degli approcci seguenti:

  • Attività multiple: Definire un processo con attività multiple che eseguono carichi di lavoro in streaming usando il trigger continuo.
  • Query multiple: definire query di streaming multiple nel codice sorgente per una singola attività.

È anche possibile combinare queste strategie. Nella tabella seguente vengono confrontati questi approcci.

Strategia Attività multiple Più richieste
Come viene condiviso il calcolo? Databricks consiglia di distribuire processi di calcolo con dimensioni appropriate per ogni attività di streaming. Facoltativamente, è possibile condividere il calcolo tra le attività. Tutte le query condividono lo stesso calcolo. Facoltativamente, è possibile assegnare query ai pool di scheduler.
Come vengono gestiti i tentativi? Tutte le attività devono fallire prima che il lavoro venga ripetuto. L'attività si ripete se qualsiasi query fallisce.

Per ulteriori informazioni su come lavorare con più attività o query, vedi Eseguire più query Structured Streaming sullo stesso cluster.

Configurare i processi di Structured Streaming per riavviare le query di streaming in caso di errore

Databricks consiglia di configurare tutti i carichi di lavoro di streaming usando il trigger continuo. Vedere Eseguire processi in modo continuo.

Il trigger continuo ha il comportamento seguente per impostazione predefinita:

  • Impedisce più di un'esecuzione simultanea dell'attività.
  • Avvia una nuova esecuzione quando un'esecuzione precedente ha esito negativo.
  • Usa il backoff esponenziale per le ripetizioni.

Databricks consiglia di usare sempre il calcolo processi anziché il calcolo multiuso per la pianificazione dei flussi di lavoro. In caso di errore del processo e di successivo tentativo, vengono distribuite nuove risorse di calcolo.

Nota

Databricks consiglia di non usare streamingQuery.awaitTermination() o spark.streams.awaitAnyTermination(). Vedere Quando usare awaitTermination().

Quando usare awaitTermination()

streamingQuery.awaitTermination() e spark.streams.awaitAnyTermination() bloccano il thread corrente fino a quando una query di streaming non termina. L'uso di queste funzioni dipende dall'ambiente di esecuzione.

Per i processi Lakeflow, non usare streamingQuery.awaitTermination() o spark.streams.awaitAnyTermination(). Queste funzioni non sono necessarie perché il servizio Processi impedisce automaticamente il completamento di un'esecuzione quando una query di streaming è attiva. Entrambe le funzioni bloccano il completamento delle celle del notebook e impediscono al servizio Lavori di monitorare la query di streaming, interrompendo così le metriche di arretrato e le notifiche dei lavori.

Usare awaitTermination() nei casi seguenti:

caso d'uso Comportamento
Notebook interattivi nel calcolo a tutti gli scopi awaitTermination() mantiene la cella in esecuzione, consente di osservare lo stato della query e garantisce che gli errori siano presenti nell'output del notebook.
Ambienti di sviluppo e locali Quando si esegue un programma Spark in locale, il processo viene chiuso al termine del thread principale. Chiamare awaitTermination() per mantenere attivo il programma fino al termine o al fallimento della query di streaming.
Propagazione degli errori nel driver Senza awaitTermination(), un errore in una query di streaming in un contesto non di lavoro potrebbe non propagarsi al thread chiamante. La query può avere esito negativo in modo invisibile all'utente, rendendo più difficile rilevare e diagnosticare gli errori. La chiamata awaitTermination() genera nuovamente l'eccezione di query sul driver.
  • Last updated on