Leggere da Pub/Sub in Dataflow

In questa pagina vengono descritte le best practice per la lettura da Pub/Sub in Dataflow.

Apache Beam fornisce un'implementazione di riferimento del connettore Pub/Sub I/O da utilizzare con i runner non Dataflow. Tuttavia, il runner Dataflow utilizza una propria implementazione personalizzata del connettore. Questa implementazione sfrutta Google Cloudle API e i servizi interni per offrire watermark a bassa latenza, elevata accuratezza del watermark e 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 consegna 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 consegna dei messaggi at-least-once. Per ottenere la semantica "at-least-once", se Pub/Sub non riceve la conferma dal sottoscrittore entro il termine di conferma, ritenta la consegna del messaggio. I tentativi possono verificarsi anche prima del termine di conferma o dopo che un messaggio è stato confermato.

Dataflow conferma i messaggi dopo che sono stati elaborati correttamente dalla prima fase unita e gli effetti collaterali di questa elaborazione sono stati scritti nello spazio di archiviazione permanente. Per ridurre il numero di messaggi duplicati, Dataflow estende continuamente il termine di conferma mentre un batch di messaggi viene elaborato in questa fase.

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

Se la pipeline può tollerare alcuni record duplicati, valuta la possibilità di utilizzare la modalità di flusso "at-least-once" invece. Questa modalità può ridurre significativamente la latenza e il costo totale della pipeline. Il compromesso è che i messaggi duplicati potrebbero essere elaborati due volte. Per ulteriori informazioni, consulta Scegliere la modalità di flusso da utilizzare.

Deduplicare per attributo del messaggio

Per impostazione predefinita, Dataflow deduplica in base all'ID messaggio. 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 oppure l'applicazione potrebbe pubblicare erroneamente lo stesso messaggio due volte. Quest'ultima situazione può verificarsi a causa dei tentativi, se la conferma è stata eliminata a causa di problemi di rete o altre interruzioni. In questi casi, 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 Pub/Sub I/O 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 tentativi, Dataflow può rilevare i duplicati. Per la deduplicazione, i messaggi devono essere pubblicati su Pub/Sub entro 10 minuti l'uno dall'altro.

Per ulteriori informazioni sull'utilizzo degli attributi ID, consulta i seguenti argomenti di riferimento dell'SDK:

Abbonamenti

Quando configuri la pipeline, specifichi un argomento Pub/Sub o un abbonamento Pub/Sub da cui leggere. Se specifichi un abbonamento, non utilizzare lo stesso abbonamento Pub/Sub per più pipeline. Se due pipeline leggono da un singolo abbonamento, 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 un abbonamento separato per ogni pipeline.

Se specifichi un argomento, il connettore crea un nuovo abbonamento temporaneo. Questo abbonamento è univoco per pipeline.

Timestamp e watermark

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

Puoi configurare il connettore per leggere il timestamp dell'evento da un attributo del messaggio Pub/Sub. In questo caso, il connettore utilizza il timestamp dell'evento per il watermark. In caso contrario, per impostazione predefinita utilizza il timestamp del messaggio Pub/Sub.

Per ulteriori informazioni 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 un abbonamento. Questa API fornisce una latenza inferiore rispetto a quella disponibile in Cloud Monitoring. Consente a Dataflow di avanzare i watermark della pipeline ed emettere risultati di calcolo con finestra con latenze ridotte.

Se configuri il connettore per utilizzare i timestamp degli eventi, Dataflow crea un secondo abbonamento Pub/Sub, chiamato abbonamento di monitoraggio. Dataflow utilizza l'abbonamento di monitoraggio per ispezionare gli orari degli eventi dei messaggi ancora nel backlog. Questo approccio consente a Dataflow di stimare con precisione il backlog dell'orario degli eventi. Il service account del worker deve disporre almeno delle seguenti autorizzazioni per il progetto che contiene l' abbonamento di monitoraggio:

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

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

Per ulteriori informazioni sui watermark, consulta la pagina di StackOverflow che illustra come Dataflow calcola i watermark 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 basate 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ò comportare 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 Pub/Sub Seek, ti consigliamo di seguire questo flusso di lavoro:

  1. Crea uno snapshot dell' abbonamento.
  2. Crea un nuovo abbonamento per l'argomento Pub/Sub. Il nuovo abbonamento eredita lo snapshot.
  3. Esegui il drain o annulla il job Dataflow corrente.
  4. Invia di nuovo la pipeline utilizzando il nuovo abbonamento.

Per ulteriori informazioni, consulta Rielaborazione dei messaggi con snapshot e ricerca Pub/Sub.

Parallelismo dell'origine Pub/Sub

L'origine Pub/Sub assegna a ogni messaggio una chiave deterministica per l'elaborazione e utilizza queste chiavi per rimescolare i messaggi. Per i job Streaming Engine, vengono utilizzate 1024 chiavi per il rimescolamento. Per i job non Streaming Engine, il numero di chiavi è la potenza di 2 più bassa maggiore di (4 * maximum workers).

Per sostituire il numero predefinito di chiavi di rimescolamento, imposta l'num_pubsub_keys opzione di servizio:

Java 

--dataflowServiceOptions=num_pubsub_keys=NUMBER_OF_KEYS

Python 

--dataflow_service_options=num_pubsub_keys=NUMBER_OF_KEYS

Go 

--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, puoi impostare questa opzione nelle seguenti situazioni:

Se imposti questa opzione, tieni presente 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à Pub/Sub non sono supportate nell'implementazione del connettore Pub/Sub I/O del runner Dataflow.

Backoff esponenziale

Quando crei un abbonamento Pub/Sub, puoi configurarlo in modo che utilizzi un criterio di ripetizione con backoff esponenziale. Tuttavia, il backoff esponenziale non funziona con Dataflow. Crea invece l'abbonamento con il criterio di ripetizione Riprova immediatamente.

Il backoff esponenziale viene attivato da una conferma negativa o alla scadenza del termine di conferma. Tuttavia, Dataflow non invia conferme negative quando il codice della pipeline non riesce. Al contrario, ritenta l'elaborazione dei messaggi a tempo indeterminato, estendendo continuamente il termine di conferma per il messaggio.

Argomenti messaggi non recapitabili

Non utilizzare gli argomenti messaggi non recapitabili di Pub/Sub 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 unita. Se la pipeline ha più fasi unite e si verificano errori in qualsiasi punto dopo la prima fase, i messaggi vengono già confermati e non vengono inviati all'argomento messaggi non recapitabili.

Implementa invece il pattern dei messaggi non recapitabili in modo esplicito nella pipeline, indirizzando i messaggi non riusciti a una destinazione per l'elaborazione successiva. Alcuni sink 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 ha i propri meccanismi per l'elaborazione "exactly-once" , non è consigliabile utilizzare la consegna "exactly-once" di Pub/Sub con Dataflow. L'abilitazione della consegna "exactly-once" di Pub/Sub riduce il rendimento 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 Pub/Sub I/O potrebbe non conservare l'ordinamento dei messaggi.
  • Apache Beam non definisce linee guida rigorose in merito all'ordine in cui vengono elaborati gli 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 il rendimento.

Trasformazioni di singoli messaggi Pub/Sub

Le trasformazioni di singoli messaggi (SMT) consentono di manipolare, convalidare e filtrare i messaggi in base ai relativi attributi o dati durante lo streaming nel sistema. Gli abbonamenti che alimentano Dataflow non devono utilizzare SMT che filtrano i messaggi, in quanto possono interferire con la scalabilità automatica. Ciò si verifica perché il filtraggio SMT dell'abbonamento può far sembrare il backlog più grande di quello consegnato a Dataflow finché i messaggi filtrati non vengono effettivamente elaborati dall'SMT. Gli SMT degli argomenti che filtrano i messaggi non causano problemi di scalabilità automatica.

Passaggi successivi