Actualiza una canalización de transmisión

En esta página, se proporcionan orientación y recomendaciones para actualizar tus canalizaciones de transmisión. Por ejemplo, es posible que debas actualizar a una versión más reciente del SDK de Apache Beam o que desees actualizar el código de tu canalización. Se proporcionan diferentes opciones para adaptarse a distintas situaciones.

Mientras que las canalizaciones por lotes se detienen cuando el trabajo se completa, las canalizaciones de transmisión a menudo se ejecutan de forma continua para proporcionar un procesamiento sin interrupciones. Por lo tanto, cuando actualices las canalizaciones de transmisión, debes tener en cuenta las siguientes consideraciones:

• Es posible que debas minimizar o evitar interrupciones en la canalización. En algunos casos, es posible que puedas tolerar una interrupción temporal del procesamiento mientras se implementa una nueva versión de una canalización. En otros casos, es posible que tu aplicación no pueda tolerar ninguna interrupción.
• Los procesos de actualización de la canalización deben controlar los cambios de esquema de una manera que minimice la interrupción del procesamiento de mensajes y de otros sistemas adjuntos. Por ejemplo, si el esquema para los mensajes en una canalización de procesamiento de eventos cambia, los cambios en el esquema también pueden ser necesarios en los receptores de datos descendentes.

Puedes usar uno de los siguientes métodos para actualizar las canalizaciones de transmisión, según los requisitos de actualización y canalización:

• Realizar actualizaciones en tránsito
• Inicia un trabajo de reemplazo
• Ejecuta canalizaciones paralelas

Para obtener más información sobre los problemas que podrías encontrar durante una actualización y cómo evitarlos, consulta Valida un trabajo de reemplazo y Verifica la compatibilidad de trabajos.

Prácticas recomendadas

• Actualiza la versión del SDK de Apache Beam por separado de cualquier cambio en el código de canalización.
• Prueba tu canalización después de cada cambio antes de realizar actualizaciones adicionales.
• Actualiza con regularidad la versión del SDK de Apache Beam que usa tu canalización.
• Usa métodos automatizados siempre que sea posible, como las actualizaciones en tránsito o las actualizaciones automatizadas de canalizaciones paralelas.
• Usa E/S administrada cuando sea posible para obtener los beneficios de las actualizaciones automáticas de las versiones del conector.

Realiza actualizaciones en tránsito

Puedes actualizar algunas canalizaciones de transmisión en curso sin detener el trabajo. Esta situación se conoce como actualización de trabajo en tránsito. Las actualizaciones de trabajos en tránsito solo están disponibles en circunstancias limitadas:

• El trabajo debe usar Streaming Engine.
• El trabajo debe estar en estado de ejecución.
• Solo estás cambiando la cantidad de trabajadores que usa el trabajo.

Para obtener más información, consulta Configura el rango de ajuste de escala automático en la página Ajuste de escala automático horizontal.

Para obtener instrucciones que expliquen cómo realizar una actualización de trabajo en tránsito, consulta Actualiza una canalización existente.

Creación o actualización automática (upsert) de plantillas

Cuando inicias canalizaciones con una plantilla (plantillas clásicas, plantillas de Flex, Terraform o Config Connector), puedes usar el experimento create_or_update_job para usar la funcionalidad de creación o actualización (upsert).

Cuando especificas create_or_update_job en el parámetro additional_experiments o la marca additional-experiments, sucede lo siguiente:

• Si ya existe un trabajo en ejecución o en proceso de vaciado con el nombre de trabajo especificado, el servicio de plantillas inicia automáticamente el trabajo nuevo como una actualización del trabajo existente.
• Si no existe ningún trabajo activo con ese nombre, el servicio de plantillas inicia el trabajo nuevo como una creación de trabajo nuevo.

Este experimento elimina la necesidad de determinar de forma programática si se debe usar la acción de la API de creación o actualización cuando se lanza una plantilla.

Para ver muestras de código de Terraform y Config Connector que usan este experimento, consulta las siguientes secciones:

• Envía una solicitud de actualización automática de detención y reemplazo
• Envía una solicitud de actualización de canalización paralela automatizada

Inicia un trabajo de reemplazo

Si el trabajo actualizado es compatible con el trabajo existente, puedes actualizar la canalización mediante la opción update. Cuando reemplazas un trabajo existente, uno nuevo ejecuta el código de la canalización actualizada. El servicio de Dataflow retiene el nombre del trabajo, pero ejecuta el trabajo de reemplazo con un ID de trabajo actualizado. Este proceso puede causar tiempo de inactividad mientras se detiene el trabajo existente, se ejecuta la verificación de compatibilidad y comienza el trabajo nuevo. Para obtener más detalles, consulta Los efectos de reemplazar un trabajo.

Dataflow realiza una verificación de compatibilidad para garantizar que el código de la canalización actualizada se pueda implementar de forma segura en la canalización en ejecución. Ciertos cambios en el código hacen que falle la verificación de compatibilidad, como cuando se agregan o quitan entradas complementarias desde un paso existente. Cuando falla la verificación de compatibilidad, no puedes realizar una actualización de trabajo local.

Para obtener instrucciones que expliquen cómo iniciar un trabajo de reemplazo, consulta Inicia un trabajo de reemplazo.

Si la actualización de la canalización no es compatible con el trabajo actual, debes detener y reemplazar la canalización. Si tu canalización no puede tolerar el tiempo de inactividad, ejecuta canalizaciones paralelas.

Detención y reemplazo manual

Para detener y reemplazar manualmente, cancela o desvía la canalización y, luego, reemplázala por la canalización actualizada. Cancelar una canalización hace que Dataflow detenga de inmediato el procesamiento y apague los recursos lo más rápido posible, lo que puede causar una pérdida de los datos que se procesan, conocidos como datos en tránsito. Para evitar la pérdida de datos, en la mayoría de los casos, se prefiere el desvío. También puedes usar instantáneas de Dataflow para guardar el estado de una canalización de transmisión, lo que te permite iniciar una versión nueva de tu trabajo de Dataflow sin perder el estado. Para obtener más información, consulta Usa instantáneas de Dataflow.

Desviar una canalización cierra de inmediato cualquier ventana en proceso y activa todos los activadores. Aunque los datos en tránsito no se pierden, el desvío puede provocar que las ventanas tengan datos incompletos. Si esto sucede, las ventanas en proceso emiten resultados parciales o incompletos. Para obtener más información, consulta Efectos de la desviación de un trabajo. Una vez que se completa el trabajo existente, puedes iniciar un trabajo de transmisión nuevo que contenga el código de la canalización actualizada, lo que permite que se reanude el procesamiento.

Con este método, incurres en un tiempo de inactividad entre el momento en que se detiene el trabajo de transmisión existente y el momento en que la canalización de reemplazo está lista para reanudar el procesamiento de datos. Sin embargo, cancelar o desviar una canalización existente y, luego, iniciar un trabajo nuevo con la canalización actualizada es menos complicado que ejecutar canalizaciones paralelas.

Para obtener más información, consulta Desvía un trabajo de Dataflow. Después de desviar el trabajo actual, inicia uno nuevo con el mismo nombre.

Detención y reemplazo automatizados

Dataflow proporciona compatibilidad con la API para iniciar una actualización automática de detención y reemplazo. Este flujo de trabajo de estilo declarativo elimina los pasos manuales del procedimiento. Declaras el trabajo que se reemplazará, y el nuevo trabajo se inicia y coordina la transición automáticamente.

Cuando usas este flujo de trabajo, se aprovisionan recursos de trabajo nuevos mientras el trabajo anterior aún se está ejecutando. Luego, el trabajo anterior recibe automáticamente un indicador de vaciado. Después de que el trabajo anterior termina de vaciarse o alcanza un tiempo de espera especificado por el usuario, el trabajo nuevo comienza a procesar datos de inmediato. Usa este flujo de trabajo para las canalizaciones que no pueden tolerar datos duplicados ni agregaciones parciales, pero que pueden aceptar una breve pausa en el procesamiento mientras se vacía el trabajo anterior.

Envía una solicitud de actualización automática de detener y reemplazar

Para usar este flujo de trabajo, sigue estos pasos:

• Debes establecer la opción parallel_replace_job_max_stop_duration.
• No debes establecer la opción parallel_replace_job_min_parallel_pipelines_duration. Si estableces una duración paralela, se activará el flujo de trabajo de actualizaciones automáticas de la canalización paralela.

Inicia una solicitud de actualización automatizada de detención y reemplazo con las siguientes opciones de servicio:

Java

Opción 1: Actualiza con el mismo nombre de trabajo

Nota: Si realizas la actualización con el mismo nombre de trabajo (opción 1), el nuevo trabajo debe comenzar al menos dos minutos después del inicio del trabajo anterior. De lo contrario, la actualización fallará con un error. Esta restricción evita las actualizaciones paralelas duplicadas causadas por reintentos de la biblioteca cliente o llamadas a procedimientos remotos desactualizadas.

--update \
--dataflowServiceOptions="update_strategy_parallel_job_update" \
--dataflowServiceOptions="parallel_replace_job_max_stop_duration=DURATION"

• Para realizar una actualización automatizada de detención y reemplazo con el mismo nombre, usa la marca --update y la opción update_strategy_parallel_job_update.
• Para realizar una actualización en un solo lugar, usa update_strategy_in_place_update.

Opción 2: Realiza la actualización con un nombre de trabajo diferente

--dataflowServiceOptions="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflowServiceOptions="parallel_replace_job_max_stop_duration=DURATION"

• Para especificar el trabajo anterior por ID en lugar de por nombre, usa --dataflowServiceOptions="parallel_replace_job_id=OLD_JOB_ID".
• Si especificas un nombre de trabajo nuevo y usas la marca --update, Dataflow buscará un trabajo existente con el nombre nuevo, lo que provocará un error.

Opcional: Inhabilita la cancelación automática

La cancelación automática está habilitada de forma predeterminada cuando especificas la opción parallel_replace_job_max_stop_duration. Para inhabilitar la cancelación automática, configura la opción parallel_replace_job_cancel_on_drain_timeout en false.

--dataflowServiceOptions="parallel_replace_job_cancel_on_drain_timeout=false"

Si inhabilitas la cancelación automática y el trabajo anterior se atasca en el estado de desvío, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.

Python

Opción 1: Actualiza con el mismo nombre de trabajo

Nota: Si realizas la actualización con el mismo nombre de trabajo (opción 1), el nuevo trabajo debe comenzar al menos dos minutos después del inicio del trabajo anterior. De lo contrario, la actualización fallará con un error. Esta restricción evita las actualizaciones paralelas duplicadas causadas por reintentos de la biblioteca cliente o llamadas a procedimientos remotos desactualizadas.

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"

• Para realizar una actualización automatizada de detención y reemplazo con el mismo nombre, usa la marca --update y la opción update_strategy_parallel_job_update.
• Para realizar una actualización en un solo lugar, usa update_strategy_in_place_update.

Opción 2: Realiza la actualización con un nombre de trabajo diferente

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"

• Para especificar el trabajo anterior por ID en lugar de por nombre, usa --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
• Si especificas un nombre de trabajo nuevo y usas la marca --update, Dataflow buscará un trabajo existente con el nombre nuevo, lo que provocará un error.

Opcional: Inhabilita la cancelación automática

La cancelación automática está habilitada de forma predeterminada cuando especificas la opción parallel_replace_job_max_stop_duration. Para inhabilitar la cancelación automática, configura la opción parallel_replace_job_cancel_on_drain_timeout en false.

--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=false"

Si inhabilitas la cancelación automática y el trabajo anterior se atasca en el estado de desvío, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.

Go

Opción 1: Actualiza con el mismo nombre de trabajo

Nota: Si realizas la actualización con el mismo nombre de trabajo (opción 1), el nuevo trabajo debe comenzar al menos dos minutos después del inicio del trabajo anterior. De lo contrario, la actualización fallará con un error. Esta restricción evita las actualizaciones paralelas duplicadas causadas por reintentos de la biblioteca cliente o llamadas a procedimientos remotos desactualizadas.

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"

• Para realizar una actualización automatizada de detención y reemplazo con el mismo nombre, usa la marca --update y la opción update_strategy_parallel_job_update.
• Para realizar una actualización en un solo lugar, usa update_strategy_in_place_update.

Opción 2: Realiza la actualización con un nombre de trabajo diferente

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_max_stop_duration=DURATION"

• Para especificar el trabajo anterior por ID en lugar de por nombre, usa --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
• Si especificas un nombre de trabajo nuevo y usas la marca --update, Dataflow buscará un trabajo existente con el nombre nuevo, lo que provocará un error.

Opcional: Inhabilita la cancelación automática

La cancelación automática está habilitada de forma predeterminada cuando especificas la opción parallel_replace_job_max_stop_duration. Para inhabilitar la cancelación automática, configura la opción parallel_replace_job_cancel_on_drain_timeout en false.

--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=false"

Si inhabilitas la cancelación automática y el trabajo anterior se atasca en el estado de desvío, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.

gcloud

Opción 1: Actualiza con el mismo nombre de trabajo

Nota: Si realizas la actualización con el mismo nombre de trabajo (opción 1), el nuevo trabajo debe comenzar al menos dos minutos después del inicio del trabajo anterior. De lo contrario, la actualización fallará con un error. Esta restricción evita las actualizaciones paralelas duplicadas causadas por reintentos de la biblioteca cliente o llamadas a procedimientos remotos desactualizadas.

--update \
--additional-experiments="update_strategy_parallel_job_update" \
--additional-experiments="parallel_replace_job_max_stop_duration=DURATION"

• Para realizar una actualización automatizada de detención y reemplazo con el mismo nombre, usa la marca --update y la opción update_strategy_parallel_job_update.
• Para realizar una actualización en un solo lugar, usa update_strategy_in_place_update.

Opción 2: Realiza la actualización con un nombre de trabajo diferente

--additional-experiments="parallel_replace_job_name=OLD_JOB_NAME" \
--additional-experiments="parallel_replace_job_max_stop_duration=DURATION"

• Para especificar el trabajo anterior por ID en lugar de por nombre, usa --additional-experiments="parallel_replace_job_id=OLD_JOB_ID".
• Si especificas un nombre de trabajo nuevo y usas la marca --update, Dataflow buscará un trabajo existente con el nombre nuevo, lo que provocará un error.

Opcional: Inhabilita la cancelación automática

La cancelación automática está habilitada de forma predeterminada cuando especificas la opción parallel_replace_job_max_stop_duration. Para inhabilitar la cancelación automática, configura la opción parallel_replace_job_cancel_on_drain_timeout en false.

--additional-experiments="parallel_replace_job_cancel_on_drain_timeout=false"

Si inhabilitas la cancelación automática y el trabajo anterior se atasca en el estado de desvío, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.

Opcional: Upsert (crea o actualiza un trabajo)

Para habilitar el comportamiento de upsert (crear o actualizar el trabajo), haz lo siguiente:

--additional-experiments="create_or_update_job"

Terraform

additional_experiments = [
  "parallel_replace_job_max_stop_duration=DURATION",
  "parallel_replace_job_cancel_on_drain_timeout=true",
  "update_strategy_parallel_job_update",
  "parallel_replace_job_preallocate_compute_resources=true",
  "create_or_update_job"
]

Config Connector

metadata:
  annotations:
    # Force to use only direct controller. Multiple controllers can cause a Dataflow job to enter into a continuous update loop.
    # https://docs.cloud.google.com/config-connector/docs/concepts/controller-types#underlying-controller-types
    alpha.cnrm.cloud.google.com/reconciler: direct
    # Optional but recommended: Dataflow batch jobs do not support the drain operation. But for Dataflow streaming jobs, prefer "drain" over "cancel" as an on-delete option.
    cnrm.cloud.google.com/on-delete: drain
    # Optional but recommended: Set deletion-policy to "abandon" to avoid accidental deletion, this will ignore the on-delete option.
    cnrm.cloud.google.com/deletion-policy: abandon
spec:
  ...
  additionalExperiments:
    - "parallel_replace_job_max_stop_duration=DURATION"
    - "parallel_replace_job_cancel_on_drain_timeout=true"
    - "update_strategy_parallel_job_update"
    - "parallel_replace_job_preallocate_compute_resources=true"
    - "create_or_update_job"

Reemplaza las siguientes variables:

• Debes proporcionar parallel_replace_job_name o parallel_replace_job_id para identificar el trabajo que se reemplazará:
  • OLD_JOB_NAME: Es el nombre del trabajo que se reemplazará.
  • OLD_JOB_ID: Es el ID del trabajo que se reemplazará.
• Debes proporcionar el valor de parallel_replace_job_max_stop_duration para activar la función de detener y reemplazar automática:
  • DURATION: Es la cantidad máxima de tiempo que el trabajo nuevo espera a que el trabajo anterior termine de vaciarse. La duración debe tener el formato de una cadena que termine en s, m o h (por ejemplo, 30m, 1h).
• No establezcas la opción parallel_replace_job_min_parallel_pipelines_duration cuando uses este flujo de trabajo. Si configuras esta opción, se activará el flujo de trabajo de actualizaciones de canalizaciones paralelas automatizadas.
• Opcional: Configura la opción parallel_replace_job_cancel_on_drain_timeout. Dado que la cancelación automática está habilitada (true de forma predeterminada) de forma predeterminada cuando se establece la opción parallel_replace_job_max_stop_duration, no es necesario que configures esta opción de forma explícita para habilitarla.
  • Para mantener el comportamiento predeterminado, omite esta opción o configúrala como true.
  • Para inhabilitar la cancelación automática, establece esta opción en false. Si configuras esta opción como false y el trabajo anterior se atasca en el estado de desvío, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.
• Opcional: Define la configuración de parallel_replace_job_preallocate_compute_resources:
  • Especifica si los trabajadores se aprovisionan por adelantado para el trabajo nuevo mientras se agota el trabajo anterior. Valores: true (predeterminado) o false. En el caso de Terraform y Config Connector, se recomienda establecer esta opción en true para evitar que se agote el tiempo de espera del aprovisionamiento de recursos. Cuando parallel_replace_job_preallocate_compute_resources se establece en false, el trabajo nuevo permanece en estado pendiente hasta que se agota el trabajo anterior.

Reprocesamiento de mensajes con Pub/Sub Snapshot y Seek

En algunas situaciones, después de que reemplazas o cancelas una canalización desviada, es posible que tengas que volver a procesar los mensajes de Pub/Sub enviados con anterioridad. Por ejemplo, es posible que debas usar la lógica empresarial actualizada para volver a procesar los datos. La Pub/Sub Seek es una función que te permite volver a reproducir mensajes de una instantánea de Pub/Sub. Puedes usar Pub/Sub Seek con Dataflow para volver a procesar los mensajes desde el momento en que se creó la instantánea de suscripción.

Durante el desarrollo y las pruebas, también puedes usar Pub/Sub Seek para volver a reproducir los mensajes conocidos de forma repetida para verificar el resultado de tu canalización. Cuando usas Pub/Sub Seek, no busques una instantánea de suscripción cuando una canalización consuma la suscripción. Si lo haces, Seek uede invalidar la lógica de marca de agua de Dataflow y podría afectar el procesamiento único de los mensajes de Pub/Sub.

El siguiente es un flujo de trabajo recomendado de gcloud CLI para usar Pub/Sub Seek con canalizaciones de Dataflow en una ventana de la terminal:

1. Para crear una instantánea de la suscripción, usa el comando gcloud pubsub snapshots create:

   gcloud pubsub snapshots create SNAPSHOT_NAME --subscription=PIPELINE_SUBSCRIPTION_NAME
   

2. Para desviar o cancelar la canalización, usa el comando gcloud dataflow jobs drain o el comando gcloud dataflow jobs cancel:

   gcloud dataflow jobs drain JOB_ID
   

   o

   gcloud dataflow jobs cancel JOB_ID
   

3. Para buscar la instantánea, usa el comando gcloud pubsub subscriptions seek:

   gcloud pubsub subscriptions seek SNAPSHOT_NAME
   

4. Implementa una canalización nueva que consuma la suscripción.

Ejecuta canalizaciones paralelas

Si necesitas evitar interrupciones en tu canalización de transmisión durante una actualización, puedes ejecutar canalizaciones paralelas. Este enfoque te permite iniciar un trabajo de transmisión nuevo con el código de la canalización actualizado y ejecutarlo en paralelo con el trabajo existente. Puedes usar el flujo de trabajo de implementación de actualización de canalización paralela automatizada de Dataflow o realizar los pasos de forma manual.

Descripción general de las canalizaciones paralelas

Cuando crees la canalización nueva, usa la misma estrategia de períodos que usaste para la canalización existente. En el caso del flujo de trabajo manual, permite que la canalización existente se siga ejecutando hasta que su marca de agua supere la marca de tiempo del primer período completo que procesó la canalización actualizada. Luego, desvía o cancela la canalización existente. Si usas el flujo de trabajo automatizado, este trabajo se realiza por ti. La canalización actualizada continúa ejecutándose en su lugar y, por lo tanto, controla el procesamiento por su cuenta.

El diagrama siguiente ilustra este proceso.

En el diagrama, Canalización B es el trabajo actualizado que toma el control de la Canalización A. El valor t es la marca de tiempo del primer período completo que procesó la canalización B. El valor w es la marca de agua de la canalización A. Por motivos de simplicidad, una marca de agua perfecta se da por sentada sin datos tardíos. El procesamiento y el tiempo se representan en el eje horizontal. Ambas canalizaciones usan ventanas fijas de cinco minutos (de saltos de tamaño constante). Los resultados se activan después de que la marca de agua pasa el final de cada período.

Debido a que el resultado simultáneo ocurre durante el período en el que se superponen las dos canalizaciones, configura las dos canalizaciones para escribir resultados en diferentes destinos. Luego, los sistemas descendentes pueden usar una abstracción sobre los dos receptores de destino, como una vista de base de datos, para consultar los resultados combinados. Estos sistemas también pueden usar la abstracción para anular los resultados duplicados del período superpuesto. Para obtener más información, consulta Cómo controlar la salida duplicada.

Limitaciones

El uso de actualizaciones de canalizaciones paralelas manuales o automatizadas tiene las siguientes limitaciones:

• Solo actualizaciones automatizadas: El nuevo trabajo paralelo debe ser un trabajo de Streaming Engine.
• No se permiten trabajos simultáneos con el mismo nombre. Sin embargo, cuando realices una actualización automatizada de detención y reemplazo o una actualización de canalización paralela con el mismo nombre de trabajo, puedes reutilizar el nombre del trabajo. En este caso, el trabajo nuevo debe comenzar al menos dos minutos después del inicio del trabajo anterior. Esta restricción evita que se realicen varias actualizaciones paralelas a partir de reintentos repetidos de la biblioteca cliente o llamadas a procedimientos remotos desactualizadas.
• La ejecución de dos canalizaciones en paralelo en la misma entrada puede generar datos duplicados, agregaciones parciales y posibles problemas de orden cuando se insertan datos en el receptor. El sistema posterior debe diseñarse para anticipar y administrar estos resultados.
• Cuando se lee desde una fuente de Pub/Sub, no se recomienda usar la misma suscripción para varias canalizaciones, ya que se pueden generar problemas de precisión. Sin embargo, en algunos casos de uso, como las canalizaciones de extracción, transformación y carga (ETL), usar la misma suscripción en dos canalizaciones puede reducir la duplicación. Es probable que haya problemas con el ajuste de escala automático cada vez que proporciones un valor distinto de cero para la duración superpuesta. Esto se puede mitigar con la función de actualización de trabajo en tránsito. Para obtener más información, consulta Optimiza el ajuste de escala automático para tus canalizaciones de transmisión de Pub/Sub.
• En el caso de Apache Kafka, puedes minimizar los duplicados habilitando la confirmación de desplazamiento en Kafka. Para habilitar la confirmación de desplazamiento en Kafka, consulta Confirmación en Kafka.

Actualizaciones automatizadas de canalizaciones paralelas

Dataflow proporciona compatibilidad con la API para iniciar un trabajo de reemplazo paralelo. Esta API de estilo declarativo abstrae el trabajo manual de ejecutar pasos de procedimiento. Declaras el trabajo que deseas actualizar y, luego, se ejecuta un trabajo nuevo en paralelo con el anterior. Después de que el trabajo nuevo se ejecute durante el período que especifiques, se detendrá el trabajo anterior. Esta función elimina las pausas de procesamiento durante las actualizaciones. También reduce el esfuerzo operativo necesario para actualizar las canalizaciones incompatibles.

Este método de actualización es ideal para las canalizaciones que pueden tolerar algunos duplicados o agregaciones parciales y no requieren un orden estricto durante la inserción de datos. Es adecuado para las canalizaciones de ETL, así como para las canalizaciones que usan el modo de transmisión al menos una vez y la transformación Redistribute con la opción allow duplicates establecida en true.

Opciones de servicio de canalización paralela automatizada

Nota: Debes establecer la opción parallel_replace_job_min_parallel_pipelines_duration, independientemente de la estrategia de actualización que elijas (opción 1: actualizar con el mismo nombre de trabajo o opción 2: actualizar con un nombre de trabajo diferente).

Usa las siguientes opciones de servicio para las actualizaciones de canalizaciones paralelas automatizadas:

Opción de servicio	Opcional u obligatorio	Descripción	Dependencias o exclusiones
update_strategy_parallel_job_update	Obligatorio (opción 1: Actualiza con el mismo nombre de trabajo)	Comando para realizar una actualización paralela, que ejecuta ambas canalizaciones de forma simultánea para minimizar el tiempo de inactividad, cuando se actualiza con el mismo nombre de trabajo.	Se debe configurar junto con las marcas --update y parallel_replace_job_min_parallel_pipelines_duration.
update_strategy_in_place_update	Optional	Es una alternativa a la actualización en paralelo. Realiza una actualización in situ estándar del trabajo.	Se debe configurar junto con la marca --update. Este campo y update_strategy_parallel_job_update son mutuamente excluyentes. Cuando se establece esta opción, se ignoran otras opciones relacionadas con los trabajos paralelos.
parallel_replace_job_min_parallel_pipelines_duration	Obligatorio	Especifica la duración mínima durante la que se ejecutan las dos canalizaciones de forma simultánea. Una vez que transcurre este período, se envía una señal de vaciado al trabajo anterior. Los valores aceptables varían de 0s (recomendado para una superposición nula) a 744h (31 días).	Debe estar vinculado con una forma de segmentar el trabajo anterior. Debe ser uno de los siguientes: Opción 1: Usar el mismo nombre de trabajo: update_strategy_parallel_job_update, o Opción 2: Usar un nombre de trabajo diferente: parallel_replace_job_name (o, como alternativa, usar parallel_replace_job_id para identificar el trabajo)
parallel_replace_job_name o parallel_replace_job_id (elige una)	Obligatorio (opción 2: Actualiza con un nombre de trabajo diferente)	Identifica el trabajo anterior por nombre o ID para reemplazarlo durante una actualización de nombre diferente.	Se requiere que se establezca parallel_replace_job_min_parallel_pipelines_duration. No uses la marca --update o parallel_replace_job_id con esta opción.
parallel_replace_job_max_stop_duration	Optional	Es la duración máxima que se permite que el trabajo anterior se agote antes de que se active la cancelación automática. Por ejemplo, 30m o 1h.	Requiere configurar un flujo de trabajo de actualización paralelo (opción 1 o 2).
parallel_replace_job_cancel_on_drain_timeout	Optional El valor predeterminado es true si se establece una duración máxima de detención.	Opción booleana que especifica si se debe cancelar el trabajo anterior si su duración de vaciado supera parallel_replace_job_max_stop_duration.	Se usa junto con parallel_replace_job_max_stop_duration. Establece el valor en false para inhabilitar la cancelación automática. Si inhabilitas la cancelación automática y el trabajo anterior se atasca en el estado de vaciado, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.

Envía una solicitud de actualización de canalización paralela automatizada

Para usar el flujo de trabajo automatizado, inicia un nuevo trabajo de transmisión. Puedes actualizar un trabajo con el mismo nombre o con uno diferente.

Java

Opción 1: Actualiza con el mismo nombre de trabajo

Nota: Si realizas la actualización con el mismo nombre de trabajo (opción 1), el nuevo trabajo debe comenzar al menos dos minutos después del inicio del trabajo anterior. De lo contrario, la actualización fallará con un error. Esta restricción evita las actualizaciones paralelas duplicadas causadas por reintentos de la biblioteca cliente o llamadas a procedimientos remotos desactualizadas.

--update \
--dataflowServiceOptions="update_strategy_parallel_job_update" \
--dataflowServiceOptions="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• Para realizar una actualización paralela con el mismo nombre, usa la marca --update y la opción update_strategy_parallel_job_update.
• Para realizar una actualización en un solo lugar sin quitar las opciones relacionadas con los trabajos paralelos, usa update_strategy_in_place_update en lugar de update_strategy_parallel_job_update.

Opción 2: Realiza la actualización con un nombre de trabajo diferente

--dataflowServiceOptions="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflowServiceOptions="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• Para especificar el trabajo anterior por ID en lugar de por nombre, usa --dataflowServiceOptions="parallel_replace_job_id=OLD_JOB_ID".
• Si especificas un nombre de trabajo nuevo y usas la marca --update, Dataflow buscará un trabajo existente con el nombre nuevo, lo que provocará un error.

Opcional: Configura el tiempo de espera para el agotamiento y la cancelación automática

Puedes agregar las siguientes opciones a cualquiera de las configuraciones para establecer un tiempo de espera de vaciado y cancelar automáticamente el trabajo anterior si se bloquea.

--dataflowServiceOptions="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--dataflowServiceOptions="parallel_replace_job_cancel_on_drain_timeout=true"

Si inhabilitas la cancelación automática y el trabajo anterior se atasca en el estado de desvío, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.

Python

Opción 1: Actualiza con el mismo nombre de trabajo

Nota: Si realizas la actualización con el mismo nombre de trabajo (opción 1), el nuevo trabajo debe comenzar al menos dos minutos después del inicio del trabajo anterior. De lo contrario, la actualización fallará con un error. Esta restricción evita las actualizaciones paralelas duplicadas causadas por reintentos de la biblioteca cliente o llamadas a procedimientos remotos desactualizadas.

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• Para realizar una actualización paralela con el mismo nombre, usa la marca --update y la opción update_strategy_parallel_job_update.
• Para realizar una actualización en un solo lugar sin quitar las opciones relacionadas con los trabajos paralelos, usa update_strategy_in_place_update en lugar de update_strategy_parallel_job_update.

Opción 2: Realiza la actualización con un nombre de trabajo diferente

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• Para especificar el trabajo anterior por ID en lugar de por nombre, usa --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
• Si especificas un nombre de trabajo nuevo y usas la marca --update, Dataflow buscará un trabajo existente con el nombre nuevo, lo que provocará un error.

Opcional: Configura el tiempo de espera para el agotamiento y la cancelación automática

Puedes agregar las siguientes opciones a cualquiera de las configuraciones para establecer un tiempo de espera de vaciado y cancelar automáticamente el trabajo anterior si se bloquea.

--dataflow_service_options="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=true"

Si inhabilitas la cancelación automática y el trabajo anterior se atasca en el estado de desvío, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.

Go

Opción 1: Actualiza con el mismo nombre de trabajo

Nota: Si realizas la actualización con el mismo nombre de trabajo (opción 1), el nuevo trabajo debe comenzar al menos dos minutos después del inicio del trabajo anterior. De lo contrario, la actualización fallará con un error. Esta restricción evita las actualizaciones paralelas duplicadas causadas por reintentos de la biblioteca cliente o llamadas a procedimientos remotos desactualizadas.

--update \
--dataflow_service_options="update_strategy_parallel_job_update" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• Para realizar una actualización paralela con el mismo nombre, usa la marca --update y la opción update_strategy_parallel_job_update.
• Para realizar una actualización en un solo lugar sin quitar las opciones relacionadas con los trabajos paralelos, usa update_strategy_in_place_update en lugar de update_strategy_parallel_job_update.

Opción 2: Realiza la actualización con un nombre de trabajo diferente

--dataflow_service_options="parallel_replace_job_name=OLD_JOB_NAME" \
--dataflow_service_options="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• Para especificar el trabajo anterior por ID en lugar de por nombre, usa --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
• Si especificas un nombre de trabajo nuevo y usas la marca --update, Dataflow buscará un trabajo existente con el nombre nuevo, lo que provocará un error.

Opcional: Configura el tiempo de espera para el agotamiento y la cancelación automática

Puedes agregar las siguientes opciones a cualquiera de las configuraciones para establecer un tiempo de espera de vaciado y cancelar automáticamente el trabajo anterior si se bloquea.

--dataflow_service_options="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=true"

Si inhabilitas la cancelación automática y el trabajo anterior se atasca en el estado de desvío, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.

gcloud

Opción 1: Actualiza con el mismo nombre de trabajo

Nota: Si realizas la actualización con el mismo nombre de trabajo (opción 1), el nuevo trabajo debe comenzar al menos dos minutos después del inicio del trabajo anterior. De lo contrario, la actualización fallará con un error. Esta restricción evita las actualizaciones paralelas duplicadas causadas por reintentos de la biblioteca cliente o llamadas a procedimientos remotos desactualizadas.

--update \
--additional-experiments="update_strategy_parallel_job_update" \
--additional-experiments="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• Para realizar una actualización paralela con el mismo nombre, usa la marca --update y la opción update_strategy_parallel_job_update.
• Para realizar una actualización en un solo lugar sin quitar las opciones relacionadas con los trabajos paralelos, usa update_strategy_in_place_update en lugar de update_strategy_parallel_job_update.

Opción 2: Realiza la actualización con un nombre de trabajo diferente

--additional-experiments="parallel_replace_job_name=OLD_JOB_NAME" \
--additional-experiments="parallel_replace_job_min_parallel_pipelines_duration=DURATION"

• Para especificar el trabajo anterior por ID en lugar de por nombre, usa --additional-experiments="parallel_replace_job_id=OLD_JOB_ID".
• Si especificas un nombre de trabajo nuevo y usas la marca --update, Dataflow buscará un trabajo existente con el nombre nuevo, lo que provocará un error.

Opcional: Configura el tiempo de espera para el agotamiento y la cancelación automática

Puedes agregar las siguientes opciones a cualquiera de las configuraciones para establecer un tiempo de espera de vaciado y cancelar automáticamente el trabajo anterior si se bloquea.

--additional-experiments="parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION" \
--additional-experiments="parallel_replace_job_cancel_on_drain_timeout=true"

Si inhabilitas la cancelación automática y el trabajo anterior se atasca en el estado de desvío, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.

Opcional: Upsert (crea o actualiza un trabajo)

Para habilitar el comportamiento de upsert (crear o actualizar el trabajo), haz lo siguiente:

--additional-experiments="create_or_update_job"

Terraform

additional_experiments = [
  "parallel_replace_job_min_parallel_pipelines_duration=DURATION",
  "parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION",
  "update_strategy_parallel_job_update",
  "create_or_update_job"
]

Config Connector

metadata:
  annotations:
    # Force to use only direct controller. Multiple controllers can cause a Dataflow job to enter into a continuous update loop.
    # https://docs.cloud.google.com/config-connector/docs/concepts/controller-types#underlying-controller-types
    alpha.cnrm.cloud.google.com/reconciler: direct
    # Optional but recommended: Dataflow batch jobs do not support the drain operation. But for Dataflow streaming jobs, prefer "drain" over "cancel" as an on-delete option.
    cnrm.cloud.google.com/on-delete: drain
    # Optional but recommended: Set deletion-policy to "abandon" to avoid accidental deletion, this will ignore the on-delete option.
    cnrm.cloud.google.com/deletion-policy: abandon
spec:
  ...
  additionalExperiments:
    - "parallel_replace_job_min_parallel_pipelines_duration=DURATION"
    - "parallel_replace_job_max_stop_duration=DRAIN_TIMEOUT_DURATION"
    - "update_strategy_parallel_job_update"
    - "create_or_update_job"

Reemplaza las siguientes variables:

• Si realizas la actualización con un nombre de trabajo diferente (opción 2), debes proporcionar parallel_replace_job_name o parallel_replace_job_id para identificar el trabajo que se reemplazará. No se admite la actualización con un nombre de trabajo diferente para Terraform ni Config Connector.
  • OLD_JOB_NAME: Es el nombre del trabajo que se reemplazará.
  • OLD_JOB_ID: Es el ID del trabajo que se reemplazará.

• DURATION: Es la cantidad mínima de tiempo que las dos canalizaciones se ejecutan en paralelo como un número entero o de punto flotante. Se recomienda una duración de 0s para que no haya superposición. Después de que transcurre este período, se envía una señal de desvío al trabajo anterior.

  La duración debe ser entre 0 segundos (0s) y 31 días (744h). Usa s, m y h para especificar segundos, minutos y horas. Por ejemplo, 10m es de 10 minutos.

• DRAIN_TIMEOUT_DURATION: Opcional Es la cantidad máxima de tiempo que el trabajo anterior debe agotar antes de que se active la cancelación automática. La duración debe tener el formato de una cadena que termine en s, m o h (por ejemplo, 30m, 1h).

• parallel_replace_job_cancel_on_drain_timeout: Opcional Indica si se debe cancelar el trabajo anterior si no termina de vaciarse antes de la duración máxima de detención. El valor predeterminado es true si se proporciona una duración de tiempo de espera de vaciado. Para inhabilitar la cancelación automática, establece esta opción en false. Si configuras esta opción como false y el trabajo anterior se atasca en el estado de desvío, tanto el trabajo anterior como el nuevo seguirán ejecutándose en paralelo.

Cuando inicias el trabajo nuevo, Dataflow espera a que se aprovisionen todos los trabajadores antes de comenzar a procesar los datos. Para supervisar el estado de la implementación, consulta los registros de trabajos de Dataflow.

Ejecuta canalizaciones paralelas de forma manual

Para situaciones más complejas o cuando necesites más control sobre el proceso de actualización, puedes ejecutar manualmente canalizaciones paralelas. Permite que la canalización existente se siga ejecutando hasta que su marca de agua supere la marca de tiempo del primer período completo que procesó la canalización actualizada. Luego, desvía o cancela la canalización existente.

Cómo controlar la salida duplicada

En el siguiente ejemplo, se describe un enfoque para controlar la salida duplicada. Las dos canalizaciones escriben el resultado en destinos diferentes, usan sistemas posteriores para consultar los resultados y anulan los resultados duplicados del período superpuesto. En este ejemplo, se usa una canalización que lee datos de entrada de Pub/Sub, realiza un procesamiento y escribe los resultados en BigQuery.

1. En el estado inicial, la canalización de transmisión existente (Canalización A) se ejecuta y lee mensajes de un tema de Pub/Sub (Tema) mediante una suscripción (Suscripción A). Los resultados se escriben en una tabla de BigQuery (Tabla A). Los resultados se procesan a través de una vista de BigQuery, que actúa como fachada para enmascarar los cambios subyacentes de la tabla. Este proceso es una aplicación de un método de diseño llamado patrón de fachada. En el siguiente diagrama, se muestra el estado inicial.

   

2. Crea una nueva suscripción (Suscripción B) para la canalización actualizada. Implementa la canalización actualizada (Canalización B), que lee del tema de Pub/Sub (Tema) con la Suscripción B y escribe en una tabla de BigQuery separada (Tabla B). En el diagrama siguiente, se ilustra este flujo:

   

   En este punto, la canalización A y la canalización B se ejecutan en paralelo y escriben resultados en tablas separadas. Debes registrar el tiempo t como la marca de tiempo del primer período completo que procesa la canalización B.

3. Cuando la marca de agua de la Canalización A supere el tiempo t, desvía la Canalización A. Cuando desvías la canalización, se cierran las ventanas abiertas y se completa el procesamiento de los datos en tránsito. Si la canalización contiene ventanas y las ventanas completas son importantes (suponiendo que no hay datos tardíos), antes de desviar la Canalización A, permite que ambas canalizaciones se ejecuten hasta que tengas ventanas superpuestas completas. Detén el trabajo de transmisión de la Canalización A después de procesar todos los datos en tránsito y escribirlos en la tabla A. En el siguiente diagrama, se muestra esta etapa.

   

4. En este punto, solo se está ejecutando la Canalización B. Puedes realizar una consulta desde una vista de BigQuery (vista de fachada), que actúa como una fachada para la tabla A y la tabla B. En las filas que tienen la misma marca de tiempo en ambas tablas, configura la vista para que muestre las filas de la Tabla B o, si las filas no existen en la Tabla B, recurra a la Tabla A. En el siguiente diagrama, se muestra la lectura de la vista (vista de fachada) de la tabla A y la tabla B.

   

   En este punto, puedes borrar la suscripción A.

Cuando se detectan problemas con una implementación de canalización nueva, tener canalizaciones paralelas puede simplificar la reversión. En este ejemplo, es posible que quieras mantener la Canalización A en ejecución mientras supervisas la Canalización B para lograr una operación correcta. Si se produce algún problema con la Canalización B, puedes recurrir a la Canalización A.

Administra las mutaciones del esquema

Los sistemas de control de datos suelen necesitar adaptarse a las mutaciones del esquema a lo largo del tiempo, a veces debido a cambios en los requisitos del negocio y otras veces por razones técnicas. Por lo general, la aplicación de actualizaciones del esquema requiere una planificación y una ejecución cuidadas para evitar la interrupción de los sistemas de información de la empresa.

Considera una canalización que lee mensajes que contienen cargas útiles de JSON desde un tema de Pub/Sub. La canalización convierte cada mensaje en una instancia de TableRow y, luego, escribe las filas en una tabla de BigQuery. El esquema de la tabla de salida es similar a los mensajes que procesa la canalización. En el siguiente diagrama, el esquema se conoce como Esquema A.

Con el tiempo, el esquema del mensaje puede mutar de maneras no triviales. Por ejemplo, se agregan, quitan o reemplazan campos. El Esquema A evoluciona a un esquema nuevo. En la discusión siguiente, se hace referencia al esquema nuevo como Esquema B. En este caso, la canalización A debe actualizarse, y el esquema de la tabla de salida debe ser compatible con el Esquema B.

Para la tabla de salida, puedes realizar algunas mutaciones del esquema sin el centro de noticias. Por ejemplo, puedes agregar campos nuevos o disminuir la rigurosidad de los modos de columna, como cambiar REQUIRED a NULLABLE, sin tiempo de inactividad. Por lo general, estas mutaciones no afectan las consultas existentes. Sin embargo, las mutaciones del esquema que modifican o quitan campos del esquema existentes interrumpen las consultas o generan otras interrupciones. El siguiente enfoque se adapta a los cambios sin requerir tiempo de inactividad.

Separa los datos que escribe la canalización en una tabla principal y en una o en más tablas de etapa de pruebas. En la tabla principal, se almacenan datos históricos que escribió la canalización. Las tablas de etapa de pruebas almacenan el resultado de la canalización más reciente. Puedes definir una vista de fachada de BigQuery en las tablas principal y de etapa de pruebas, lo que permite que los consumidores consulten los datos históricos y los actualizados.

En el siguiente diagrama, se revisa el flujo de canalización anterior para incluir una tabla de etapa de pruebas (Tabla de etapa de pruebas A), una tabla principal y una vista de fachada.

En el flujo revisado, la canalización A procesa mensajes que usan el Esquema A y escribe el resultado en la Tabla de etapa de pruebas A, que tiene un esquema compatible. La tabla principal contiene datos históricos que escribieron las versiones anteriores de la canalización, además de resultados que se combinan periódicamente de la tabla de etapa de pruebas. Los consumidores pueden consultar datos actualizados, incluidos los datos históricos y en tiempo real, con la vista de fachada.

Cuando el esquema de los mensajes cambia del Esquema A al Esquema B, puedes actualizar el código de la canalización para que sea compatible con los mensajes que usan el Esquema B. La canalización existente debe actualizarse con la implementación nueva. Si ejecutas canalizaciones paralelas, puedes garantizar que el procesamiento de datos de transmisión continúe sin interrupciones. Finalizar y reemplazar canalizaciones da como resultado una interrupción del procesamiento, ya que no se ejecuta ninguna canalización por un tiempo.

La canalización actualizada escribe en una tabla de etapa de pruebas adicional (tabla de etapa de pruebas B) que usa el esquema B. Puedes usar un flujo de trabajo organizado para crear la tabla de etapa de pruebas nueva antes de actualizar la canalización. Actualiza la vista de fachada para incluir los resultados de la nueva tabla de etapa de pruebas, que puede usar un paso de flujo de trabajo relacionado.

En el siguiente diagrama, aparece el flujo actualizado que muestra la tabla de etapa de pruebas B con el esquema B, y cómo se actualizó la vista de fachada para incluir el contenido de la tabla principal y de ambas tablas de etapa de pruebas.

Como un proceso separado de la actualización de la canalización, puedes combinar las tablas de etapa de pruebas en la tabla principal, ya sea de forma periódica o según sea necesario. En el siguiente diagrama, se muestra cómo la Tabla de etapa de pruebas A se fusiona en la tabla principal.

¿Qué sigue?

• Encuentra pasos detallados para actualizar una canalización existente.