Solución de problemas y depuración de canalizaciones

En esta página, se proporcionan sugerencias para la solución de problemas y estrategias de depuración que pueden resultarte útiles si tienes problemas cuando compilas o ejecutas una canalización de Dataflow. Esta información puede ayudarte a detectar un error en la canalización, determinar el motivo por el cual falla una ejecución de la canalización y sugerir algunas acciones para corregir el problema.

En el diagrama siguiente, se muestra el flujo de trabajo de solución de problemas de Dataflow que se describe en esta página.

Dataflow proporciona comentarios en tiempo real sobre tu trabajo, y existe un conjunto básico de pasos que puedes usar a fin de verificar los mensajes de error, los registros y los estados, como un retraso en el progreso de tu trabajo.

Para obtener asesoramiento sobre los errores comunes que puedes encontrar cuando ejecutas tu trabajo de Dataflow, consulta Solución de errores de Dataflow. Para supervisar y solucionar problemas del rendimiento de la canalización, consulta Supervisa el rendimiento de la canalización.

Prácticas recomendadas para las canalizaciones

Las siguientes son las prácticas recomendadas para las canalizaciones de Java, Python y Go.

Java

• En el caso de los trabajos por lotes, te recomendamos que establezcas un tiempo de actividad (TTL) para la ubicación temporal.

• Antes de configurar el TTL y como práctica recomendada general, asegúrate de establecer la ubicación de etapa de pruebas y la ubicación temporal en diferentes ubicaciones.

• No borres los objetos en la ubicación de etapa de pruebas, ya que estos se reutilizan.

• Si un trabajo se completa o se detiene y los objetos temporales no se limpian, quita estos archivos de forma manual del bucket de Cloud Storage que se usa como ubicación temporal.

Python

Las ubicaciones temporales y de etapa de pruebas tienen el prefijo <job_name>.<time>.

• Asegúrate de configurar la ubicación temporal y de etapa de pruebas en ubicaciones diferentes.

• Si es necesario, borra los objetos en la ubicación de etapa de pruebas después de que se completa o se detiene un trabajo. Además, los objetos publicados en etapa de pruebas no se reutilizan en las canalizaciones de Python.

• Si un trabajo finaliza y los objetos temporales no se limpian, quita estos archivos de forma manual del bucket de Cloud Storage que se usa como ubicación temporal.

• En los trabajos por lotes, te recomendamos que establezcas un tiempo de actividad (TTL) para las ubicaciones temporales y de etapa de pruebas.

Go

• Las ubicaciones temporales y de etapa de pruebas tienen el prefijo <job_name>.<time>.

• Asegúrate de configurar la ubicación temporal y de etapa de pruebas en ubicaciones diferentes.

• Si es necesario, borra los objetos en la ubicación de etapa de pruebas después de que se completa o se detiene un trabajo. Además, los objetos publicados en etapa de pruebas no se reutilizan en las canalizaciones de Go.

• Si un trabajo finaliza y los objetos temporales no se limpian, quita estos archivos de forma manual del bucket de Cloud Storage que se usa como ubicación temporal.

• En los trabajos por lotes, te recomendamos que establezcas un tiempo de actividad (TTL) para las ubicaciones temporales y de etapa de pruebas.

Verifica el estado de la canalización

Puedes detectar cualquier error en la ejecución de tu canalización mediante la interfaz de supervisión de Dataflow.

1. Ve a la consola de Google Cloud.
2. Selecciona tu proyecto de Google Cloud de la lista de proyectos.
3. En el menú de navegación, en Macrodatos, haz clic en Dataflow. Aparece una lista de proyectos en ejecución en el panel de la derecha.
4. Selecciona el trabajo de canalización que deseas ver. Puedes observar rápidamente el estado del trabajo en el campo Estado (Status): “En ejecución” (Running), “Finalizado correctamente” (Succeeded) o “Con errores” (Failed).

Figura 1: Una lista de trabajos de Dataflow en Developers Console con trabajos en los estados en ejecución, finalizado de forma correcta y con errores.

Obtén información sobre errores de canalización

Si uno de tus trabajos de canalización falla, puedes seleccionar el trabajo para ver información más detallada sobre los errores y resultados de la ejecución. Cuando seleccionas un trabajo, puedes ver los gráficos clave de tu canalización, el grafo de ejecución, el panel Información del trabajo y el panel Registros con las pestañas Registros de trabajo, Registros de trabajador, Diagnóstico y Recomendaciones.

Verifica los mensajes de error del trabajo

Para ver los Registros del trabajo generados por el código de canalización y el servicio de Dataflow, en el panel Registros, haz clic enMostrar.

Para filtrar los mensajes que aparecen en Registros de trabajos, haz clic en Información y Filtro. Para mostrar solo mensajes de error, haz clic en Información y selecciona Error.

Para expandir un mensaje de error, haz clic en la sección desplegable .

También puedes hacer clic en la pestaña Diagnóstico. En esta pestaña, se muestra dónde se produjeron errores durante el cronograma elegido, un recuento de todos los errores registrados y las recomendaciones posibles para tu canalización.

Ve los registros de pasos para el trabajo

Cuando seleccionas un paso en tu grafo de canalización, en el panel de registros, se muestran de forma alternada los Registros de trabajos que genera el servicio de Dataflow y los registros de las instancias de Compute Engine que ejecutan tu paso de canalización.

Cloud Logging combina todos los registros recopilados de las instancias de Compute Engine de tu proyecto en una ubicación. Consulta Registra mensajes de canalizaciones para obtener más información sobre el uso de las diversas capacidades de registro de Dataflow.

Controla el rechazo automático de la canalización

En algunos casos, el servicio de Dataflow identifica que tu canalización puede provocar problemas del SDK conocidos. A fin de evitar el envío de canalizaciones que puedan generar problemas, Dataflow rechaza la canalización de forma automática y muestra el siguiente mensaje:

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].

Si, después de leer las advertencias en los detalles de error vinculados, quieres seguir tratando de ejecutar tu canalización, puedes anular el rechazo automático. Agrega la marca --experiments=<override-flag> y vuelve a enviar tu canalización.

Determina la causa de una falla de canalización

Comúnmente, una ejecución con errores de la canalización en Apache Beam puede atribuirse a una de las siguientes causas:

• Errores de construcción de grafo o canalización. Estos errores ocurren cuando Dataflow se ejecuta con un problema y compila el grafo de pasos que compone tu canalización, como se describe en la canalización de Apache Beam.
• Errores en la validación de un trabajo. El servicio de Dataflow valida cualquier trabajo de canalización que inicies. Los errores en el proceso de validación pueden evitar que tu trabajo se cree o ejecute de forma correcta. Los errores de validación pueden incluir problemas con el bucket de Cloud Storage del proyecto de Google Cloud o con los permisos de tu proyecto.
• Excepciones en el código del trabajador. Estos errores se producen cuando hay errores en el código que proporciona el usuario y que Dataflow distribuye a trabajadores paralelos, como las instancias de DoFn de una transformación de ParDo.
• Errores causados por fallas transitorias en otros servicios de Google Cloud. La canalización puede fallar debido a una interrupción temporal o a otro problema en los servicios de Google Cloud que usa Dataflow, como Compute Engine o Cloud Storage.

Detecta errores de construcción de canalización o grafo

Un error de construcción de grafo puede producirse cuando Dataflow compila el grafo de ejecución para la canalización a partir del código en tu programa de Dataflow. Durante el tiempo de construcción del grafo, Dataflow verifica la existencia de operaciones ilegales.

Si Dataflow detecta un error en la construcción del grafo, recuerda que no se crea ningún trabajo en el servicio de Dataflow. Por lo tanto, no verás comentarios en la interfaz de supervisión de Dataflow. En cambio, aparece un mensaje de error similar al siguiente en la ventana de terminal o consola en la que ejecutaste tu canalización de Apache Beam:

Java

Por ejemplo, si la canalización intenta realizar una agregación como GroupByKey en una PCollection no activada y no delimitada de un sistema de ventanas global, aparecerá un mensaje de error similar al siguiente:

...
... 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

Por ejemplo, si tu canalización usa sugerencias de tipo y el tipo de argumento en una de las transformaciones no es el esperado, aparece un mensaje de error similar al siguiente:

... 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

Por ejemplo, si tu canalización usa un “DoFn” que no toma ninguna entrada, se produce un mensaje de error similar al siguiente:

... 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)

Si te encuentras con este error, verifica el código de tu canalización para asegurarte de que la operación de tu canalización sea legal.

Detecta errores en la validación de trabajo de Dataflow

Una vez que el servicio de Dataflow recibe el grafo de tu canalización, el servicio intenta validar el trabajo. Esta validación incluye las siguientes acciones:

• Asegurarse de que el servicio pueda acceder a los depósitos de Cloud Storage relacionados con tu trabajo para la etapa de pruebas de archivos y el resultado temporal.
• Verificar los permisos necesarios en tu proyecto de Google Cloud.
• Garantizar que el servicio pueda acceder a las fuentes de entrada y salida, como los archivos.

Si el trabajo no pasa el proceso de validación, aparecerá un mensaje de error en la interfaz de supervisión de Dataflow, así como en la ventana de terminal o consola si usas la ejecución de bloqueo. El mensaje de error es similar al siguiente:

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

Por el momento, la validación del trabajo que se describe en esta sección no es compatible con Go. Los errores debido a estos problemas aparecen como excepciones de trabajador.

Detecta una excepción en un código de trabajador

Mientras se ejecuta el trabajo, puede experimentar errores o excepciones en el código del trabajador. En general, estos errores indican que los DoFn de tu código de canalización generaron excepciones no controladas, que provocaron la falla de tareas en tu trabajo de Dataflow.

Las excepciones en el código del usuario (por ejemplo, tus instancias de DoFn) se informan en la interfaz de supervisión de Dataflow. Si ejecutas la canalización con una ejecución de bloqueo, se imprimen mensajes de error en la ventana de terminal o consola, como el siguiente:

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.

Nota: El servicio de Dataflow vuelve a intentar las tareas con errores hasta 4 veces en el modo por lotes y una cantidad ilimitada de veces en el modo de transmisión. En el modo por lotes, tu trabajo falla; en el modo de transmisión, podría detenerse de forma indefinida.

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'

Nota: El servicio de Dataflow vuelve a intentar las tareas con errores hasta 4 veces en el modo por lotes y una cantidad ilimitada de veces en el modo de transmisión. En el modo por lotes, tu trabajo falla; en el de transmisión, podría detenerse de forma indefinida.

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!

Nota: El servicio de Dataflow vuelve a intentar las tareas con errores hasta 4 veces en el modo por lotes y una cantidad ilimitada de veces en el modo de transmisión. En el modo por lotes, tu trabajo falla; en el de transmisión, podría detenerse de forma indefinida.

Se recomienda que agregue controladores de excepciones para protegerse contra los errores del código. Por ejemplo, si deseas descartar elementos que producen fallas en una validación de entrada personalizada en ParDo, controla la excepción en tu DoFn y quita el elemento.

También puedes hacer seguimiento de los elementos con errores de diferentes formas:

• Puedes registrar los elementos con errores y verificar la salida mediante Cloud Logging.
• Puedes verificar los registros de inicio del trabajador y del trabajador de Dataflow en busca de advertencias o errores si sigues las instrucciones en Visualizar registros.
• Puedes hacer que tu ParDo escriba los elementos con errores en un resultado adicional para analizarlo más tarde.

Para hacer un seguimiento de las propiedades de una canalización en ejecución, puedes usar la clase Metrics, como se muestra en el siguiente ejemplo:

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)
}

Solución de problemas de canalizaciones de ejecución lenta o falta de resultado

Consulta Soluciona problemas de trabajos lentos y atascados.

Errores comunes y acciones

Cuando sepas qué error causó la falla de la canalización, consulta la página Soluciona errores de Dataflow para obtener orientación sobre la solución de errores.