Questa pagina fornisce una panoramica del ciclo di vita della pipeline, dal codice della pipeline a un job Dataflow.
Questa pagina spiega i seguenti concetti:
- Cos'è un grafico di esecuzione e come una pipeline Apache Beam diventa un job Dataflow
- Come Dataflow gestisce gli errori
- Come Dataflow parallelizza e distribuisce automaticamente la logica di elaborazione nella pipeline ai worker che eseguono il job
- Ottimizzazioni dei job che Dataflow potrebbe apportare
Grafico di esecuzione
Quando esegui la pipeline Dataflow, Dataflow crea un grafico di esecuzione dal codice che costruisce l'oggetto Pipeline, incluse tutte le trasformazioni e le relative funzioni di elaborazione, come gli oggetti DoFn. Questo è il grafico di esecuzione della pipeline e la fase è chiamata
tempo di costruzione del grafico.
Durante la costruzione del grafico, Apache Beam esegue localmente il codice dal punto di ingresso principale del codice della pipeline, arrestandosi alle chiamate a un'origine, un sink o un passaggio di trasformazione e trasformando queste chiamate in nodi del grafico.
Di conseguenza, una parte di codice nel punto di ingresso di una pipeline (metodo main di Java e Go o livello superiore di uno script Python) viene eseguita localmente sulla macchina che esegue la pipeline. Lo stesso codice dichiarato in un metodo di un oggetto DoFn viene eseguito nei worker di Dataflow.
Ad esempio, l'esempio WordCount incluso negli SDK Apache Beam contiene una serie di trasformazioni per leggere, estrarre, contare, formattare e scrivere le singole parole in una raccolta di testo, insieme a un conteggio delle occorrenze per ogni parola. Il seguente diagramma mostra come le trasformazioni nella pipeline WordCount vengono espanse in un grafico di esecuzione:
Figura 1: grafico di esecuzione dell'esempio WordCount
Il grafico di esecuzione spesso differisce dall'ordine in cui hai specificato le trasformazioni durante la costruzione della pipeline. Questa differenza esiste perché il servizio Dataflow esegue varie ottimizzazioni e fusioni sul grafico di esecuzione prima che venga eseguito sulle risorse cloud gestite. Il servizio Dataflow rispetta le dipendenze dei dati durante l'esecuzione della pipeline. Tuttavia, i passaggi senza dipendenze dei dati tra loro potrebbero essere eseguiti in qualsiasi ordine.
Per visualizzare il grafico di esecuzione non ottimizzato generato da Dataflow per la pipeline, seleziona il job nell'interfaccia di monitoraggio di Dataflow. Per ulteriori informazioni sulla visualizzazione dei job, consulta Utilizzare l'interfaccia di monitoraggio di Dataflow.
Durante la costruzione del grafico, Apache Beam verifica che tutte le risorse a cui fa riferimento la pipeline, come i bucket Cloud Storage, le tabelle BigQuery e gli argomenti o gli abbonamenti Pub/Sub, esistano effettivamente e siano accessibili. La convalida viene eseguita tramite chiamate API standard ai rispettivi servizi, pertanto è fondamentale che l'account utente utilizzato per eseguire una pipeline abbia una connettività corretta ai servizi necessari e sia autorizzato a chiamare le API dei servizi. Prima di inviare la pipeline al servizio Dataflow, Apache Beam verifica anche la presenza di altri errori e si assicura che il grafico della pipeline non contenga operazioni illegali.
Il grafico di esecuzione viene quindi tradotto in formato JSON e il grafico di esecuzione JSON viene trasmesso all'endpoint del servizio Dataflow.
Il servizio Dataflow convalida quindi il grafico di esecuzione JSON. Una volta convalidato, il grafico diventa un job nel servizio Dataflow. Puoi visualizzare il job, il relativo grafico di esecuzione, lo stato e le informazioni sui log utilizzando l'interfaccia di monitoraggio di Dataflow.
Java
Il servizio Dataflow invia una risposta alla macchina in cui esegui il programma Dataflow. Questa risposta è incapsulata nell'oggetto DataflowPipelineJob, che contiene il jobId del job Dataflow.
Utilizza jobId per monitorare, tenere traccia e risolvere i problemi del job utilizzando
l'interfaccia di monitoraggio di Dataflow
e l'interfaccia a riga di comando di Dataflow.
Per ulteriori informazioni, consulta il
riferimento API per DataflowPipelineJob.
Python
Il servizio Dataflow invia una risposta alla macchina in cui esegui il programma Dataflow. Questa risposta è incapsulata nell'oggetto DataflowPipelineResult, che contiene il job_id del job Dataflow.
Utilizza il job_id per monitorare, tenere traccia e risolvere i problemi del job
utilizzando l'
interfaccia di monitoraggio di Dataflow
e l'
interfaccia a riga di comando di Dataflow.
Vai
Il servizio Dataflow invia una risposta alla macchina in cui esegui il programma Dataflow. Questa risposta è incapsulata nell'oggetto dataflowPipelineResult, che contiene il jobID del job Dataflow.
Utilizza il jobID per monitorare, tenere traccia e risolvere i problemi del job
utilizzando l'
interfaccia di monitoraggio di Dataflow
e l'
interfaccia a riga di comando di Dataflow.
La costruzione del grafico avviene anche quando esegui la pipeline localmente, ma il grafico non viene tradotto in JSON o trasmesso al servizio. Il grafico viene invece eseguito localmente sulla stessa macchina in cui hai avviato il programma Dataflow. Per ulteriori informazioni, consulta Configurare PipelineOptions per l'esecuzione locale.
Gestione di errori ed eccezioni
La pipeline potrebbe generare eccezioni durante l'elaborazione dei dati. Alcuni di questi errori sono temporanei, ad esempio una difficoltà temporanea di accesso a un servizio esterno. Altri errori sono permanenti, ad esempio errori causati da dati di input danneggiati o non analizzabili o puntatori nulli durante il calcolo.
Dataflow elabora gli elementi in bundle arbitrari e ritenta l'intero bundle quando viene generato un errore per qualsiasi elemento del bundle. Quando viene eseguito in modalità batch, i bundle che includono un elemento non riuscito vengono ritentati quattro volte. La pipeline non riesce completamente quando un singolo bundle non è riuscito quattro volte. Quando viene eseguito in modalità flusso, un bundle che include un elemento non riuscito viene ritentato indefinitamente, il che potrebbe causare il blocco permanente della pipeline.
Quando esegui l'elaborazione in modalità batch, potresti visualizzare un numero elevato di singoli errori prima che un job della pipeline non riesca completamente, il che si verifica quando un determinato bundle non riesce dopo quattro tentativi di ripetizione. Ad esempio, se la pipeline tenta di elaborare 100 bundle, Dataflow potrebbe generare diverse centinaia di singoli errori finché un singolo bundle non raggiunge la condizione di quattro errori per l'uscita.
Gli errori dei worker di avvio, come l'impossibilità di installare i pacchetti sui worker, sono temporanei. Questo scenario comporta tentativi di ripetizione indefiniti e potrebbe causare il blocco permanente della pipeline.
Parallelizzazione e distribuzione
Il servizio Dataflow parallelizza e distribuisce automaticamente la logica di elaborazione nella pipeline ai worker che assegni per eseguire il job. Dataflow utilizza le astrazioni nel
modello di programmazione per rappresentare
le funzioni di elaborazione parallela. Ad esempio, le trasformazioni ParDo in una pipeline fanno sì che Dataflow distribuisca automaticamente il codice di elaborazione, rappresentato dagli oggetti DoFn, a più worker da eseguire in parallelo.
Esistono due tipi di parallelismo dei job:
Il parallelismo orizzontale si verifica quando i dati della pipeline vengono suddivisi ed elaborati su più worker contemporaneamente. L'ambiente di runtime di Dataflow è basato su un pool di worker distribuiti. Una pipeline ha un potenziale di parallelismo più elevato quando il pool contiene più worker, ma questa configurazione ha anche un costo più elevato. In teoria, il parallelismo orizzontale non ha un limite superiore. Tuttavia, Dataflow limita il pool di worker a 4000 worker per ottimizzare l'utilizzo delle risorse a livello di flotta.
Il parallelismo verticale si verifica quando i dati della pipeline vengono suddivisi ed elaborati da più core della CPU sullo stesso worker. Ogni worker è alimentato da una VM Compute Engine. Una VM può eseguire più processi per saturare tutti i core della CPU. Una VM con più core ha un potenziale di parallelismo verticale più elevato, ma questa configurazione comporta costi più elevati. Un numero maggiore di core spesso comporta un aumento della memoria utilizzata, quindi il numero di core viene in genere scalato insieme alle dimensioni della memoria. Dato il limite fisico delle architetture dei computer, il limite superiore del parallelismo verticale è molto inferiore al limite superiore del parallelismo orizzontale.
Parallelismo gestito
Per impostazione predefinita, Dataflow gestisce automaticamente il parallelismo dei job. Dataflow monitora le statistiche di runtime del job, come l'utilizzo di CPU e memoria utilizzata, per determinare come scalare il job. A seconda delle impostazioni del job, Dataflow può scalare i job orizzontalmente, operazione denominata scalabilità automatica orizzontale, o verticalmente, operazione denominata scalabilità verticale. La scalabilità automatica per il parallelismo ottimizza il costo e il rendimento del job.
Per migliorare il rendimento dei job, Dataflow ottimizza anche le pipeline internamente. Le ottimizzazioni tipiche sono l'ottimizzazione della fusione e l'ottimizzazione della combinazione. Unendo i passaggi della pipeline, Dataflow elimina i costi non necessari associati al coordinamento dei passaggi in un sistema distribuito e all'esecuzione separata di ogni singolo passaggio.
Fattori che influiscono sul parallelismo
I seguenti fattori influiscono sul funzionamento del parallelismo nei job Dataflow.
Sorgente di ingresso
Quando una sorgente di ingresso non consente il parallelismo, il passaggio di importazione della sorgente di ingresso può diventare un collo di bottiglia in un job Dataflow. Ad esempio, quando importi dati da un singolo file di testo compresso, Dataflow non può parallelizzare i dati di input. Poiché la maggior parte dei formati di compressione non può essere suddivisa arbitrariamente in shard durante l'importazione, Dataflow deve leggere i dati in sequenza dall'inizio del file. La velocità effettiva complessiva della pipeline viene rallentata dalla parte non parallela della pipeline. La soluzione a questo problema è utilizzare una sorgente di ingresso più scalabile.
In alcuni casi, anche la fusione dei passaggi riduce il parallelismo. Se la sorgente di ingresso non consente il parallelismo, se Dataflow unisce il passaggio di importazione dati con i passaggi successivi e assegna questo passaggio unito a un singolo thread, l'intera pipeline potrebbe essere eseguita più lentamente.
Per evitare questo scenario, inserisci un passaggio Redistribute dopo il passaggio di importazione della sorgente di ingresso. Per ulteriori informazioni, consulta la sezione
Impedire la fusione di questo documento.
Fanout predefinito e forma dei dati
Il fanout predefinito di un singolo passaggio di trasformazione può diventare un collo di bottiglia e limitare il parallelismo. Ad esempio, una trasformazione ParDo "a elevato fan-out" può far sì che la fusione limiti la capacità di Dataflow di ottimizzare l'utilizzo dei worker. In un'operazione di questo tipo, potresti avere una raccolta di input con un numero relativamente ridotto di elementi, ma ParDo produce un output con centinaia o migliaia di volte più elementi, seguito da un altro ParDo. Se il servizio Dataflow unisce queste operazioni ParDo, il parallelismo in questo passaggio è limitato al massimo al numero di elementi nella raccolta di input, anche se la PCollection intermedia contiene molti più elementi.
Per le possibili soluzioni, consulta la sezione Impedire la fusione di questo documento.
Forma dei dati
La forma dei dati, che si tratti di dati di input o intermedi, può limitare il parallelismo.
Ad esempio, quando un passaggio GroupByKey su una chiave naturale, come una città, è seguito da un passaggio map o Combine, Dataflow unisce i due passaggi. Quando lo spazio delle chiavi è piccolo, ad esempio cinque città, e una chiave è molto attiva, ad esempio una grande città, la maggior parte degli elementi nell'output del passaggio GroupByKey viene distribuita a un processo. Questo processo diventa un collo di bottiglia e rallenta il job.
In questo esempio, puoi ridistribuire i risultati del passaggio GroupByKey in uno spazio delle chiavi artificiale più grande anziché utilizzare le chiavi naturali. Inserisci un passaggio Redistribute tra il passaggio GroupByKey e il passaggio map o Combine. Nel passaggio Redistribute, crea lo spazio delle chiavi artificiale, ad esempio utilizzando una funzione hash, per superare il parallelismo limitato causato dalla forma dei dati.
Per ulteriori informazioni, consulta la sezione Impedire la fusione di questo documento.
Sink di output
Un sink è una trasformazione che scrive in un sistema di archiviazione dati esterno, ad esempio un file o un database. In pratica, i sink vengono modellati e implementati come oggetti DoFn standard e vengono utilizzati per materializzare una PCollection in sistemi esterni.
In questo caso, la PCollection contiene i risultati finali della pipeline. I thread che chiamano le API sink possono essere eseguiti in parallelo per scrivere i dati nei sistemi esterni. Per impostazione predefinita, non si verifica alcun coordinamento tra i thread. Senza un livello intermedio per memorizzare nel buffer le richieste di scrittura e il flusso di controllo, il sistema esterno può essere sovraccarico e ridurre la velocità effettiva di scrittura. L'aumento delle risorse aggiungendo più parallelismo potrebbe rallentare ulteriormente la pipeline.
La soluzione a questo problema è ridurre il parallelismo nel passaggio di scrittura.
Puoi aggiungere un passaggio GroupByKey subito prima del passaggio di scrittura. Il passaggio GroupByKey raggruppa i dati di output in un insieme più piccolo di batch per ridurre il numero totale di chiamate RPC e connessioni ai sistemi esterni. Ad esempio, utilizza GroupByKey per creare uno spazio hash di 50 punti dati su 1 milione.
Lo svantaggio di questo approccio è che introduce un limite hardcoded al parallelismo. Un'altra opzione è implementare il backoff esponenziale nel sink durante la scrittura dei dati. Questa opzione può fornire una limitazione del client minima.
Monitorare il parallelismo
Per monitorare il parallelismo, puoi utilizzare la Google Cloud console per visualizzare eventuali ritardatari rilevati. Per ulteriori informazioni, consulta Risolvere i problemi relativi ai ritardatari nei job batch e Risolvere i problemi relativi ai ritardatari nei job di streaming.
Ottimizzazione della fusione
Dopo che il formato JSON del grafico di esecuzione della pipeline è stato convalidato, il servizio Dataflow potrebbe modificare il grafico per eseguire ottimizzazioni.
Le ottimizzazioni possono includere la fusione di più passaggi o trasformazioni nel grafico di esecuzione della pipeline in singoli passaggi. La fusione dei passaggi impedisce al servizio Dataflow di dover materializzare ogni PCollection intermedia nella pipeline, il che può essere costoso in termini di memoria e overhead di elaborazione.
Sebbene tutte le trasformazioni specificate nella costruzione della pipeline vengano eseguite sul servizio, per garantire l'esecuzione più efficiente della pipeline, le trasformazioni potrebbero essere eseguite in un ordine diverso o come parte di una trasformazione unita più grande. Il servizio Dataflow rispetta le dipendenze dei dati tra i passaggi nel grafico di esecuzione, ma in caso contrario i passaggi potrebbero essere eseguiti in qualsiasi ordine.
Esempio di fusione
Il seguente diagramma mostra come il grafico di esecuzione dell'esempio WordCount incluso nell' SDK Apache Beam per Java potrebbe essere ottimizzato e unito dal servizio Dataflow per un'esecuzione efficiente:
Figura 2: grafico di esecuzione ottimizzato dell'esempio WordCount
Impedire la fusione
In alcuni casi, Dataflow potrebbe indovinare in modo errato il modo ottimale per unire le operazioni nella pipeline, il che può limitare la capacità di Dataflow di utilizzare tutti i worker disponibili. In questi casi, puoi dare un suggerimento a Dataflow per ridistribuire i dati utilizzando una trasformazione Redistribute.
Per aggiungere una trasformazione Redistribute, chiama uno dei seguenti metodi:
Redistribute.arbitrarily: indica che è probabile che i dati siano sbilanciati. Dataflow sceglie l'algoritmo migliore per ridistribuire i dati.Redistribute.byKey: indica che è probabile che unaPCollectiondi coppie chiave-valore sia sbilanciata e debba essere ridistribuita in base alle chiavi. In genere, Dataflow colloca tutti gli elementi di una singola chiave nello stesso thread di lavoro. Tuttavia, la collocazione delle chiavi non è garantita e gli elementi vengono elaborati in modo indipendente.
Se la pipeline contiene una trasformazione Redistribute, Dataflow in genere impedisce la fusione dei passaggi prima e dopo la trasformazione Redistribute e esegue lo shuffling dei dati in modo che i passaggi a valle della trasformazione Redistribute abbiano un parallelismo più ottimale.
Monitorare la fusione
Puoi accedere al grafico ottimizzato e alle fasi unite laconsole, Google Cloud utilizzando gcloud CLI o l'API.
Console
Per visualizzare le fasi e i passaggi uniti del grafico nella console, nella scheda Dettagli esecuzione del job Dataflow, apri la Flusso di lavoro della fase visualizzazione del grafico.
Per visualizzare i passaggi dei componenti uniti per una fase, fai clic sulla fase unita nel grafico. Nel riquadro Informazioni sulla fase, la riga Passaggi dei componenti mostra le fasi unite. A volte, le parti di una singola trasformazione composita vengono unite in più fasi.
gcloud
Per accedere al grafico ottimizzato e alle fasi unite utilizzando gcloud CLI, esegui il seguente comando gcloud:
gcloud dataflow jobs describe --full JOB_ID --format json
Sostituisci JOB_ID con l'ID del job Dataflow.
Per estrarre i bit pertinenti, invia l'output del comando gcloud a jq:
gcloud dataflow jobs describe --full JOB_ID --format json | jq '.pipelineDescription.executionPipelineStage\[\] | {"stage_id": .id, "stage_name": .name, "fused_steps": .componentTransform }'
Per visualizzare la descrizione delle fasi unite nel file di risposta di output, all'interno del
ComponentTransform
array, consulta l'
ExecutionStageSummary
oggetto.
API
Per accedere al grafico ottimizzato e alle fasi unite utilizzando l'API, chiama
project.locations.jobs.get.
Per visualizzare la descrizione delle fasi unite nel file di risposta di output, all'interno del
ComponentTransform
array, consulta l'
ExecutionStageSummary
oggetto.
Ottimizzazione della combinazione
Le operazioni di aggregazione sono un concetto importante nell'elaborazione di dati su larga scala.
L'aggregazione riunisce dati concettualmente distanti, rendendoli estremamente utili per la correlazione. Il modello di programmazione Dataflow rappresenta le operazioni di aggregazione come le trasformazioni GroupByKey, CoGroupByKey, e
Combine.
Le operazioni di aggregazione di Dataflow combinano i dati nell'intero set di dati, inclusi i dati che potrebbero essere distribuiti su più worker. Durante queste operazioni di aggregazione, spesso è più efficiente combinare il maggior numero possibile di dati localmente prima di combinare i dati tra le istanze. Quando applichi una trasformazione GroupByKey o un'altra trasformazione di aggregazione, il servizio Dataflow esegue automaticamente una combinazione parziale localmente prima dell'operazione di raggruppamento principale.
Quando esegue la combinazione parziale o multilivello, il servizio Dataflow prende decisioni diverse a seconda che la pipeline funzioni con dati batch o di streaming. Per i dati delimitati, il servizio privilegia l'efficienza ed esegue la maggior parte della combinazione locale possibile. Per i dati non delimitati, il servizio privilegia una latenza inferiore e potrebbe non eseguire la combinazione parziale, perché potrebbe aumentare la latenza.