Exactly-once in Dataflow

Dataflow supporta l'elaborazione "exactly-once" dei record. Questa pagina spiega come Dataflow implementa l'elaborazione "exactly-once" garantendo al contempo una bassa latenza.

Panoramica

Le pipeline batch utilizzano sempre l'elaborazione "exactly-once". Le pipeline di flusso utilizzano l'elaborazione "exactly-once" per impostazione predefinita, ma possono anche utilizzare l'elaborazione "at-least-once".

L'elaborazione "exactly-once" fornisce garanzie sui risultati dell'elaborazione dei record, inclusi i risultati di ogni fase della pipeline. Nello specifico, per ogni record che arriva alla pipeline da un'origine o che arriva a una fase da una fase precedente, Dataflow garantisce quanto segue:

  • Il record viene elaborato e non viene perso.
  • I risultati dell'elaborazione che rimangono all'interno della pipeline vengono riflessi al massimo una volta.

In altre parole, i record vengono elaborati almeno una volta e i risultati vengono sottoposti a commit esattamente una volta.

L'elaborazione "exactly-once" garantisce l'accuratezza dei risultati, senza record duplicati nell'output. Dataflow è ottimizzato per ridurre al minimo la latenza mantenendo la semantica "exactly-once". Tuttavia, l'elaborazione "exactly-once" comporta comunque un costo per eseguire la deduplicazione. Per i casi d'uso che possono tollerare i record duplicati, spesso puoi ridurre i costi e migliorare la latenza attivando la modalità "at-least-once". Per ulteriori informazioni sulla scelta tra flusso "exactly-once" e "at-least-once" in streaming, consulta Impostare la modalità di flusso della pipeline.

Dati in ritardo

L'elaborazione "exactly-once" garantisce l'accuratezza della pipeline: se la pipeline elabora un record, Dataflow garantisce che il record venga riflesso nell'output e che non venga duplicato.

In una pipeline di flusso, tuttavia, l'elaborazione "exactly-once" non può garantire che i risultati siano completi, perché i record potrebbero arrivare in ritardo. Supponiamo, ad esempio, che la pipeline esegua un'aggregazione su una finestra temporale, ad esempio Count. Con l'elaborazione "exactly-once", il risultato è accurato per i record che arrivano entro la finestra in modo tempestivo, ma i record in ritardo potrebbero essere eliminati.

In genere, non è possibile garantire la completezza in una pipeline di flusso, perché in teoria i record possono arrivare in ritardo in modo arbitrario. Nel caso limite, dovresti attendere per sempre per produrre un risultato. In modo più pratico, Apache Beam ti consente di configurare la soglia per l'eliminazione dei dati in ritardo e quando emettere i risultati aggregati. Per ulteriori informazioni, consulta Filigrane e dati in ritardo nella documentazione di Apache Beam.

Effetto collaterale

Non è garantito che gli effetti collaterali abbiano una semantica "exactly-once". È importante sottolineare che ciò include la scrittura dell'output in uno store esterno, a meno che anche il sink non implementi la semantica "exactly-once".

Nello specifico, Dataflow non garantisce che ogni record passi attraverso ogni trasformazione esattamente una volta. A causa di nuovi tentativi o errori dei worker, Dataflow potrebbe inviare un record a una trasformazione più volte o anche contemporaneamente su più worker.

Nell'ambito dell'elaborazione "exactly-once", Dataflow deduplica gli output. Tuttavia, se il codice in una trasformazione ha effetti collaterali, questi effetti potrebbero verificarsi più volte. Ad esempio, se una trasformazione effettua una chiamata a un servizio remoto, questa chiamata potrebbe essere effettuata più volte per lo stesso record. In alcune situazioni, gli effetti collaterali possono persino causare la perdita di dati. Supponiamo, ad esempio, che una trasformazione legga un file per produrre l'output e poi lo elimini immediatamente senza attendere il commit dell'output. Se si verifica un errore durante il commit del risultato, Dataflow riprova la trasformazione, ma ora la trasformazione non può leggere il file eliminato.

Logging

L'output dei log dell'elaborazione indica che l'elaborazione è avvenuta, ma non indica se i dati sono stati sottoposti a commit. Pertanto, i file di log potrebbero indicare che i dati sono stati elaborati più volte, anche se i risultati dei dati elaborati vengono sottoposti a commit nello spazio di archiviazione permanente una sola volta. Inoltre, i log non riflettono sempre i dati elaborati e sottoposti a commit. I log potrebbero essere eliminati a causa della limitazione o persi a causa di altri problemi del servizio di logging.

Flusso "exactly-once"

Questa sezione spiega come Dataflow implementa l'elaborazione "exactly-once" per i job di flusso, incluso come Dataflow gestisce le complessità come l'elaborazione non deterministica, i dati in ritardo e il codice personalizzato.

Shuffle di flusso di Dataflow

I job di flusso di Dataflow vengono eseguiti in parallelo su molti worker diversi assegnando intervalli di lavoro a ogni worker. Sebbene le assegnazioni possano cambiare nel tempo in risposta a errori dei worker, scalabilità automatica o altri eventi, dopo ogni GroupByKey trasformazione , tutti i record con la stessa chiave vengono elaborati sullo stesso worker. La trasformazione GroupByKey viene spesso utilizzata dalle trasformazioni composite, come Count o FileIO. Per garantire che i record per una determinata chiave finiscano sullo stesso worker, i worker di Dataflow eseguono lo shuffle dei dati tra loro utilizzando chiamate di procedura remota (RPC).

Per assicurarsi che i record non vengano persi durante lo shuffle, Dataflow utilizza il backup upstream. Con il backup upstream, il worker che invia i record riprova le RPC finché non riceve una conferma positiva della ricezione del record. Gli effetti collaterali dell'elaborazione del record vengono sottoposti a commit nello spazio di archiviazione permanente downstream. Se il worker che invia i record non è più disponibile, Dataflow continua a riprovare le RPC, il che garantisce che ogni record venga consegnato almeno una volta.

Poiché questi nuovi tentativi potrebbero creare duplicati, ogni messaggio viene taggato con un ID univoco. Ogni destinatario memorizza un catalogo di tutti gli ID già visualizzati ed elaborati. Quando viene ricevuto un record, Dataflow cerca il suo ID nel catalogo. Se l'ID viene trovato, il record è già stato ricevuto e sottoposto a commit e viene eliminato come duplicato. Per garantire la stabilità degli ID record, ogni output da un passaggio all'altro viene sottoposto a checkpoint nello spazio di archiviazione. Di conseguenza, se lo stesso messaggio viene inviato più volte a causa di chiamate RPC ripetute, il messaggio viene sottoposto a commit nello spazio di archiviazione una sola volta.

Garantire una bassa latenza

Affinché l'elaborazione "exactly-once" sia fattibile, l'I/O deve essere ridotto, in particolare impedendo l'I/O su ogni record. Per raggiungere questo obiettivo, Dataflow utilizza i filtri Bloom e la garbage collection.

Filtri Bloom

I filtri Bloom sono strutture di dati compatte che consentono di eseguire rapidamente i controlli di appartenenza al set. In Dataflow, ogni worker mantiene un filtro Bloom di ogni ID visualizzato. Quando arriva un nuovo ID record, il worker cerca l'ID nel filtro. Se il filtro restituisce false, questo record non è un duplicato e il worker non cerca l'ID nello spazio di archiviazione stabile.

Dataflow mantiene un set di filtri Bloom rotanti suddivisi per tempo. Quando arriva un record, Dataflow seleziona il filtro appropriato da controllare in base al timestamp di sistema. Questo passaggio impedisce la saturazione dei filtri Bloom man mano che i filtri vengono sottoposti a garbage collection e limita anche la quantità di dati da scansionare all'avvio.

Garbage collection

Per evitare di riempire lo spazio di archiviazione con gli ID record, Dataflow utilizza la garbage collection per rimuovere i record precedenti. Dataflow utilizza il timestamp di sistema per calcolare una filigrana di garbage collection.

Questa filigrana si basa sulla quantità di tempo fisico trascorso in attesa in una determinata fase. Pertanto, fornisce anche informazioni sulle parti della pipeline che sono lente. Questi metadati sono la base per la metrica di ritardo del sistema mostrata nell' interfaccia di monitoraggio di Dataflow.

Se arriva un record con un timestamp precedente alla filigrana e se gli ID per questo periodo sono già stati sottoposti a garbage collection, il record viene ignorato. Poiché la filigrana bassa che attiva la garbage collection non avanza finché le consegne dei record non vengono riconosciute, questi record in ritardo sono duplicati.

Origini non deterministiche

Dataflow utilizza l'SDK Apache Beam per leggere i dati nelle pipeline. Se l'elaborazione non riesce, Dataflow potrebbe riprovare a leggere da un'origine. In questa situazione, Dataflow deve assicurarsi che ogni record univoco prodotto da un'origine venga registrato esattamente una volta. Per le origini deterministiche, come Pub/Sub Lite o Kafka, i record vengono letti in base a un offset registrato, riducendo la necessità di questo passaggio.

Poiché Dataflow non può assegnare automaticamente gli ID record, le origini non deterministiche devono comunicare a Dataflow quali sono gli ID record per evitare la duplicazione. Quando un'origine fornisce ID univoci per ogni record, il connettore utilizza uno shuffle nella pipeline per rimuovere i duplicati. I record con lo stesso ID vengono filtrati. Per un esempio di come Dataflow implementa l'elaborazione "exactly-once" quando si utilizza Pub/Sub come origine, consulta la sezione Elaborazione "exactly-once" nella pagina Flusso con Pub/Sub.

Quando esegui DoFn personalizzati come parte della pipeline, Dataflow non garantisce che questo codice venga eseguito una sola volta per record. Per garantire l'elaborazione "at-least-once" in caso di errori dei worker, Dataflow potrebbe eseguire un determinato record tramite una trasformazione più volte oppure potrebbe eseguire lo stesso record contemporaneamente su più worker. Se includi codice nella pipeline che esegue operazioni come contattare un servizio esterno, le azioni potrebbero essere eseguite più volte per un determinato record.

Per rendere l'elaborazione non deterministica effettivamente deterministica, utilizza il checkpoint. Quando utilizzi il checkpoint, ogni output di una trasformazione viene sottoposto a checkpoint nello spazio di archiviazione stabile con il suo ID univoco prima di essere consegnato alla fase successiva. I nuovi tentativi nella consegna dello shuffle di Dataflow inoltrano l'output sottoposto a checkpoint. Sebbene il codice possa essere eseguito più volte, Dataflow garantisce che venga memorizzato l'output di una sola di queste esecuzioni. Dataflow utilizza uno store coerente che impedisce la scrittura di duplicati nello spazio di archiviazione stabile.

Consegna dell'output "exactly-once"

L'SDK Apache Beam include sink integrati progettati per garantire che non producano duplicati. Quando possibile, utilizza uno di questi sink integrati.

Se devi scrivere il tuo sink, l'approccio migliore è rendere l'oggetto funzione idempotente in modo che possa essere riprovato tutte le volte che è necessario senza causare effetti collaterali indesiderati. Tuttavia, spesso alcuni componenti della trasformazione che implementano la funzionalità del sink non sono deterministici e potrebbero cambiare se vengono riprovati.

Ad esempio, in un'aggregazione con finestra, il set di record nella finestra potrebbe non essere deterministico. Nello specifico, la finestra potrebbe tentare di attivarsi con gli elementi e0, e1, e2. Il worker potrebbe arrestarsi in modo anomalo prima di eseguire il commit dell'elaborazione della finestra, ma non prima che questi elementi vengano inviati come effetto collaterale. Quando il worker viene riavviato, la finestra si attiva di nuovo e arriva un elemento in ritardo e3. Poiché questo elemento arriva prima del commit della finestra, non viene conteggiato come dati in ritardo, quindi DoFn viene chiamato di nuovo con gli elementi e0, e1, e2, e3. Questi elementi vengono quindi inviati all'operazione di effetto collaterale. L'idempotenza non è utile in questo scenario, perché ogni volta vengono inviati set di record logici diversi.

Per risolvere la non determinazione in Dataflow, utilizza la trasformazione Reshuffle integrata. Quando Dataflow esegue lo shuffle dei dati, li scrive in modo durevole in modo che gli elementi generati in modo non deterministico siano stabili se le operazioni vengono riprovate dopo lo shuffle. L'utilizzo della trasformazione Reshuffle contribuisce a garantire che solo una versione dell'output di DoFn possa superare un limite di shuffle. Il seguente pattern garantisce che l'operazione di effetto collaterale riceva sempre un record deterministico da generare come output:

c.apply(Window.<..>into(FixedWindows.of(Duration.standardMinutes(1))))
 .apply(GroupByKey.<..>.create())
 .apply(new PrepareOutputData())
 .apply(Reshuffle.<..>of())
 .apply(WriteToSideEffect());

Per assicurarti che il runner di Dataflow sappia che gli elementi devono essere stabili prima di eseguire un DoFn, aggiungi l' RequiresStableInput annotazione a DoFn.

Scopri di più