Questa pagina fornisce una panoramica del ciclo di vita della pipeline, dal codice alla del job Dataflow.
In questa pagina vengono spiegati i seguenti concetti:
- Che cos'è un grafico di esecuzione e come una pipeline Apache Beam diventa Job Dataflow
- In che modo Dataflow gestisce gli errori
- In che modo Dataflow carica e distribuisce automaticamente logica di elaborazione nella pipeline per i worker che eseguono il job
- Ottimizzazioni dei job che Dataflow potrebbe rendere
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 associate,
come oggetti DoFn
. Questo è il grafico di esecuzione della pipeline, dove la fase viene chiamata
tempo di creazione del grafico.
Durante la creazione del grafico, Apache Beam esegue localmente il codice
il principale punto di ingresso del codice della pipeline, fermandosi alle chiamate a origini,
o di trasformazione, trasformando queste chiamate in nodi del grafico.
Di conseguenza, una porzione di codice nel punto di ingresso di una pipeline (Java e Go main
o il livello più alto di uno script Python) viene eseguito localmente sulla macchina
esegue la pipeline. Lo stesso codice dichiarato in un metodo di un oggetto DoFn
viene eseguita nei worker Dataflow.
Ad esempio, il valore WordCount incluso negli SDK Apache Beam, contiene una serie di trasformazioni leggere, estrarre, contare, formattare e scrivere le singole parole di una raccolta di testo, insieme a un conteggio di occorrenze per ogni parola. Il seguente diagramma mostra come le trasformazioni nella pipeline di conteggio parole vengono espanse grafico di esecuzione:
Figura 1: grafico di esecuzione di esempio di conteggio parole
Il grafico di esecuzione è spesso diverso dall'ordine in cui hai specificato i quando crei la pipeline. Questa differenza esiste perché Il servizio Dataflow esegue varie ottimizzazioni e fusioni di esecuzione prima che venga eseguito su risorse cloud gestite. La Il servizio Dataflow rispetta le dipendenze dei dati durante l'esecuzione una pipeline o un blocco note personalizzato. Tuttavia, i passaggi senza dipendenze tra i dati possono essere eseguiti in qualsiasi ordine.
Per vedere il grafico di esecuzione non ottimizzato di Dataflow generati per la pipeline, seleziona il job Interfaccia di monitoraggio di Dataflow. Per ulteriori informazioni sulla visualizzazione offerte di lavoro, consulta Utilizzare l'interfaccia di monitoraggio di Dataflow.
Durante la creazione dei grafici, Apache Beam verifica che tutte le risorse a cui fa riferimento la pipeline, come i bucket Cloud Storage, tabelle BigQuery e argomenti Pub/Sub o esistono e sono accessibili. La convalida viene eseguita tramite di chiamate API standard ai rispettivi servizi, quindi è fondamentale che l'utente L'account utilizzato per eseguire una pipeline disponga di una connettività adeguata ai servizi necessari ed è autorizzato a chiamare le API dei servizi. Prima di inviare la pipeline nel servizio Dataflow, Apache Beam controlla anche altri errori, e garantisce che il grafico della pipeline non contenga operazioni illegali.
Il grafico di esecuzione viene quindi tradotto in formato JSON e l'esecuzione in JSON viene trasmesso all'endpoint del servizio Dataflow.
Il servizio Dataflow convalida quindi il grafico di esecuzione JSON. Quando convalidato, diventa un job in Dataflow completamente gestito di Google Cloud. Puoi visualizzare il job, il relativo grafico di esecuzione, lo stato e il log utilizzando l'interfaccia di monitoraggio di Dataflow.
Java
Il servizio Dataflow invia una risposta alla macchina su cui esegui
il tuo programma Dataflow. Questa risposta è incapsulata nell'oggetto
DataflowPipelineJob
, che contiene il valore jobId
del tuo job Dataflow.
Utilizza jobId
per monitorare, tracciare e risolvere i problemi del job tramite l'
Interfaccia di monitoraggio di Dataflow
e l'interfaccia a riga di comando di Dataflow.
Per ulteriori informazioni, consulta
Riferimento API per DataflowPipelineJob.
Python
Il servizio Dataflow invia una risposta alla macchina su cui esegui
il tuo programma Dataflow. Questa risposta è incapsulata nell'oggetto
DataflowPipelineResult
, che contiene il valore job_id
del tuo job Dataflow.
Utilizza l'job_id
per monitorare e tracciare il job, nonché per risolvere eventuali problemi
utilizzando
Interfaccia di monitoraggio di Dataflow
e ai
Interfaccia a riga di comando di Dataflow.
Vai
Il servizio Dataflow invia una risposta alla macchina su cui esegui
il tuo programma Dataflow. Questa risposta è incapsulata nell'oggetto
dataflowPipelineResult
, che contiene il valore jobID
del tuo job Dataflow.
Utilizza l'jobID
per monitorare e tracciare il job, nonché per risolvere eventuali problemi
utilizzando
Interfaccia di monitoraggio di Dataflow
e ai
Interfaccia a riga di comando di Dataflow.
La creazione del grafico avviene anche quando esegui la pipeline localmente, ma il grafico non viene tradotto in JSON né trasmesso al servizio. Il grafico viene invece eseguito localmente sulla stessa macchina in cui hai avviato programma Dataflow. Per ulteriori informazioni, vedi Configurazione di 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 difficoltà temporanee di accesso a un servizio esterno. Altro gli errori sono permanenti, ad esempio errori causati da dati di input corrotti o non analizzabili, o puntatori nulli durante il calcolo.
Dataflow elabora gli elementi in bundle arbitrari e riprova il bundle completo quando viene generato un errore per un qualsiasi elemento del bundle. Quando in modalità batch, i bundle che includono un elemento con errori vengono ritentati quattro volte. Si verifica un errore completo della pipeline quando un singolo bundle presenta errori per quattro volte. Quando in esecuzione in modalità flusso di dati, viene ritentato un bundle che include un elemento con errori a tempo indeterminato, il che potrebbe causare il blocco permanente della pipeline.
Durante l'elaborazione in modalità batch, potresti notare singoli errori prima di un errore completo di un job della pipeline, che si verifica quando un determinato bundle fallisce dopo quattro tentativi. Ad esempio, se la tua pipeline tenta di elaborare 100 pacchetti, Dataflow generare diverse centinaia di singoli errori finché un singolo bundle non raggiunge condizione "4-failure" per l'uscita.
Gli errori del worker di avvio, come la mancata installazione di pacchetti sui worker, sono: transitorio. Questo scenario comporta nuovi tentativi indefiniti e potrebbe causare la pipeline si blocchi definitivamente.
Parallelizzazione e distribuzione
Il servizio Dataflow lo parallelizza e distribuisce automaticamente
la logica di elaborazione della pipeline ai worker assegnati per l'esecuzione
il tuo lavoro. Dataflow utilizza le astrazioni
modello di programmazione per rappresentare
funzioni di elaborazione parallela. Ad esempio, ParDo
trasforma in una pipeline
fanno sì che Dataflow distribuisca automaticamente il codice di elaborazione,
rappresentati da DoFn
oggetti, a più worker da eseguire in parallelo.
Esistono due tipi di parallelismo dei lavori:
Il parallelismo orizzontale si verifica quando i dati della pipeline vengono suddivisi ed elaborati più worker contemporaneamente. Il runtime di Dataflow è alimentato da un pool di worker distribuiti. Una pipeline ha un parallelismo potenziale più alto quando il pool contiene più worker, ma anche questa configurazione ha un costo maggiore. In teoria, orizzontale il parallelismo non ha un limite massimo. Tuttavia, Dataflow limita il pool di worker a 4000 worker per ottimizzare l'utilizzo delle risorse a livello di parco risorse.
Il parallelismo verticale si verifica quando i dati della pipeline vengono suddivisi ed elaborati più core CPU sullo stesso worker. Ogni worker è alimentato da un alla VM di Compute Engine. Una VM può eseguire più processi per saturare tutti i suoi Core della CPU. Una VM con più core ha un potenziale più elevato parallelismo verticale, ma questa configurazione comporta costi più alti. Spesso è necessario un numero di core più alto comporta un aumento dell'utilizzo della memoria, quindi il numero di core solitamente è scalate insieme alle dimensioni della memoria. Dato il limite fisico del computer di architetture, il limite superiore di parallelismo verticale è molto più basso limite superiore del parallelismo orizzontale.
Parallelismo gestito
Per impostazione predefinita, Dataflow gestisce automaticamente il parallelismo dei job. Dataflow monitora le statistiche di runtime per il job, ad esempio la CPU e memoria utilizzata per determinare come scalare il job. A seconda del lavoro le impostazioni, Dataflow può scalare i job orizzontalmente, Scalabilità automatica orizzontale oppure in verticale, noto come scalabilità verticale. La scalabilità automatica per il parallelismo ottimizza i costi del job le prestazioni lavorative.
Per migliorare le prestazioni del job, Dataflow ottimizza anche le pipeline internamente. Le ottimizzazioni tipiche sono l'ottimizzazione della fusione e combinare l'ottimizzazione. Unendo i passaggi della pipeline, Dataflow elimina le i costi associati al coordinamento delle varie fasi in un sistema distribuito ed eseguire ogni singolo passaggio separatamente.
Fattori che influenzano il parallelismo
I seguenti fattori influiscono sul funzionamento del parallelismo di job Dataflow.
Origine di input
Quando un'origine di input non consente il parallelismo, il passaggio di importazione dell'origine di input può un collo di bottiglia in un job Dataflow. Ad esempio, quando importi dati da un singolo file di testo compresso, Dataflow non può caricare i dati di input. Poiché la maggior parte dei formati di compressione non può essere suddivisa arbitrariamente più piccoli durante l'importazione, Dataflow deve leggere i dati in sequenza all'inizio del file. La velocità effettiva complessiva della pipeline viene rallentata parte non parallela della pipeline. La soluzione a questo problema consiste nel utilizza un'origine di input più scalabile.
In alcuni casi, la step fusion riduce anche il parallelismo. Quando l'origine di input non consente il parallelismo, se Dataflow unisce il passaggio di importazione dati con i passaggi successivi e assegna questo passaggio integrato in un singolo thread, l'intera pipeline potrebbe essere eseguita più lentamente.
Per evitare questo scenario, inserisci un passaggio Reshuffle
dopo l'origine di input
fase di importazione. Per ulteriori informazioni, consulta
Sezione Impedisci fusione di questo documento.
Fanout e forma dei dati predefiniti
Il fan-out predefinito di un singolo passaggio di trasformazione può diventare un
collo di bottiglia e limitare il parallelismo. Ad esempio, "fan-out elevato" La trasformazione ParDo
può causare
fusione per limitare la capacità di Dataflow di ottimizzare l'utilizzo dei worker. Nella
un'operazione di questo tipo, potresti avere una raccolta di input con
ma ParDo
produce un output con centinaia o migliaia di volte
tanti elementi, seguiti da altri ParDo
. Se la Dataflow
il servizio fonde insieme queste operazioni ParDo
, il parallelismo in questo passaggio è
limitato al massimo al numero di elementi nella raccolta di input, anche se
il campo PCollection
intermedio contiene molti più elementi.
Per le potenziali soluzioni, vedi Sezione Impedisci 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 GroupByKey
sale su una chiave naturale, come una città,
seguito da un passaggio map
o Combine
, Dataflow fonde i due
passaggi. Quando lo spazio delle chiavi è piccolo, ad esempio cinque città, e una chiave è molto
caldo, ad esempio una grande città, la maggior parte degli elementi nell'output di GroupByKey
vengono distribuiti in un unico processo. Questo processo diventa un collo di bottiglia e rallenta
fino al completamento del lavoro.
In questo esempio, puoi ridistribuire i risultati del passaggio GroupByKey
in un
uno spazio artificiale più ampio rispetto ai tasti naturali. Inserisci un
Reshuffle
passaggio tra il passaggio GroupByKey
e il passaggio map
o Combine
.
Nel passaggio Reshuffle
, crea lo spazio della chiave artificiale, ad esempio utilizzando un
hash
, per superare il limitato parallelismo causato dalla forma dei dati.
Per ulteriori informazioni, consulta Sezione Impedisci fusione di questo documento.
Sink di output
Un sink è una trasformazione che scrive in un sistema di archiviazione dati esterno, come
o un database. In pratica, i sink vengono modellati e implementati come standard
DoFn
e vengono utilizzati per materializzare un PCollection
ai sistemi esterni.
In questo caso, PCollection
contiene i risultati finali della pipeline. Thread che
le API di sink delle chiamate possono essere eseguite in parallelo per scrivere dati nei sistemi esterni. Di
per impostazione predefinita, non si verifica alcun coordinamento tra i thread. Senza una intermedia
per il buffering delle richieste di scrittura e del flusso di controllo, il sistema esterno
si sovraccarichi e si riducono
la velocità effettiva di scrittura. Fai lo scale up delle risorse aggiungendo altre
il parallelismo potrebbe rallentare ulteriormente la pipeline.
La soluzione a questo problema consiste nel ridurre il parallelismo nella fase di scrittura.
Puoi aggiungere un passaggio GroupByKey
subito prima di quello di scrittura. GroupByKey
i gruppi di passaggi inviano i dati in un insieme più piccolo di batch per ridurre il numero totale di chiamate RPC
e connessioni a sistemi esterni. Ad esempio, utilizza GroupByKey
per creare un
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 e i dati di Google Cloud. Questa opzione può offrire una limitazione minima del client.
Monitora il parallelismo
Per monitorare il parallelismo, puoi utilizzare la console Google Cloud per visualizzare in ritardo. Per ulteriori informazioni, vedi Risolvi i problemi relativi agli elementi in ritardo nei job batch e Risolvi i problemi degli elementi in ritardo nei job di flussi di dati.
Ottimizzazione Fusion
Dopo aver convalidato il formato JSON del grafico di esecuzione della pipeline,
Il servizio Dataflow potrebbe modificare il grafico per eseguire le ottimizzazioni.
Le ottimizzazioni possono includere la fusione di più passaggi o trasformazioni
il grafico di esecuzione della pipeline in singoli passaggi. L'unione dei passaggi impedisce
dal servizio Dataflow alla necessità di materializzare ogni intermedio
PCollection
nella pipeline, il che può essere costoso in termini di memoria
sull'overhead di elaborazione.
Sebbene tutte le trasformazioni specificate nella creazione della pipeline siano eseguiti sul servizio, al fine di garantire l'esecuzione più efficiente del pipeline, le trasformazioni possono essere eseguite in un ordine diverso o come parte di una trasformazione fusa più grande. Il servizio Dataflow rispetta i dati dipendenze tra i passaggi nel grafico di esecuzione, ma altrimenti i passaggi potrebbero essere eseguiti in qualsiasi ordine.
Esempio di Fusion
Il seguente diagramma mostra come il grafico di esecuzione Esempio di WordCount incluso con L'SDK Apache Beam per Java può essere ottimizzato e integrato Servizio Dataflow per un'esecuzione efficiente:
Figura 2: esempio di grafico di esecuzione ottimizzato per il conteggio parole
Prevenzione della fusione
In alcuni casi, Dataflow potrebbe indovinare erroneamente il modo ottimale per fondere le operazioni nella pipeline, Abilità di Dataflow di utilizzare tutti i worker disponibili. In questi casi, puoi impedire a Dataflow di eseguire ottimizzazioni della fusione.
Puoi impedire la fusione dei passaggi aggiungendo alla pipeline un'operazione che
obbliga il servizio Dataflow a materializzare l'intermedio
PCollection
. Prendi in considerazione l'utilizzo di una delle seguenti operazioni:
- Inserisci un
GroupByKey
e separali dopo il primoParDo
. La Il servizio Dataflow non fonde mai le operazioniParDo
in un e aggregazione. - Supera il tuo livello
PCollection
intermedio come input aggiuntivo a un altroParDo
. Il servizio Dataflow si materializza sempre input aggiuntivi. - Inserisci un passaggio
Reshuffle
.Reshuffle
impedisce la fusione, i checkpoint dei dati ed esegue la deduplicazione dei record. Il rimpasto è supportato da Dataflow anche se è contrassegnato come deprecato nel documentazione di Apache Beam.
Monitora Fusion
Puoi accedere al grafico ottimizzato e alle fasi raggruppate nella console Google Cloud. con gcloud CLI o l'API.
Console
Per visualizzare le fasi e i passaggi del grafico Nella console, nella scheda Dettagli esecuzione relativa al tuo Job Dataflow, apri Flusso di lavoro della fase visualizzazione grafico.
Per visualizzare i passaggi del componente integrati per una fase, nel grafico fai clic su nella fase di fusione. Nel riquadro Informazioni sulla fase, la riga Passaggi dei componenti mostra tutte le fasi. A volte le parti di una singola trasformazione composita sono fusi in più fasi.
gcloud
Per accedere al grafico ottimizzato e alle fasi unite utilizzando il
gcloud CLI, esegui questo comando gcloud
:
gcloud dataflow jobs describe --full JOB_ID --format json
Sostituisci JOB_ID
con l'ID del tuo 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 vedere la descrizione delle fasi fusi nel file di risposta di output, all'interno della
ComponentTransform
osserva
ExecutionStageSummary
.
API
Per accedere al grafico ottimizzato e alle fasi unite utilizzando l'API, richiama
project.locations.jobs.get
Per vedere la descrizione delle fasi fusi nel file di risposta di output, all'interno della
ComponentTransform
osserva
ExecutionStageSummary
.
Combina ottimizzazione
Le operazioni di aggregazione sono un concetto importante nell'elaborazione dati su larga scala.
L'aggregazione riunisce dati concettualmente lontani, in modo che
estremamente utile per la correlazione. Il modello di programmazione Dataflow rappresenta le operazioni di aggregazione come GroupByKey
, CoGroupByKey
e
Combine
trasforma.
Le operazioni di aggregazione di Dataflow combinano dati nell'intero
tra cui quelli che potrebbero essere distribuiti su più worker. Durante l'esecuzione
di aggregazione, spesso è più efficiente combinare quanti più dati
localmente prima di combinare i dati tra le istanze. Quando applichi un
GroupByKey
o altra trasformazione di aggregazione, il servizio Dataflow
esegue automaticamente la combinazione parziale in locale prima del raggruppamento principale
operativa.
Quando si esegue una combinazione parziale o multilivello, il Dataflow servizio prende decisioni diverse a seconda che la pipeline stia utilizzando o meno in modalità flusso o batch. Per i dati limitati, il servizio favorisce l'efficienza eseguirà il maggior numero possibile di combinazioni locali. Per i dati illimitati, favorisce una latenza più bassa e potrebbe non eseguire una combinazione parziale, potrebbe aumentare la latenza.