Questa pagina fornisce suggerimenti per la risoluzione dei problemi e strategie di debug utili in caso di problemi durante la creazione o l'esecuzione della pipeline Dataflow. Queste informazioni possono aiutarti a rilevare un errore della pipeline, a determinare il motivo di un'esecuzione non riuscita di una pipeline e a suggerire alcuni provvedimenti per risolvere il problema.
Il seguente diagramma mostra il flusso di lavoro per la risoluzione dei problemi di Dataflow descritto in questa pagina.
Dataflow fornisce feedback in tempo reale sul job ed è disponibile un insieme di base di passaggi che puoi utilizzare per verificare i messaggi di errore, i log e la presenza di condizioni, come l'arresto del job.
Per indicazioni sugli errori comuni che potresti riscontrare durante l'esecuzione del job Dataflow, vedi Risolvere i problemi relativi agli errori di Dataflow. Per monitorare e risolvere i problemi delle prestazioni della pipeline, consulta Monitorare le prestazioni della pipeline.
Best practice per le pipeline
Di seguito sono riportate le best practice per le pipeline Java, Python e Go.
Java
Per i job batch, ti consigliamo di impostare una durata (TTL) per la località temporanea.
Prima di impostare il TTL e come best practice generale, assicurati di impostare sia la località temporanea sia quella temporanea su località diverse.
Non eliminare gli oggetti nella posizione temporanea perché vengono riutilizzati.
Se un job viene completato o arrestato e gli oggetti temporanei non vengono eliminati, rimuovi manualmente questi file dal bucket Cloud Storage utilizzato come posizione temporanea.
Python
Entrambe le posizioni temporanee e temporanee hanno un prefisso di <job_name>.<time>.
Assicurati di impostare sia la località temporanea sia quella temporanea su località diverse.
Se necessario, elimina gli oggetti nella posizione temporanea dopo il completamento o l'arresto di un job. Inoltre, gli oggetti in fasi non vengono riutilizzati nelle pipeline Python.
Se un job termina e gli oggetti temporanei non vengono ripuliti, rimuovi manualmente questi file dal bucket Cloud Storage utilizzato come posizione temporanea.
Per i job batch, ti consigliamo di impostare una durata (TTL) sia per la posizione temporanea che per quella temporanea.
Go
Entrambe le posizioni temporanee e temporanee hanno un prefisso di
<job_name>.<time>.Assicurati di impostare sia la località temporanea sia quella temporanea su località diverse.
Se necessario, elimina gli oggetti nella posizione temporanea dopo il completamento o l'arresto di un job. Inoltre, gli oggetti in fasi non vengono riutilizzati nelle pipeline Go.
Se un job termina e gli oggetti temporanei non vengono ripuliti, rimuovi manualmente questi file dal bucket Cloud Storage utilizzato come posizione temporanea.
Per i job batch, ti consigliamo di impostare una durata (TTL) sia per la posizione temporanea che per quella temporanea.
Controllare lo stato della pipeline
Puoi rilevare eventuali errori nelle esecuzioni della pipeline utilizzando l'interfaccia di monitoraggio di Dataflow.
- Vai alla console Google Cloud.
- Seleziona il tuo progetto Google Cloud dall'elenco dei progetti.
- Nel menu di navigazione, in Big Data, fai clic su Dataflow. Nel riquadro di destra viene visualizzato un elenco dei job in esecuzione.
- Seleziona il job di pipeline che vuoi visualizzare. Puoi visualizzare un riepilogo dello stato dei job nel campo Stato: "In esecuzione", "Riuscito" o "Non riuscito".
Trovare informazioni sugli errori della pipeline
Se uno dei job di pipeline ha esito negativo, puoi selezionarlo per visualizzare informazioni più dettagliate sugli errori e sull'esecuzione dei risultati. Quando selezioni un job, puoi visualizzare i grafici chiave per la pipeline, il grafico di esecuzione, il riquadro Informazioni job e il riquadro Log con le schede Log dei job, Log dei worker, Diagnostica e Consigli.
Controlla i messaggi di errore del job
Per visualizzare i log dei job generati dal codice pipeline e dal servizio Dataflow, nel riquadro Log fai clic su segmentMostra.
Puoi filtrare i messaggi che vengono visualizzati nei log dei job facendo clic su Informazioniarrow_drop_down e filter_listFiltra. Per visualizzare solo i messaggi di errore, fai clic su Informazioniarrow_drop_down e seleziona Errore.
Per espandere un messaggio di errore, fai clic sulla sezione espandibile arrow_right.
In alternativa, puoi fare clic sulla scheda Diagnostica. Questa scheda mostra dove si sono verificati gli errori lungo la sequenza temporale scelta, un conteggio di tutti gli errori registrati e i possibili suggerimenti per la pipeline.
Visualizza i log dei passaggi per il job
Quando selezioni un passaggio nel grafico della pipeline, il riquadro dei log passa dalla visualizzazione dei log del job generati dal servizio Dataflow a quella delle istanze di Compute Engine che eseguono il passaggio della pipeline.
Cloud Logging combina tutti i log raccolti dalle istanze di Compute Engine del tuo progetto in un'unica posizione. Consulta Messaggi della pipeline di Logging per ulteriori informazioni sull'utilizzo delle varie funzionalità di logging di Dataflow.
Gestisci il rifiuto delle pipeline automatizzate
In alcuni casi, il servizio Dataflow identifica che la pipeline potrebbe attivare problemi noti con gli SDK. Per impedire l'invio di pipeline che potrebbero presentare problemi, Dataflow rifiuta automaticamente la pipeline e visualizza il seguente messaggio:
The workflow was automatically rejected by the service because it might trigger an identified bug in the SDK (details below). If you think this identification is in error, and would like to override this automated rejection, please re-submit this workflow with the following override flag: [OVERRIDE FLAG]. Bug details: [BUG DETAILS]. Contact Google Cloud Support for further help. Please use this identifier in your communication: [BUG ID].
Dopo aver letto le avvertenze nei dettagli del bug collegato, se vuoi provare a eseguire comunque la pipeline, puoi ignorare il rifiuto automatico. Aggiungi il flag --experiments=<override-flag> e invia nuovamente la pipeline.
Determinare la causa di un errore della pipeline
In genere, un'esecuzione di pipeline Apache Beam non riuscita può essere attribuita a una delle seguenti cause:
- Errori di creazione di grafici o pipeline. Questi errori si verificano quando Dataflow si verifica un problema durante la creazione del grafico dei passaggi che compongono la pipeline, come descritto dalla pipeline Apache Beam.
- Errori nella convalida del job. Il servizio Dataflow convalida qualsiasi job di pipeline avviato. Gli errori nel processo di convalida possono impedire la creazione o l'esecuzione del job. Gli errori di convalida possono includere problemi con il bucket Cloud Storage del progetto Google Cloud o con le autorizzazioni del progetto.
- Eccezioni nel codice worker. Questi errori si verificano quando si verificano errori o bug nel codice fornito dall'utente che Dataflow distribuisce ai worker paralleli, ad esempio le istanze
DoFndi una trasformazioneParDo. - Errori causati da guasti temporanei in altri servizi Google Cloud. La pipeline potrebbe non riuscire a causa di un'interruzione temporanea o di altri problemi nei servizi Google Cloud da cui dipende Dataflow, ad esempio Compute Engine o Cloud Storage.
Rileva errori di costruzione di grafici o pipeline
Quando Dataflow crea il grafico di esecuzione per la pipeline a partire dal codice nel tuo programma Dataflow, può verificarsi un errore di creazione del grafico. Durante la creazione del grafico, Dataflow controlla la presenza di operazioni illegali.
Se Dataflow rileva un errore nella creazione del grafico, tieni presente che non viene creato alcun job nel servizio Dataflow. Pertanto, non vedrai alcun feedback nell'interfaccia di monitoraggio di Dataflow. Nella finestra della console o del terminale in cui hai eseguito la pipeline Apache Beam viene invece visualizzato un messaggio di errore simile al seguente:
Java
Ad esempio, se la tua pipeline tenta di eseguire un'aggregazione come
GroupByKey su un elemento PCollection con finestra globale, non attivato e
illimitato, viene visualizzato un messaggio di errore simile al seguente:
... ... Exception in thread "main" java.lang.IllegalStateException: ... GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without a trigger. ... Use a Window.into or Window.triggering transform prior to GroupByKey ...
Python
Ad esempio, se la pipeline utilizza suggerimenti del tipo e il tipo di argomento in una delle trasformazioni non è come previsto, viene visualizzato un messaggio di errore simile al seguente:
... in <module> run()
... in run | beam.Map('count', lambda (word, ones): (word, sum(ones))))
... in __or__ return self.pipeline.apply(ptransform, self)
... in apply transform.type_check_inputs(pvalueish)
... in type_check_inputs self.type_check_inputs_or_outputs(pvalueish, 'input')
... in type_check_inputs_or_outputs pvalue_.element_type))
google.cloud.dataflow.typehints.decorators.TypeCheckError: Input type hint violation at group: expected Tuple[TypeVariable[K], TypeVariable[V]], got <type 'str'>
Go
Ad esempio, se la pipeline utilizza un valore "DoFn" che non accetta alcun input, viene visualizzato un messaggio di errore simile al seguente:
... panic: Method ProcessElement in DoFn main.extractFn is missing all inputs. A main input is required. ... Full error: ... inserting ParDo in scope root/CountWords ... graph.AsDoFn: for Fn named main.extractFn ... ProcessElement method has no main inputs ... goroutine 1 [running]: ... github.com/apache/beam/sdks/v2/go/pkg/beam.MustN(...) ... (more stacktrace)
Se si verifica un errore di questo tipo, controlla il codice della pipeline per verificare che le operazioni della pipeline siano legali.
Rileva gli errori nella convalida dei job di Dataflow
Dopo che il servizio Dataflow ha ricevuto il grafico della pipeline, tenterà di convalidare il job. Questa convalida include quanto segue:
- Assicurati che il servizio possa accedere ai bucket Cloud Storage associati al job per la gestione temporanea dei file e l'output temporaneo.
- È in corso la verifica delle autorizzazioni richieste nel progetto Google Cloud.
- Assicurati che il servizio possa accedere alle origini di input e di output, ad esempio i file.
Se il job non riesce a completare la procedura di convalida, viene visualizzato un messaggio di errore nell'interfaccia di monitoraggio di Dataflow e nella finestra della console o del terminale, se utilizzi il blocco dell'esecuzione. Il messaggio di errore è simile al seguente:
Java
INFO: To access the Dataflow monitoring console, please navigate to
https://console.developers.google.com/project/google.com%3Aclouddfe/dataflow/job/2016-03-08_18_59_25-16868399470801620798
Submitted job: 2016-03-08_18_59_25-16868399470801620798
...
... Starting 3 workers...
... Executing operation BigQuery-Read+AnonymousParDo+BigQuery-Write
... Executing BigQuery import job "dataflow_job_16868399470801619475".
... Stopping worker pool...
... Workflow failed. Causes: ...BigQuery-Read+AnonymousParDo+BigQuery-Write failed.
Causes: ... BigQuery getting table "non_existent_table" from dataset "cws_demo" in project "my_project" failed.
Message: Not found: Table x:cws_demo.non_existent_table HTTP Code: 404
... Worker pool stopped.
... com.google.cloud.dataflow.sdk.runners.BlockingDataflowPipelineRunner run
INFO: Job finished with status FAILED
Exception in thread "main" com.google.cloud.dataflow.sdk.runners.DataflowJobExecutionException:
Job 2016-03-08_18_59_25-16868399470801620798 failed with status FAILED
at com.google.cloud.dataflow.sdk.runners.DataflowRunner.run(DataflowRunner.java:155)
at com.google.cloud.dataflow.sdk.runners.DataflowRunner.run(DataflowRunner.java:56)
at com.google.cloud.dataflow.sdk.Pipeline.run(Pipeline.java:180)
at com.google.cloud.dataflow.integration.BigQueryCopyTableExample.main(BigQueryCopyTableExample.java:74)
Python
INFO:root:Created job with id: [2016-03-08_14_12_01-2117248033993412477] ... Checking required Cloud APIs are enabled. ... Job 2016-03-08_14_12_01-2117248033993412477 is in state JOB_STATE_RUNNING. ... Combiner lifting skipped for step group: GroupByKey not followed by a combiner. ... Expanding GroupByKey operations into optimizable parts. ... Lifting ValueCombiningMappingFns into MergeBucketsMappingFns ... Annotating graph with Autotuner information. ... Fusing adjacent ParDo, Read, Write, and Flatten operations ... Fusing consumer split into read ... ... Starting 1 workers... ... ... Executing operation read+split+pair_with_one+group/Reify+group/Write ... Executing failure step failure14 ... Workflow failed. Causes: ... read+split+pair_with_one+group/Reify+group/Write failed. Causes: ... Unable to view metadata for files: gs://dataflow-samples/shakespeare/missing.txt. ... Cleaning up. ... Tearing down pending resources... INFO:root:Job 2016-03-08_14_12_01-2117248033993412477 is in state JOB_STATE_FAILED.
Go
La convalida del job descritta in questa sezione non è attualmente supportata per Go. Gli errori dovuti a questi problemi vengono visualizzati come eccezioni worker.
Rileva un'eccezione nel codice worker
Mentre il job è in esecuzione, potresti riscontrare errori o eccezioni nel codice worker. In genere, questi errori indicano che gli elementi DoFn nel codice della pipeline hanno generato eccezioni non gestite, che comportano attività non riuscite nel job di Dataflow.
Le eccezioni nel codice utente, ad esempio le istanze DoFn, sono riportate
nell'interfaccia di monitoraggio Dataflow.
Se esegui la pipeline con il blocco dell'esecuzione, i messaggi di errore vengono stampati nella console o nella finestra del terminale, come quelli riportati di seguito:
Java
INFO: To access the Dataflow monitoring console, please navigate to https://console.developers.google.com/project/example_project/dataflow/job/2017-05-23_14_02_46-1117850763061203461
Submitted job: 2017-05-23_14_02_46-1117850763061203461
...
... To cancel the job using the 'gcloud' tool, run: gcloud beta dataflow jobs --project=example_project cancel 2017-05-23_14_02_46-1117850763061203461
... Autoscaling is enabled for job 2017-05-23_14_02_46-1117850763061203461.
... The number of workers will be between 1 and 15.
... Autoscaling was automatically enabled for job 2017-05-23_14_02_46-1117850763061203461.
...
... Executing operation BigQueryIO.Write/BatchLoads/Create/Read(CreateSource)+BigQueryIO.Write/BatchLoads/GetTempFilePrefix+BigQueryIO.Write/BatchLoads/TempFilePrefixView/BatchViewOverrides.GroupByWindowHashAsKeyAndWindowAsSortKey/ParDo(UseWindowHashAsKeyAndWindowAsSortKey)+BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/WithKeys/AddKeys/Map/ParMultiDo(Anonymous)+BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/Combine.perKey(Singleton)/GroupByKey/Reify+BigQueryIO.Write/BatchLoads/TempFilePrefixView/Combine.GloballyAsSingletonView/Combine.globally(Singleton)/Combine.perKey(Singleton)/GroupByKey/Write+BigQueryIO.Write/BatchLoads/TempFilePrefixView/BatchViewOverrides.GroupByWindowHashAsKeyAndWindowAsSortKey/BatchViewOverrides.GroupByKeyAndSortValuesOnly/Write
... Workers have started successfully.
...
... org.apache.beam.runners.dataflow.util.MonitoringUtil$LoggingHandler process SEVERE: 2017-05-23T21:06:33.711Z: (c14bab21d699a182): java.lang.RuntimeException: org.apache.beam.sdk.util.UserCodeException: java.lang.ArithmeticException: / by zero
at com.google.cloud.dataflow.worker.runners.worker.GroupAlsoByWindowsParDoFn$1.output(GroupAlsoByWindowsParDoFn.java:146)
at com.google.cloud.dataflow.worker.runners.worker.GroupAlsoByWindowFnRunner$1.outputWindowedValue(GroupAlsoByWindowFnRunner.java:104)
at com.google.cloud.dataflow.worker.util.BatchGroupAlsoByWindowAndCombineFn.closeWindow(BatchGroupAlsoByWindowAndCombineFn.java:191)
...
... Cleaning up.
... Stopping worker pool...
... Worker pool stopped.
Python
INFO:root:Job 2016-03-08_14_21_32-8974754969325215880 is in state JOB_STATE_RUNNING. ... INFO:root:... Expanding GroupByKey operations into optimizable parts. INFO:root:... Lifting ValueCombiningMappingFns into MergeBucketsMappingFns INFO:root:... Annotating graph with Autotuner information. INFO:root:... Fusing adjacent ParDo, Read, Write, and Flatten operations ... INFO:root:...: Starting 1 workers... INFO:root:...: Executing operation group/Create INFO:root:...: Value "group/Session" materialized. INFO:root:...: Executing operation read+split+pair_with_one+group/Reify+group/Write INFO:root:Job 2016-03-08_14_21_32-8974754969325215880 is in state JOB_STATE_RUNNING. INFO:root:...: ...: Workers have started successfully. INFO:root:Job 2016-03-08_14_21_32-8974754969325215880 is in state JOB_STATE_RUNNING. INFO:root:...: Traceback (most recent call last): File ".../dataflow_worker/batchworker.py", line 384, in do_work self.current_executor.execute(work_item.map_task) ... File ".../apache_beam/examples/wordcount.runfiles/py/apache_beam/examples/wordcount.py", line 73, in <lambda> ValueError: invalid literal for int() with base 10: 'www'
Go
... 2022-05-26T18:32:52.752315397Zprocess bundle failed for instruction ... process_bundle-4031463614776698457-2 using plan s02-6 : while executing ... Process for Plan[s02-6] failed: Oh no! This is an error message!
Valuta la possibilità di evitare errori nel codice aggiungendo gestori delle eccezioni. Ad esempio, se vuoi eliminare gli elementi che non superano la convalida di un input personalizzato eseguita in un elemento ParDo, gestisci l'eccezione all'interno di DoFn e rilascia l'elemento.
Puoi anche monitorare gli elementi non funzionanti in diversi modi:
- Puoi registrare gli elementi con errori e verificare l'output utilizzando Cloud Logging.
- Puoi verificare la presenza di avvisi o errori nei log di avvio dei worker e dei worker Dataflow seguendo le istruzioni riportate in Visualizzazione dei log.
- Puoi fare in modo che
ParDoscriva gli elementi con errori in un output aggiuntivo per un'ispezione successiva.
Per monitorare le proprietà di una pipeline in esecuzione, puoi utilizzare la classe Metrics, come mostrato nell'esempio seguente:
Java
final Counter counter = Metrics.counter("stats", "even-items");
PCollection<Integer> input = pipeline.apply(...);
...
input.apply(ParDo.of(new DoFn<Integer, Integer>() {
@ProcessElement
public void processElement(ProcessContext c) {
if (c.element() % 2 == 0) {
counter.inc();
}
});
Python
class FilterTextFn(beam.DoFn):
"""A DoFn that filters for a specific key based on a regex."""
def __init__(self, pattern):
self.pattern = pattern
# A custom metric can track values in your pipeline as it runs. Create
# custom metrics to count unmatched words, and know the distribution of
# word lengths in the input PCollection.
self.word_len_dist = Metrics.distribution(self.__class__,
'word_len_dist')
self.unmatched_words = Metrics.counter(self.__class__,
'unmatched_words')
def process(self, element):
word = element
self.word_len_dist.update(len(word))
if re.match(self.pattern, word):
yield element
else:
self.unmatched_words.inc()
filtered_words = (
words | 'FilterText' >> beam.ParDo(FilterTextFn('s.*')))
Go
func addMetricDoFnToPipeline(s beam.Scope, input beam.PCollection) beam.PCollection {
return beam.ParDo(s, &MyMetricsDoFn{}, input)
}
func executePipelineAndGetMetrics(ctx context.Context, p *beam.Pipeline) (metrics.QueryResults, error) {
pr, err := beam.Run(ctx, runner, p)
if err != nil {
return metrics.QueryResults{}, err
}
// Request the metric called "counter1" in namespace called "namespace"
ms := pr.Metrics().Query(func(r beam.MetricResult) bool {
return r.Namespace() == "namespace" && r.Name() == "counter1"
})
// Print the metric value - there should be only one line because there is
// only one metric called "counter1" in the namespace called "namespace"
for _, c := range ms.Counters() {
fmt.Println(c.Namespace(), "-", c.Name(), ":", c.Committed)
}
return ms, nil
}
type MyMetricsDoFn struct {
counter beam.Counter
}
func init() {
beam.RegisterType(reflect.TypeOf((*MyMetricsDoFn)(nil)))
}
func (fn *MyMetricsDoFn) Setup() {
// While metrics can be defined in package scope or dynamically
// it's most efficient to include them in the DoFn.
fn.counter = beam.NewCounter("namespace", "counter1")
}
func (fn *MyMetricsDoFn) ProcessElement(ctx context.Context, v beam.V, emit func(beam.V)) {
// count the elements
fn.counter.Inc(ctx, 1)
emit(v)
}
Risolvere i problemi relativi a pipeline eseguite lentamente o alla mancanza di output
Vedi Risolvere i problemi relativi a job lenti e bloccati.
Errori comuni e azioni da intraprendere
Una volta individuato l'errore che ha causato l'errore della pipeline, consulta la pagina Risoluzione degli errori di Dataflow per indicazioni sulla risoluzione degli errori.