Leggere da Pub/Sub in Dataflow

Questa pagina descrive le best practice per la lettura da Pub/Sub in Dataflow.

Apache Beam fornisce un'implementazione di riferimento del connettore I/O Pub/Sub per l'utilizzo da parte di runner non Dataflow. Tuttavia, il runner Dataflow utilizza la propria implementazione personalizzata del connettore. Questa implementazione sfrutta le API e i servizi interni di Google Cloudper offrire watermark a bassa latenza, un'elevata precisione dei watermark e una deduplicazione efficiente per l'elaborazione dei messaggi "exactly-once". Il connettore è disponibile per Java, Python e Go.

Elaborazione "exactly-once"

Pub/Sub disaccoppia i publisher di eventi dai consumer di eventi. L'applicazione pubblica i messaggi in un argomento e Pub/Sub li recapita in modo asincrono ai sottoscrittori.

Pub/Sub assegna un ID messaggio univoco a ogni messaggio pubblicato correttamente in un argomento. Per impostazione predefinita, Pub/Sub esegue la distribuzione dei messaggi at-least-once. Per ottenere la semantica "at-least-once", se Pub/Sub non riceve la conferma dal sottoscrittore entro la scadenza della conferma, riprova a consegnare il messaggio. I nuovi tentativi potrebbero verificarsi anche prima della scadenza della conferma o dopo la conferma di un messaggio.

Dataflow conferma i messaggi dopo che sono stati elaborati correttamente dalla prima fase fusa e gli effetti collaterali di questa elaborazione sono stati scritti nell'archiviazione permanente. Per ridurre il numero di messaggi duplicati, Dataflow estende continuamente la scadenza della conferma mentre un gruppo di messaggi viene elaborato in questa fase.

Poiché Pub/Sub potrebbe riconsegnare un messaggio, è possibile che nella pipeline arrivino messaggi duplicati. Se la pipeline Dataflow utilizza la modalità flusso "exactly-once", Dataflow deduplica questi messaggi per ottenere la semantica "exactly-once".

Se la tua pipeline può tollerare alcuni record duplicati, valuta la possibilità di utilizzare la modalità di flussi di dati at-least-once. Questa modalità può ridurre notevolmente la latenza e il costo totale della pipeline. Il compromesso è che i messaggi duplicati potrebbero essere elaborati due volte. Per saperne di più, consulta Scegli la modalità di streaming da utilizzare.

Deduplica per attributo del messaggio

Per impostazione predefinita, Dataflow deduplica i messaggi in base al relativo ID. Tuttavia, un'applicazione potrebbe inviare lo stesso record due volte come due messaggi Pub/Sub distinti. Ad esempio, i dati di origine originali potrebbero contenere record duplicati o l'applicazione potrebbe pubblicare erroneamente lo stesso messaggio due volte. Quest'ultimo caso può verificarsi a causa di nuovi tentativi, se la conferma è stata ignorata a causa di problemi di rete o altre interruzioni. In queste situazioni, i messaggi duplicati hanno ID messaggio diversi.

A seconda dello scenario, i dati potrebbero contenere un campo univoco che può essere utilizzato per la deduplicazione. Ad esempio, i record potrebbero contenere un ID transazione univoco. Puoi configurare il connettore di I/O Pub/Sub per deduplicare i messaggi in base al valore di un attributo del messaggio, anziché utilizzare l'ID messaggio Pub/Sub. Se il publisher imposta questo attributo in modo coerente durante i nuovi tentativi, Dataflow può rilevare i duplicati. I messaggi devono essere pubblicati in Pub/Sub entro 10 minuti l'uno dall'altro per la deduplicazione.

Per saperne di più sull'utilizzo degli attributi ID, consulta i seguenti argomenti di riferimento dell'SDK:

Abbonamenti

Quando configuri la pipeline, specifichi un argomento o una sottoscrizione Pub/Sub da cui leggere. Se specifichi una sottoscrizione, non utilizzare la stessa sottoscrizione Pub/Sub per più pipeline. Se due pipeline leggono da una singola sottoscrizione, ogni pipeline riceve parte dei dati in modo non deterministico, il che potrebbe causare messaggi duplicati, ritardo del watermark e scalabilità automatica inefficiente. Crea invece una sottoscrizione separata per ogni pipeline.

Se specifichi un argomento, il connettore crea una nuova sottoscrizione temporanea. Questa sottoscrizione è univoca per ogni pipeline.

Timestamp e filigrane

Tutti i messaggi Pub/Sub hanno un timestamp, che rappresenta l'ora in cui Pub/Sub riceve il messaggio. I dati potrebbero anche avere un timestamp dell'evento, ovvero l'ora in cui il record è stato generato dall'origine.

Puoi configurare il connettore in modo che legga il timestamp dell'evento da un attributo nel messaggio Pub/Sub. In questo caso, il connettore utilizza il timestamp dell'evento per la filigrana. In caso contrario, per impostazione predefinita utilizza il timestamp del messaggio Pub/Sub.

Per saperne di più sull'utilizzo dei timestamp degli eventi, consulta i seguenti argomenti di riferimento dell'SDK:

Il connettore Pub/Sub ha accesso all'API privata di Pub/Sub, che fornisce l'età del messaggio non confermato più vecchio in una sottoscrizione. Questa API offre una latenza inferiore rispetto a quella disponibile in Cloud Monitoring. Consente a Dataflow di far avanzare i watermark della pipeline ed emettere risultati di calcolo con finestra con latenze ridotte.

Se configuri il connettore in modo che utilizzi i timestamp degli eventi, Dataflow crea una seconda sottoscrizione Pub/Sub, chiamata sottoscrizione di monitoraggio. Dataflow utilizza la sottoscrizione di monitoraggio per ispezionare i timestamp degli eventi dei messaggi ancora in attesa. Questo approccio consente a Dataflow di stimare con precisione il backlog del tempo dell'evento. Il service account worker deve disporre almeno delle seguenti autorizzazioni per il progetto che contiene la sottoscrizione di monitoraggio:

  • pubsub.subscriptions.create
  • pubsub.subscriptions.consume
  • pubsub.subscriptions.delete

Inoltre, deve disporre dell'autorizzazione pubsub.topics.attachSubscription sull'argomento Pub/Sub. Ti consigliamo di creare un ruolo Identity and Access Management personalizzato che contenga solo queste autorizzazioni.

Per saperne di più sulle watermark, consulta la pagina di StackOverflow che spiega come Dataflow calcola le watermark di Pub/Sub.

Se una pipeline ha più origini Pub/Sub e una di queste ha un volume molto basso o è inattiva, ritarda l'avanzamento dell'intero watermark, aumentando la latenza complessiva della pipeline. Se nella pipeline sono presenti timer o aggregazioni di finestre basati sul watermark, anche questi vengono ritardati.

Pub/Sub Seek

Pub/Sub Seek consente agli utenti di riprodurre i messaggi confermati in precedenza. Puoi utilizzare Pub/Sub Seek con Dataflow per rielaborare i messaggi in una pipeline.

Tuttavia, non è consigliabile utilizzare Pub/Sub Seek in una pipeline in esecuzione. La ricerca all'indietro in una pipeline in esecuzione può causare la duplicazione o l'eliminazione dei messaggi. Inoltre, invalida la logica del watermark di Dataflow ed entra in conflitto con lo stato di una pipeline che incorpora i dati elaborati.

Per rielaborare i messaggi utilizzando la funzionalità di ricerca di Pub/Sub, ti consigliamo il seguente flusso di lavoro:

  1. Crea uno snapshot della sottoscrizione.
  2. Crea una nuova sottoscrizione per l'argomento Pub/Sub. La nuova sottoscrizione eredita lo snapshot.
  3. Svuota o annulla il job Dataflow corrente.
  4. Invia nuovamente la pipeline utilizzando la nuova sottoscrizione.

Per saperne di più, consulta Nuova elaborazione dei messaggi con Pub/Sub Snapshot e Seek.

Parallelismo dell'origine Pub/Sub

L'origine Pub/Sub assegna a ogni messaggio una chiave deterministica per l'elaborazione e utilizza queste chiavi per eseguire lo shuffling dei messaggi.

  • Per i job Streaming Engine, Dataflow calcola il numero di chiavi in modo dinamico in base ai worker attivi in un determinato istante: (number of workers * threads per worker) / 2. Il valore calcolato è limitato a un intervallo compreso tra 128 e 65.536 chiavi, quindi arrotondato per difetto alla potenza di 2 più vicina (la potenza di 2 più alta inferiore o uguale al valore calcolato).

  • Per i job non di Streaming Engine, il numero di chiavi è la potenza di 2 più bassa maggiore o uguale a (4 * maximum workers).

Per eseguire l'override del numero predefinito di chiavi di shuffling, imposta l'opzione di servizio num_pubsub_keys:

Java 

--dataflowServiceOptions=num_pubsub_keys=NUMBER_OF_KEYS

Python 

--dataflow_service_options=num_pubsub_keys=NUMBER_OF_KEYS

Vai 

--dataflow_service_options=num_pubsub_keys=NUMBER_OF_KEYS

Sostituisci NUMBER_OF_KEYS con il numero di chiavi. Viene utilizzata la potenza di 2 successiva maggiore o uguale al valore specificato.

Ad esempio, potresti impostare questa opzione nelle seguenti situazioni:

Se imposti questa opzione, considera i compromessi descritti in Parallelizzazione e distribuzione.

Non puoi modificare il numero di chiavi nell'ambito di un aggiornamento della pipeline. Per modificare il numero di chiavi per un job della pipeline esistente, devi avviare un nuovo job.

Funzionalità Pub/Sub non supportate

Le seguenti funzionalità di Pub/Sub non sono supportate nell'implementazione del connettore I/O Pub/Sub del runner Dataflow.

Backoff esponenziale

Quando crei una sottoscrizione Pub/Sub, puoi configurarla in modo che utilizzi una policy di ripetizione con backoff esponenziale. Tuttavia, il backoff esponenziale non funziona con Dataflow. Crea invece la sottoscrizione con la policy di ripetizione Riprova immediatamente.

Il backoff esponenziale viene attivato da una conferma negativa o alla scadenza della conferma. Tuttavia, Dataflow non invia conferme negative quando il codice della pipeline non va a buon fine. Invece, riprova l'elaborazione dei messaggi a tempo indeterminato, estendendo continuamente la scadenza di conferma per il messaggio.

Argomenti messaggi non recapitabili

Non utilizzare gli argomenti Pub/Sub per i messaggi non recapitabili con Dataflow per i seguenti motivi:

  • Dataflow invia conferme negative per vari motivi interni (ad esempio, se un worker è in fase di arresto). Di conseguenza, i messaggi potrebbero essere recapitati all'argomento messaggi non recapitabili anche se non si verificano errori nel codice della pipeline.

  • Dataflow conferma i messaggi dopo che un bundle di messaggi è stato elaborato correttamente dalla prima fase fusa. Se la pipeline ha più fasi fuse e si verificano errori in qualsiasi momento dopo la prima fase, i messaggi sono già stati confermati e non vengono inviati all'argomento messaggi non recapitabili.

Implementa invece esplicitamente il pattern dei messaggi non recapitabili nella pipeline, indirizzando i messaggi non riusciti a una destinazione per l'elaborazione successiva. Alcuni sink di I/O hanno il supporto integrato per le code di messaggi non recapitabili. I seguenti esempi implementano pattern di messaggi non recapitabili:

Consegna "exactly-once" di Pub/Sub

Poiché Dataflow dispone di meccanismi propri per l'elaborazione "exactly-once", non è consigliabile utilizzare la consegna "exactly-once" di Pub/Sub con Dataflow. L'attivazione della distribuzione exactly-once di Pub/Sub riduce le prestazioni della pipeline, perché limita il numero di messaggi disponibili per l'elaborazione parallela.

Ordinamento dei messaggi Pub/Sub

L'ordinamento dei messaggi è una funzionalità di Pub/Sub che consente a un sottoscrittore di ricevere i messaggi nell'ordine in cui sono stati pubblicati.

Non è consigliabile utilizzare l'ordinamento dei messaggi con Dataflow per i seguenti motivi:

  • Il connettore I/O Pub/Sub potrebbe non conservare l'ordine dei messaggi.
  • Apache Beam non definisce linee guida rigide per l'ordine di elaborazione degli elementi. Pertanto, l'ordinamento potrebbe non essere conservato nelle trasformazioni downstream.
  • L'utilizzo dell'ordinamento dei messaggi Pub/Sub con Dataflow può aumentare la latenza e ridurre le prestazioni.

Trasformazioni di singoli messaggi Pub/Sub

Le trasformazioni di singoli messaggi (SMT) consentono di manipolare, convalidare e filtrare i messaggi in base ai loro attributi o dati durante lo streaming nel sistema. Le sottoscrizioni che alimentano Dataflow non devono utilizzare SMT che filtrano i messaggi, in quanto ciò può interferire con la scalabilità automatica. Questo accade perché il filtro SMT della sottoscrizione può far apparire il backlog più grande di quello consegnato a Dataflow finché i messaggi filtrati non vengono effettivamente elaborati dall'SMT. Le SMT degli argomenti che filtrano i messaggi non causano problemi con la scalabilità automatica.

Passaggi successivi