Implementa canalizaciones de Dataflow

En este documento, se proporciona una descripción general de la implementación de canalizaciones y se destacan algunas de las operaciones que puedes realizar en una canalización implementada.

Ejecuta tu canalización

Después de create y probar tu canalización de Apache Beam, ejecuta tu canalización. Puedes ejecutar la canalización de manera local, lo que te permite probar y depurar tu canalización de Apache Beam, o en Dataflow, un sistema de procesamiento de datos disponible para ejecutar canalizaciones de Apache Beam.

Ejecuta de forma local

Ejecuta tu canalización de manera local.

Java

En el siguiente código de ejemplo, tomado de la guía de inicio rápido, se muestra cómo ejecutar la canalización de WordCount de manera local. Para obtener más información, consulta cómo ejecutar tu canalización de Java de forma local.

En tu terminal, ejecuta el siguiente comando:

  mvn compile exec:java \
      -Dexec.mainClass=org.apache.beam.examples.WordCount \
      -Dexec.args="--output=counts"

Python

En tu terminal, ejecuta el siguiente comando:

python -m apache_beam.examples.wordcount \ --output outputs

Go

En tu terminal, ejecuta el siguiente comando:

    go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \ --output outputs

Aprende a ejecutar la canalización de manera local, en tu máquina, mediante el ejecutor directo.

Ejecuta en Dataflow

Ejecuta tu canalización en Dataflow.

Java

En el siguiente código de ejemplo, tomado de la guía de inicio rápido, se muestra cómo ejecutar la canalización de WordCount en Dataflow. Para obtener más información, consulta cómo ejecutar tu canalización de Java en Dataflow.

En tu terminal, ejecuta el siguiente comando (desde tu directorio word-count-beam):

  mvn -Pdataflow-runner compile exec:java \
    -Dexec.mainClass=org.apache.beam.examples.WordCount \
    -Dexec.args="--project=PROJECT_ID \
    --gcpTempLocation=gs://BUCKET_NAME/temp/ \
    --output=gs://BUCKET_NAME/output \
    --runner=DataflowRunner \
    --region=REGION"

Reemplaza lo siguiente:

PROJECT_ID: El ID del proyecto de Google Cloud.
BUCKET_NAME: Es el nombre de tu bucket de Cloud Storage.
REGION: Es una región de Dataflow, como us-central1

Python

En tu terminal, ejecuta el siguiente comando:

python -m apache_beam.examples.wordcount \
    --region DATAFLOW_REGION \
    --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://STORAGE_BUCKET/results/outputs \
    --runner DataflowRunner \
    --project PROJECT_ID \
    --temp_location gs://STORAGE_BUCKET/tmp/

Reemplaza lo siguiente:

DATAFLOW_REGION: Es la región en la que deseas implementar el trabajo de Dataflow, por ejemplo, europe-west1
La opción --region anula la región predeterminada que está configurada en el servidor de metadatos, el cliente local o las variables de entorno.
STORAGE_BUCKET: Es el nombre de Cloud Storage que copiaste antes.
PROJECT_ID por el ID del proyecto de Google Cloud que copiaste antes

Go

En tu terminal, ejecuta el siguiente comando:

  posix-terminal go run wordcount.go --input gs://dataflow-samples/shakespeare/kinglear.txt \
    --output gs://STORAGE_BUCKET/results/outputs \
    --runner dataflow \
    --project PROJECT_ID \
    --region DATAFLOW_REGION \
    --staging_location gs://STORAGE_BUCKET/binaries/

Reemplaza lo siguiente:

STORAGE_BUCKET: el nombre del bucket de Cloud Storage.
PROJECT_ID: el ID del proyecto de Google Cloud.
DATAFLOW_REGION: la región en la que deseas implementar el trabajo de Dataflow. Por ejemplo, europe-west1. Para obtener una lista de las ubicaciones disponibles, consulta Ubicaciones de Dataflow. Ten en cuenta que la marca --region anula la región predeterminada que está configurada en el servidor de metadatos, el cliente local o las variables de entorno.

Obtén información sobre cómo ejecutar tu canalización en el servicio de Dataflow mediante el ejecutor de Dataflow.

Cuando ejecutas la canalización en Dataflow, Dataflow convierte el código de la canalización de Apache Beam en un trabajo de Dataflow. Dataflow administra por completo los servicios de Google Cloud, como Compute Engine y Cloud Storage, para ejecutar el trabajo de Dataflow y, de forma automática, inicia y elimina los recursos necesarios. Puedes obtener más información sobre cómo Dataflow convierte tu código de Apache Beam en un trabajo de Dataflow en Ciclo de vida de las canalizaciones.

Validación de la canalización

Cuando ejecutas tu canalización en Dataflow, antes de que se inicie el trabajo, Dataflow realiza pruebas de validación en la canalización. Cuando una prueba de validación encuentra problemas con la canalización, Dataflow falla en el envío del trabajo de manera anticipada. En los registros del trabajo, Dataflow incluye mensajes con el siguiente texto. Cada mensaje también incluye detalles sobre los resultados de la validación y las instrucciones para resolver el problema.

The preflight pipeline validation failed for job JOB_ID.

Las pruebas de validación que se ejecuten dependen de los recursos y servicios que usa tu trabajo de Dataflow.

Si la API de Service Usage está habilitada para tu proyecto, las pruebas de validación de canalización verifican si los servicios necesarios para ejecutar el trabajo de Dataflow están habilitados.
Si la API de Cloud Resource Manager está habilitada para tu proyecto, las pruebas de validación de canalización verifican si tienes las configuraciones a nivel de proyecto necesarias para ejecutar el trabajo de Dataflow.

Para obtener más información sobre cómo habilitar servicios, consulta Habilita e inhabilita servicios.

Para obtener información sobre cómo resolver problemas de permisos detectados durante la validación de la canalización, consulta Error en la validación de la canalización.

Si deseas anular la validación de la canalización y, luego, iniciar tu trabajo con errores de validación, usa la siguiente opción de servicio de canalización:

Java

--dataflowServiceOptions=enable_preflight_validation=false

Python

--dataflow_service_options=enable_preflight_validation=false

Go

--dataflow_service_options=enable_preflight_validation=false

Configura las opciones de canalización

Puedes controlar algunos aspectos sobre cómo Dataflow ejecuta el trabajo si configuras las opciones de canalización en el código de canalización de Apache Beam. Por ejemplo, puedes usar opciones de canalización para establecer si tu canalización se ejecuta en máquinas virtuales de trabajador, en el backend del servicio de Dataflow o de forma local.

Administra las dependencias de la canalización

Muchas canalizaciones de Apache Beam pueden ejecutarse mediante los entornos de ejecución predeterminados de Dataflow. Sin embargo, algunos casos de uso de procesamiento de datos se benefician del uso de clases o bibliotecas adicionales. En estos casos, es posible que debas administrar tus dependencias de canalización. Para obtener más información sobre cómo administrar dependencias, consulta Administra dependencias de canalizaciones en Dataflow.

Supervisa tu trabajo

Dataflow proporciona visibilidad de tus trabajos a través de herramientas como la interfaz de supervisión de Dataflow y la interfaz de línea de comandos de Dataflow.

Accede a las VMs de trabajador

Puedes ver las instancias de VM de una canalización determinada mediante la consola de Google Cloud. Desde allí, puedes acceder a cada instancia con SSH. Sin embargo, después de que el trabajo se completa o falla, el servicio de Dataflow cierra y limpia las instancias de VM de forma automática.

Optimizaciones de trabajos

Además de administrar los recursos de Google Cloud, Dataflow realiza y optimiza de forma automática muchos aspectos del procesamiento paralelo distribuido.

Paralelización y distribución

Dataflow particiona los datos de forma automática y distribuye el código de trabajador a las instancias de Compute Engine para el procesamiento paralelo. Para obtener más información, consulta Paralelización y distribución.

Optimizaciones de fusiones y combinaciones

Dataflow usa el código de canalización para crear un grafo de ejecución que representa las transformaciones y las PCollection de la canalización. Además, optimiza el grafo para lograr el rendimiento y el uso de recursos más eficientes. Dataflow también optimiza de forma automática las operaciones que pueden ser costosas, como la agregación de datos. Para obtener más información, consulta Optimización de fusiones y Optimización de combinaciones.

Características de ajuste automático.

El servicio de Dataflow incluye varias características que proporcionan ajustes sobre la marcha de la asignación de recursos y la partición de datos. Estas características ayudan a que Dataflow ejecute el trabajo de la manera más rápida y eficiente posible. Entre estas características, se incluyen las siguientes:

Streaming Engine

De forma predeterminada, el ejecutor de canalizaciones de Dataflow ejecuta los pasos de la canalización de transmisión por completo en máquinas virtuales de trabajador, lo que consume CPU, memoria y almacenamiento en Persistent Disk del trabajador. Streaming Engine de Dataflow retira la ejecución de canalizaciones de las VMs de trabajador y la traslada al backend del servicio de Dataflow. Para obtener más información, consulta Streaming Engine.

Programación flexible de recursos de Dataflow

Dataflow FlexRS reduce los costos de procesamiento por lotes mediante el uso de técnicas de programación avanzadas, el servicio Dataflow Shuffle y una combinación de instancias de máquinas virtuales (VMs) interrumpibles y de VMs normales. Mediante la ejecución de VMs interrumpibles y VMs normales en paralelo, Dataflow mejora la experiencia del usuario en caso de que Compute Engine detenga las instancias de VMs interrumpibles durante un evento del sistema. FlexRS ayuda a garantizar que la canalización progrese y no pierdas trabajo anterior cuando Compute Engine interrumpe las VMs interrumpibles. Para obtener más información sobre FlexRS, consulta Usa la programación flexible de recursos en Dataflow.

VM protegida de Dataflow

A partir del 1 de junio de 2022, el servicio de Dataflow usa VMs protegidas para todos los trabajadores. Para obtener más información sobre las capacidades de las VMs protegidas, consulta VM protegida.