Mettre à niveau un pipeline de traitement par flux

Cette page fournit des conseils et des recommandations pour mettre à niveau vos pipelines de traitement en flux continu. Par exemple, vous devrez peut-être passer à une version plus récente du SDK Apache Beam ou mettre à jour le code de votre pipeline. Différentes options sont proposées pour répondre à différents scénarios.

Contrairement aux pipelines de traitement par lot qui s'arrêtent une fois le job terminé, les pipelines de traitement en flux continu doivent souvent s'exécuter en continu pour assurer un traitement sans interruption. Par conséquent, lorsque vous mettez à niveau des pipelines de traitement en flux continu, vous devez prendre en compte les éléments suivants :

• Il peut s'avérer nécessaire de minimiser ou d'éviter les interruptions du pipeline. Dans certains cas, il est possible de tolérer une interruption temporaire du traitement pendant le déploiement d'une nouvelle version du pipeline. Dans d'autres cas, votre application ne pourra tolérer aucune interruption.
• Les processus de mise à jour de pipeline doivent gérer les modifications de schéma de manière à minimiser les interruptions du traitement des messages et des autres systèmes associés. Par exemple, si le schéma des messages dans un pipeline de traitement des événements change, il peut être nécessaire de reporter également ces modifications de schéma dans les récepteurs de données en aval.

Vous pouvez utiliser l'une des méthodes suivantes pour mettre à jour les pipelines de traitement en flux continu, en fonction de vos exigences de pipeline et de mise à jour :

• Effectuer des mises à jour en cours
• Lancer un job de remplacement
• Exécuter des pipelines en parallèle

Pour plus d'informations sur les problèmes que vous pouvez rencontrer lors d'une mise à jour et sur la façon de les éviter, consultez les sections Valider un job de remplacement et Vérifier la compatibilité des jobs.

Bonnes pratiques

• Mettez à niveau la version du SDK Apache Beam séparément de toute modification du code du pipeline.
• Testez votre pipeline après chaque modification avant d'effectuer des mises à jour supplémentaires.
• Mettez régulièrement à niveau la version du SDK Apache Beam utilisée par votre pipeline.
• Utilisez des méthodes automatisées dans la mesure du possible, comme les mises à jour en cours ou les mises à jour automatisées de pipelines parallèles.
• Utilisez Managed I/O lorsque cela est possible pour bénéficier des mises à niveau automatiques des versions de connecteurs.

Effectuer des mises à jour en cours

Vous pouvez mettre à jour certains pipelines de traitement en flux continu en cours sans arrêter le job. Ce scénario est appelé une mise à jour de job en cours. Les mises à jour de job en cours ne sont disponibles que dans des circonstances limitées :

• Le job doit utiliser Streaming Engine.
• Le job doit être à l'état "en cours d'exécution".
• Vous modifiez uniquement le nombre de nœuds de calcul utilisés par le job.

Pour plus d'informations, consultez la section Définir la plage d'autoscaling de la page "Autoscaling horizontal".

Pour obtenir des instructions sur la façon d'effectuer une mise à jour de job en cours, consultez la page Mettre à jour un pipeline existant.

Création ou mise à jour (upsert) automatiques pour les modèles

Lorsque vous lancez des pipelines à l'aide d'un modèle (modèles classiques, modèles Flex, Terraform ou Config Connector), vous pouvez utiliser l'expérience create_or_update_job pour utiliser la fonctionnalité de création ou de mise à jour (upsert).

Lorsque vous spécifiez create_or_update_job dans le paramètre additional_experiments ou l'indicateur additional-experiments :

• Si une tâche en cours d'exécution ou de vidange portant le nom spécifié existe déjà, le service de modèles lance automatiquement la nouvelle tâche en tant que mise à jour de la tâche existante.
• Si aucune tâche active portant ce nom n'existe, le service de modèles lance la nouvelle tâche comme une nouvelle création de tâche.

Ce test élimine la nécessité de déterminer par programmation s'il faut utiliser l'action d'API de création ou de mise à jour lors du lancement d'un modèle.

Pour obtenir des exemples de code Terraform et Config Connector qui utilisent ce test, consultez les sections suivantes :

• Envoyer une demande de mise à jour automatisée de type "stop-and-replace"
• Envoyer une demande de mise à jour automatisée du pipeline parallèle

Lancer un job de remplacement

Si le job mis à jour est compatible avec le job existant, vous pouvez mettre à jour votre pipeline à l'aide de l'option update. Lorsque vous remplacez un job existant, un nouveau job exécute le code de votre pipeline mis à jour. Le service Dataflow conserve le nom du job, mais exécute le job de remplacement avec un ID de job mis à jour. Ce processus peut entraîner des temps d'arrêt pendant l'arrêt de la tâche existante, l'exécution de la vérification de compatibilité et le démarrage de la nouvelle tâche. Pour en savoir plus, consultez la section Effets du remplacement d'un job.

Dataflow effectue une vérification de compatibilité pour s'assurer que le code du pipeline mis à jour peut être déployé en toute sécurité sur le pipeline en cours d'exécution. Certaines modifications de code entraînent l'échec de la vérification de compatibilité, par exemple lorsque des entrées secondaires sont ajoutées ou supprimées d'une étape existante. Lorsque la vérification de compatibilité échoue, vous ne pouvez pas effectuer de mise à jour de job sur place.

Pour obtenir des instructions expliquant comment lancer un job de remplacement, consultez la section Lancer un job de remplacement.

Si la mise à jour du pipeline n'est pas compatible avec le job actuel, vous devez arrêter et remplacer le pipeline. Si votre pipeline ne peut pas tolérer de temps d'arrêt, exécutez des pipelines en parallèle.

Arrêter et remplacer manuellement

Pour effectuer un arrêt et un remplacement manuels, annulez ou drainez le pipeline, puis remplacez-le par le pipeline mis à jour. L'annulation d'un pipeline oblige Dataflow à interrompre immédiatement le traitement et à arrêter les ressources le plus rapidement possible, ce qui peut entraîner une perte des données en cours de traitement, appelées données en cours de transfert. Dans la plupart des cas, le drainage est l'action recommandée pour éviter ce problème. Vous pouvez également utiliser des instantanés Dataflow pour enregistrer l'état d'un pipeline de traitement en flux continu, ce qui vous permet de démarrer une nouvelle version de votre tâche Dataflow sans perdre l'état. Pour en savoir plus, consultez Utiliser des instantanés Dataflow.

Le drainage d'un pipeline ferme immédiatement toutes les fenêtres en cours et actionne tous les déclencheurs. Bien que les données en cours de transfert ne soient pas perdues, le drainage peut entraîner des données incomplètes pour les fenêtres. Dans ce cas, les fenêtres en cours de traitement émettent des résultats partiels ou incomplets. Pour en savoir plus, consultez Effets liés au drainage d'un job. Une fois le job existant terminé, lancez un nouveau job de traitement en flux continu contenant le code de votre pipeline mis à jour, ce qui vous permet de reprendre le traitement.

Avec cette méthode, vous subissez un temps d'arrêt entre le moment où le job de traitement en flux continu existant s'arrête et le moment où le pipeline de remplacement est prêt à prendre le relais. Cependant, le fait d'annuler ou de drainer un pipeline existant puis de lancer un nouveau job avec le pipeline mis à jour est moins compliqué que d'exécuter des pipelines en parallèle.

Pour en savoir plus, consultez Drainer une tâche Dataflow. Après avoir drainé la tâche actuelle, démarrez-en une nouvelle portant le même nom.

Arrêt et remplacement automatisés

Dataflow fournit une assistance API pour lancer une mise à jour automatisée d'arrêt et de remplacement. Ce workflow déclaratif élimine les étapes manuelles. Vous déclarez le job à remplacer, et le nouveau job se lance et coordonne la transition automatiquement.

Lorsque vous utilisez ce workflow, de nouvelles ressources de job sont provisionnées alors que l'ancien job est toujours en cours d'exécution. L'ancien job reçoit alors automatiquement un signal de vidange. Une fois l'ancienne tâche drainée ou le délai d'expiration spécifié par l'utilisateur atteint, la nouvelle tâche commence immédiatement à traiter les données. Utilisez ce workflow pour les pipelines qui ne peuvent pas tolérer les données en double ni les agrégations partielles, mais qui peuvent accepter une brève pause de traitement pendant que l'ancien job se vide.

Envoyer une demande de mise à jour automatisée de type "stop-and-replace"

Pour utiliser ce workflow :

• Vous devez définir l'option parallel_replace_job_max_stop_duration.
• Vous ne devez pas définir l'option parallel_replace_job_min_parallel_pipelines_duration. Si vous définissez une durée parallèle, le workflow de mise à jour automatique du pipeline parallèle est déclenché à la place.

Lancez une demande de mise à jour automatisée de type "stop-and-replace" (arrêter et remplacer) à l'aide des options de service suivantes :

Java

Option 1 : Mettre à jour en utilisant le même nom de job

Remarque : Si vous effectuez une mise à jour en utilisant le même nom de job (option 1), le nouveau job doit démarrer au moins deux minutes après le début du job précédent. Sinon, la mise à jour échoue et renvoie une erreur. Cette restriction empêche les mises à jour parallèles en double causées par les nouvelles tentatives de la bibliothèque cliente ou les appels de procédure à distance obsolètes.

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

• Pour effectuer une mise à jour automatisée avec arrêt et remplacement en utilisant le même nom, utilisez l'option --update et l'option update_strategy_parallel_job_update.
• Pour effectuer une mise à jour sur place, utilisez plutôt update_strategy_in_place_update.

Option 2 : Mettre à jour à l'aide d'un autre nom de job

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

• Pour spécifier l'ancien job par ID au lieu de son nom, utilisez --dataflowServiceOptions="parallel_replace_job_id=OLD_JOB_ID".
• Si vous spécifiez un nouveau nom de job et utilisez l'indicateur --update, Dataflow recherche un job existant portant le nouveau nom, ce qui provoque une erreur.

Facultatif : Désactiver la résiliation automatique

L'annulation automatique est activée par défaut lorsque vous spécifiez l'option parallel_replace_job_max_stop_duration. Pour désactiver l'annulation automatique, définissez l'option parallel_replace_job_cancel_on_drain_timeout sur false.

--dataflowServiceOptions="parallel_replace_job_cancel_on_drain_timeout=false"

Si vous désactivez l'annulation automatique et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.

Python

Option 1 : Mettre à jour en utilisant le même nom de job

Remarque : Si vous effectuez une mise à jour en utilisant le même nom de job (option 1), le nouveau job doit démarrer au moins deux minutes après le début du job précédent. Sinon, la mise à jour échoue et renvoie une erreur. Cette restriction empêche les mises à jour parallèles en double causées par les nouvelles tentatives de la bibliothèque cliente ou les appels de procédure à distance obsolètes.

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

• Pour effectuer une mise à jour automatisée avec arrêt et remplacement en utilisant le même nom, utilisez l'option --update et l'option update_strategy_parallel_job_update.
• Pour effectuer une mise à jour sur place, utilisez plutôt update_strategy_in_place_update.

Option 2 : Mettre à jour à l'aide d'un autre nom de job

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

• Pour spécifier l'ancien job par ID au lieu de son nom, utilisez --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
• Si vous spécifiez un nouveau nom de job et utilisez l'indicateur --update, Dataflow recherche un job existant portant le nouveau nom, ce qui provoque une erreur.

Facultatif : Désactiver la résiliation automatique

L'annulation automatique est activée par défaut lorsque vous spécifiez l'option parallel_replace_job_max_stop_duration. Pour désactiver l'annulation automatique, définissez l'option parallel_replace_job_cancel_on_drain_timeout sur false.

--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=false"

Si vous désactivez l'annulation automatique et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.

Go

Option 1 : Mettre à jour en utilisant le même nom de job

Remarque : Si vous effectuez une mise à jour en utilisant le même nom de job (option 1), le nouveau job doit démarrer au moins deux minutes après le début du job précédent. Sinon, la mise à jour échoue et renvoie une erreur. Cette restriction empêche les mises à jour parallèles en double causées par les nouvelles tentatives de la bibliothèque cliente ou les appels de procédure à distance obsolètes.

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

• Pour effectuer une mise à jour automatisée avec arrêt et remplacement en utilisant le même nom, utilisez l'option --update et l'option update_strategy_parallel_job_update.
• Pour effectuer une mise à jour sur place, utilisez plutôt update_strategy_in_place_update.

Option 2 : Mettre à jour à l'aide d'un autre nom de job

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

• Pour spécifier l'ancien job par ID au lieu de son nom, utilisez --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
• Si vous spécifiez un nouveau nom de job et utilisez l'indicateur --update, Dataflow recherche un job existant portant le nouveau nom, ce qui provoque une erreur.

Facultatif : Désactiver la résiliation automatique

L'annulation automatique est activée par défaut lorsque vous spécifiez l'option parallel_replace_job_max_stop_duration. Pour désactiver l'annulation automatique, définissez l'option parallel_replace_job_cancel_on_drain_timeout sur false.

--dataflow_service_options="parallel_replace_job_cancel_on_drain_timeout=false"

Si vous désactivez l'annulation automatique et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.

gcloud

Option 1 : Mettre à jour en utilisant le même nom de job

Remarque : Si vous effectuez une mise à jour en utilisant le même nom de job (option 1), le nouveau job doit démarrer au moins deux minutes après le début du job précédent. Sinon, la mise à jour échoue et renvoie une erreur. Cette restriction empêche les mises à jour parallèles en double causées par les nouvelles tentatives de la bibliothèque cliente ou les appels de procédure à distance obsolètes.

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

• Pour effectuer une mise à jour automatisée avec arrêt et remplacement en utilisant le même nom, utilisez l'option --update et l'option update_strategy_parallel_job_update.
• Pour effectuer une mise à jour sur place, utilisez plutôt update_strategy_in_place_update.

Option 2 : Mettre à jour à l'aide d'un autre nom de job

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

• Pour spécifier l'ancien job par ID au lieu de son nom, utilisez --additional-experiments="parallel_replace_job_id=OLD_JOB_ID".
• Si vous spécifiez un nouveau nom de job et utilisez l'indicateur --update, Dataflow recherche un job existant portant le nouveau nom, ce qui provoque une erreur.

Facultatif : Désactiver la résiliation automatique

L'annulation automatique est activée par défaut lorsque vous spécifiez l'option parallel_replace_job_max_stop_duration. Pour désactiver l'annulation automatique, définissez l'option parallel_replace_job_cancel_on_drain_timeout sur false.

--additional-experiments="parallel_replace_job_cancel_on_drain_timeout=false"

Si vous désactivez l'annulation automatique et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.

Facultatif : Upsert (créer ou mettre à jour un job)

Pour activer le comportement d'upsert (créer ou mettre à jour une tâche) :

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

Remplacez les variables suivantes :

• Vous devez fournir parallel_replace_job_name ou parallel_replace_job_id pour identifier la tâche à remplacer :
  • OLD_JOB_NAME : nom du job à remplacer.
  • OLD_JOB_ID : ID du job à remplacer.
• Vous devez fournir la valeur parallel_replace_job_max_stop_duration pour activer l'arrêt et le remplacement automatiques :
  • DURATION : durée maximale pendant laquelle le nouveau job attend que l'ancien job ait fini de vider les données. La durée doit être mise en forme sous forme de chaîne se terminant par s, m ou h (par exemple, 30m ou 1h).
• Ne définissez pas l'option parallel_replace_job_min_parallel_pipelines_duration lorsque vous utilisez ce workflow. Si vous définissez cette option, le workflow de mise à jour parallèle automatisée du pipeline est déclenché à la place.
• Facultatif : Configurez l'option parallel_replace_job_cancel_on_drain_timeout. Étant donné que l'annulation automatique est activée (par défaut sur true) par défaut lorsque l'option parallel_replace_job_max_stop_duration est définie, vous n'avez pas besoin de configurer explicitement cette option pour l'activer.
  • Pour conserver le comportement par défaut, omettez cette option ou définissez-la sur true.
  • Pour désactiver la résiliation automatique, définissez cette option sur false. Si vous définissez cette option sur false et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.
• Facultatif : Définissez la configuration parallel_replace_job_preallocate_compute_resources :
  • Indique si les nœuds de calcul sont provisionnés à l'avance pour le nouveau job pendant que l'ancien job est vidé. Valeurs : true (par défaut) ou false. Pour Terraform et Config Connector, il est recommandé de définir cette option sur true afin d'éviter les délais d'expiration du provisionnement des ressources. Lorsque parallel_replace_job_preallocate_compute_resources est défini sur false, le nouveau job reste à l'état "En attente" jusqu'à ce que l'ancien job soit vidé.

Relancer le traitement des messages avec des fonctionnalités d'instantanés et de recherche Pub/Sub

Dans certains cas, après avoir remplacé ou annulé un pipeline drainé, vous devrez peut-être relancer le traitement des messages Pub/Sub déjà distribués. Par exemple, vous devrez peut-être utiliser une logique métier mise à jour pour relancer le traitement des données. La recherche Pub/Sub est une fonctionnalité qui vous permet de relire les messages d'un instantané Pub/Sub. Vous pouvez utiliser la recherche Pub/Sub avec Dataflow pour relancer le traitement des messages à partir du moment où l'instantané d'abonnement est créé.

Pendant le développement et les tests, vous pouvez également utiliser la fonctionnalité de recherche Pub/Sub pour relire plusieurs fois les messages connus afin de vérifier le résultat de votre pipeline. Lorsque vous utilisez la recherche Pub/Sub, ne recherchez pas d'instantané d'abonnement lorsque l'abonnement est utilisé par un pipeline. Si vous le faites, la recherche pourrait invalider la logique du filigrane de Dataflow et affecter le traitement "exactement une fois" des messages Pub/Sub.

Voici un workflow recommandé de gcloud CLI pour utiliser les fonctionnalités d'instantané et de recherche Pub/Sub avec des pipelines Dataflow dans une fenêtre de terminal :

1. Pour créer un instantané de l'abonnement, exécutez la commande gcloud pubsub snapshots create :

   gcloud pubsub snapshots create SNAPSHOT_NAME --subscription=PIPELINE_SUBSCRIPTION_NAME
   

2. Pour drainer ou annuler le pipeline, utilisez la commande gcloud dataflow jobs drain ou la commande gcloud dataflow jobs cancel :

   gcloud dataflow jobs drain JOB_ID
   

   ou

   gcloud dataflow jobs cancel JOB_ID
   

3. Pour rechercher l'instantané, exécutez la commande gcloud pubsub subscriptions seek :

   gcloud pubsub subscriptions seek SNAPSHOT_NAME
   

4. Déployez un nouveau pipeline qui utilise l'abonnement.

Exécuter des pipelines en parallèle

Pour éviter toute interruption de votre pipeline de traitement en flux continu lors d'une mise à jour, vous pouvez exécuter des pipelines en parallèle. Cette approche vous permet de lancer un nouveau job de traitement en flux continu avec le code de votre pipeline mis à jour et de l'exécuter en parallèle avec le job existant. Vous pouvez utiliser le workflow de déploiement de mise à jour de pipeline parallèle automatisé de Dataflow ou effectuer les étapes manuellement.

Présentation des pipelines parallèles

Lorsque vous créez le pipeline, utilisez la même stratégie de fenêtrage que celle utilisée pour le pipeline existant. Pour le workflow manuel, laissez le pipeline existant s'exécuter jusqu'à ce que son filigrane dépasse l'horodatage de la première fenêtre complète traitée par le pipeline mis à jour. Ensuite, drainez ou annulez le pipeline existant. Si vous utilisez le workflow automatisé, ce travail est effectué pour vous. Le pipeline mis à jour continue de s'exécuter pour prendre le relais et s'occuper du traitement de manière autonome.

Le diagramme suivant illustre ce processus.

Dans le diagramme, le Pipeline B correspond à la tâche mise à jour qui relaye le Pipeline A. La valeur t correspond à l'horodatage de la fenêtre complète la plus ancienne traitée par le pipeline B. La valeur w correspond au filigrane du Pipeline A. Pour des raisons de simplicité, un filigrane est considéré comme étant parfait sans données tardives. Le traitement et la durée d'exécution sont représentés sur l'axe horizontal. Les deux pipelines utilisent des fenêtres à durée fixe (fenêtres bascules) de cinq minutes. Les résultats sont déclenchés une fois que le filigrane dépasse la fin de chaque fenêtre.

Comme des résultats simultanés se produisent pendant la période où les deux pipelines se chevauchent, configurez les deux pipelines pour écrire les résultats dans différentes destinations. Les systèmes en aval peuvent ainsi utiliser un extrait des deux récepteurs de destination, par exemple une vue de base de données, pour interroger les résultats combinés. Ces systèmes peuvent également utiliser l'extrait pour dédupliquer les résultats de la période de chevauchement. Pour en savoir plus, consultez Gérer les sorties en double.

Limites

L'utilisation de mises à jour parallèles automatiques ou manuelles des pipelines présente les limites suivantes :

• Mises à jour automatiques uniquement : le nouveau job parallèle doit être un job Streaming Engine.
• Les jobs simultanés portant le même nom ne sont pas autorisés. Toutefois, lorsque vous effectuez un arrêt et remplacement automatisés ou une mise à jour parallèle du pipeline en utilisant le même nom de job, vous pouvez réutiliser le nom du job. Dans ce cas, le nouveau job doit démarrer au moins deux minutes après le début du job précédent. Cette restriction empêche les mises à jour parallèles multiples provenant de nouvelles tentatives répétées de la bibliothèque cliente ou d'appels de procédure à distance obsolètes.
• L'exécution de deux pipelines en parallèle sur la même entrée peut entraîner des données en double, des agrégations partielles et des problèmes d'ordre potentiels lorsque les données sont insérées dans le récepteur. Le système en aval doit être conçu pour anticiper et gérer ces résultats.
• Lors de la lecture à partir d'une source Pub/Sub, il n'est pas recommandé d'utiliser le même abonnement pour plusieurs pipelines, car cela peut entraîner des problèmes d'exactitude. Toutefois, dans certains cas d'utilisation, comme les pipelines d'extraction, de transformation et de chargement (ETL), utiliser le même abonnement dans deux pipelines peut réduire la duplication. Des problèmes d'autoscaling sont susceptibles de se produire chaque fois que vous fournissez une valeur non nulle pour la durée de chevauchement. Cela peut être atténué en utilisant la fonctionnalité de mise à jour des jobs en cours. Pour en savoir plus, consultez Affiner l'autoscaling pour vos pipelines de traitement en flux continu Pub/Sub.
• Pour Apache Kafka, vous pouvez réduire le nombre de doublons en activant la validation du décalage dans Kafka. Pour activer la validation du décalage dans Kafka, consultez Validation dans Kafka.

Mises à jour automatiques et parallèles des pipelines

Dataflow fournit une assistance API pour lancer un job de remplacement parallèle. Cette API de style déclaratif abstrait le travail manuel d'exécution des étapes procédurales. Vous déclarez le job que vous souhaitez mettre à jour, puis un nouveau job s'exécute en parallèle avec l'ancien. Une fois que le nouveau job s'exécute pendant la durée que vous spécifiez, l'ancien job est vidé. Cette fonctionnalité élimine les pauses de traitement lors des mises à jour. Cela réduit également l'effort opérationnel nécessaire pour mettre à jour les pipelines incompatibles.

Cette méthode de mise à jour est idéale pour les pipelines qui peuvent tolérer certains doublons ou agrégations partielles, et qui n'ont pas besoin d'un ordre strict lors de l'insertion de données. Il est bien adapté aux pipelines ETL, ainsi qu'aux pipelines utilisant le mode de traitement en flux continu de type "au moins une fois" et la transformation Redistribute avec l'option "Autoriser les doublons" définie sur true.

Options de service de pipeline parallèle automatisé

Remarque : Vous devez définir l'option parallel_replace_job_min_parallel_pipelines_duration, quelle que soit la stratégie de mise à jour que vous choisissez (option 1 : mise à jour avec le même nom de job ou option 2 : mise à jour avec un nom de job différent).

Utilisez les options de service suivantes pour les mises à jour automatisées et parallèles des pipelines :

Option de service	Facultatif ou obligatoire	Description	Dépendances ou exclusions
update_strategy_parallel_job_update	Obligatoire (option 1 : mise à jour avec le même nom de job)	Commande permettant d'effectuer une mise à jour parallèle, qui exécute les deux pipelines simultanément pour minimiser les temps d'arrêt, lors de la mise à jour sous le même nom de job.	Doit être défini avec l'option --update et parallel_replace_job_min_parallel_pipelines_duration.
update_strategy_in_place_update	Optional	Alternative à la mise à jour parallèle. Effectue une mise à jour standard sur place du job.	Doit être défini avec l'option --update. Incompatible avec update_strategy_parallel_job_update. Lorsque cette option est définie, les autres options liées aux tâches parallèles sont ignorées.
parallel_replace_job_min_parallel_pipelines_duration	Obligatoire	Spécifie la durée minimale pendant laquelle les deux pipelines s'exécutent simultanément. Une fois cette durée écoulée, un signal de vidange est envoyé à l'ancien job. Les valeurs acceptables sont comprises entre 0s (recommandé pour une absence de chevauchement) et 744h (31 jours).	Doit être associé à un moyen de cibler l'ancien emploi. Parmi les suivants : Option 1 : Utiliser le même nom de job : update_strategy_parallel_job_update, ou Option 2 : utiliser un autre nom de job : parallel_replace_job_name (ou parallel_replace_job_id pour identifier le job)
parallel_replace_job_name ou parallel_replace_job_id (au choix)	Obligatoire (option 2 : mettre à jour à l'aide d'un autre nom de job)	Identifie l'ancien job (par son nom ou son ID) à remplacer lors d'une mise à jour avec un nom différent.	Nécessite que parallel_replace_job_min_parallel_pipelines_duration soit défini. N'utilisez pas l'option --update ni parallel_replace_job_id avec cette option.
parallel_replace_job_max_stop_duration	Optional	Durée maximale pendant laquelle l'ancien job est autorisé à se vider avant le déclenchement de l'annulation automatique. Par exemple, 30m ou 1h.	Nécessite de définir un workflow de mise à jour parallèle (option 1 ou option 2).
parallel_replace_job_cancel_on_drain_timeout	Optional La valeur par défaut est true si une durée d'arrêt maximale est définie.	Option booléenne indiquant s'il faut annuler l'ancienne tâche si sa durée de vidange dépasse parallel_replace_job_max_stop_duration.	À utiliser conjointement avec parallel_replace_job_max_stop_duration. Définissez la valeur sur false pour désactiver la résiliation automatique. Si vous désactivez l'annulation automatique et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.

Envoyer une demande de mise à jour automatisée et parallèle du pipeline

Pour utiliser le workflow automatisé, lancez un nouveau job en flux continu. Vous pouvez mettre à jour un job en utilisant le même nom de job ou un nom de job différent.

Java

Option 1 : Mettre à jour en utilisant le même nom de job

Remarque : Si vous effectuez une mise à jour en utilisant le même nom de job (option 1), le nouveau job doit démarrer au moins deux minutes après le début du job précédent. Sinon, la mise à jour échoue et renvoie une erreur. Cette restriction empêche les mises à jour parallèles en double causées par les nouvelles tentatives de la bibliothèque cliente ou les appels de procédure à distance obsolètes.

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

• Pour effectuer une mise à jour parallèle en utilisant le même nom, utilisez l'indicateur --update et l'option update_strategy_parallel_job_update.
• Pour effectuer une mise à jour sur place sans supprimer les options liées aux jobs parallèles, utilisez update_strategy_in_place_update au lieu de update_strategy_parallel_job_update.

Option 2 : Mettre à jour à l'aide d'un autre nom de job

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

• Pour spécifier l'ancien job par ID au lieu de son nom, utilisez --dataflowServiceOptions="parallel_replace_job_id=OLD_JOB_ID".
• Si vous spécifiez un nouveau nom de job et utilisez l'indicateur --update, Dataflow recherche un job existant portant le nouveau nom, ce qui provoque une erreur.

Facultatif : configurez le délai avant expiration du drainage et l'annulation automatique

Vous pouvez ajouter les options suivantes à l'une ou l'autre des configurations pour définir un délai avant vidange et annuler automatiquement l'ancienne tâche si elle est bloquée.

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

Si vous désactivez l'annulation automatique et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.

Python

Option 1 : Mettre à jour en utilisant le même nom de job

Remarque : Si vous effectuez une mise à jour en utilisant le même nom de job (option 1), le nouveau job doit démarrer au moins deux minutes après le début du job précédent. Sinon, la mise à jour échoue et renvoie une erreur. Cette restriction empêche les mises à jour parallèles en double causées par les nouvelles tentatives de la bibliothèque cliente ou les appels de procédure à distance obsolètes.

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

• Pour effectuer une mise à jour parallèle en utilisant le même nom, utilisez l'indicateur --update et l'option update_strategy_parallel_job_update.
• Pour effectuer une mise à jour sur place sans supprimer les options liées aux jobs parallèles, utilisez update_strategy_in_place_update au lieu de update_strategy_parallel_job_update.

Option 2 : Mettre à jour à l'aide d'un autre nom de job

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

• Pour spécifier l'ancien job par ID au lieu de son nom, utilisez --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
• Si vous spécifiez un nouveau nom de job et utilisez l'indicateur --update, Dataflow recherche un job existant portant le nouveau nom, ce qui provoque une erreur.

Facultatif : configurez le délai avant expiration du drainage et l'annulation automatique

Vous pouvez ajouter les options suivantes à l'une ou l'autre des configurations pour définir un délai avant vidange et annuler automatiquement l'ancienne tâche si elle est bloquée.

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

Si vous désactivez l'annulation automatique et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.

Go

Option 1 : Mettre à jour en utilisant le même nom de job

Remarque : Si vous effectuez une mise à jour en utilisant le même nom de job (option 1), le nouveau job doit démarrer au moins deux minutes après le début du job précédent. Sinon, la mise à jour échoue et renvoie une erreur. Cette restriction empêche les mises à jour parallèles en double causées par les nouvelles tentatives de la bibliothèque cliente ou les appels de procédure à distance obsolètes.

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

• Pour effectuer une mise à jour parallèle en utilisant le même nom, utilisez l'indicateur --update et l'option update_strategy_parallel_job_update.
• Pour effectuer une mise à jour sur place sans supprimer les options liées aux jobs parallèles, utilisez update_strategy_in_place_update au lieu de update_strategy_parallel_job_update.

Option 2 : Mettre à jour à l'aide d'un autre nom de job

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

• Pour spécifier l'ancien job par ID au lieu de son nom, utilisez --dataflow_service_options="parallel_replace_job_id=OLD_JOB_ID".
• Si vous spécifiez un nouveau nom de job et utilisez l'indicateur --update, Dataflow recherche un job existant portant le nouveau nom, ce qui provoque une erreur.

Facultatif : configurez le délai avant expiration du drainage et l'annulation automatique

Vous pouvez ajouter les options suivantes à l'une ou l'autre des configurations pour définir un délai avant vidange et annuler automatiquement l'ancienne tâche si elle est bloquée.

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

Si vous désactivez l'annulation automatique et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.

gcloud

Option 1 : Mettre à jour en utilisant le même nom de job

Remarque : Si vous effectuez une mise à jour en utilisant le même nom de job (option 1), le nouveau job doit démarrer au moins deux minutes après le début du job précédent. Sinon, la mise à jour échoue et renvoie une erreur. Cette restriction empêche les mises à jour parallèles en double causées par les nouvelles tentatives de la bibliothèque cliente ou les appels de procédure à distance obsolètes.

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

• Pour effectuer une mise à jour parallèle en utilisant le même nom, utilisez l'indicateur --update et l'option update_strategy_parallel_job_update.
• Pour effectuer une mise à jour sur place sans supprimer les options liées aux jobs parallèles, utilisez update_strategy_in_place_update au lieu de update_strategy_parallel_job_update.

Option 2 : Mettre à jour à l'aide d'un autre nom de job

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

• Pour spécifier l'ancien job par ID au lieu de son nom, utilisez --additional-experiments="parallel_replace_job_id=OLD_JOB_ID".
• Si vous spécifiez un nouveau nom de job et utilisez l'indicateur --update, Dataflow recherche un job existant portant le nouveau nom, ce qui provoque une erreur.

Facultatif : configurez le délai avant expiration du drainage et l'annulation automatique

Vous pouvez ajouter les options suivantes à l'une ou l'autre des configurations pour définir un délai avant vidange et annuler automatiquement l'ancienne tâche si elle est bloquée.

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

Si vous désactivez l'annulation automatique et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.

Facultatif : Upsert (créer ou mettre à jour un job)

Pour activer le comportement d'upsert (créer ou mettre à jour une tâche) :

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

Remplacez les variables suivantes :

• Si vous effectuez une mise à jour en utilisant un nom de job différent (option 2), vous devez fournir parallel_replace_job_name ou parallel_replace_job_id pour identifier le job à remplacer. La mise à jour avec un nom de tâche différent n'est pas compatible avec Terraform ni Config Connector.
  • OLD_JOB_NAME : nom du job à remplacer.
  • OLD_JOB_ID : ID du job à remplacer.

• DURATION : durée minimale pendant laquelle les deux pipelines s'exécutent en parallèle, sous forme de nombre entier ou à virgule flottante. Une durée de 0s est recommandée pour éviter tout chevauchement. Une fois cette durée écoulée, un signal de vidange est envoyé à l'ancien job.

  La durée doit être comprise entre 0 seconde (0s) et 31 jours (744h). Utilisez s, m et h pour spécifier les secondes, les minutes et les heures. Par exemple, 10m correspond à 10 minutes.

• DRAIN_TIMEOUT_DURATION : facultatif. Durée maximale pendant laquelle l'ancien job doit se vider avant le déclenchement de l'annulation automatique. La durée doit être mise en forme sous forme de chaîne se terminant par s, m ou h (par exemple, 30m ou 1h).

• parallel_replace_job_cancel_on_drain_timeout : facultatif. Indique s'il faut annuler le job précédent s'il ne se termine pas avant la durée d'arrêt maximale. La valeur par défaut est true si une durée de délai avant vidange est fournie. Pour désactiver la résiliation automatique, définissez cette option sur false. Si vous définissez cette option sur false et que l'ancien job reste bloqué dans l'état de vidange, l'ancien et le nouveau job continuent de s'exécuter en parallèle.

Lorsque vous lancez le nouveau job, Dataflow attend que tous les nœuds de calcul soient provisionnés avant de commencer à traiter les données. Pour surveiller l'état du déploiement, consultez les journaux des jobs Dataflow.

Exécuter manuellement des pipelines en parallèle

Pour les scénarios plus complexes ou lorsque vous avez besoin de mieux contrôler le processus de mise à jour, vous pouvez exécuter manuellement des pipelines parallèles. Laissez le pipeline existant s'exécuter jusqu'à ce que son filigrane dépasse l'horodatage de la première fenêtre complète traitée par le pipeline mis à jour. Ensuite, drainez ou annulez le pipeline existant.

Gérer les résultats en double

L'exemple suivant décrit une approche pour gérer les résultats en double. Les deux pipelines écrivent les résultats dans des destinations différentes, utilisent des systèmes en aval pour interroger les résultats et dédupliquent les résultats de la période de chevauchement. Cet exemple utilise un pipeline qui lit les données d'entrée à partir de Pub/Sub, effectue un traitement et écrit les résultats dans BigQuery.

1. Dans l'état initial, le pipeline de streaming existant (Pipeline A) s'exécute et lit des messages à partir d'un sujet Pub/Sub (Topic) à l'aide d'un abonnement (Subscription A). Les résultats sont écrits dans une table BigQuery (Table A). Ils sont utilisés via une vue BigQuery, qui sert de façade pour masquer les modifications de table sous-jacentes. Il s'agit d'une méthode de conception appelée modèle de façade. Le diagramme suivant illustre l'état initial.

   

2. Créez un nouvel abonnement (Subscription B) pour le pipeline mis à jour. Déployez le pipeline mis à jour (Pipeline B), qui lit le sujet Pub/Sub (Topic) à l'aide de l'abonnement Subscription B et écrit dans une table BigQuery distincte (Table B). Le diagramme suivant illustre ce flux.

   

   À ce stade, les pipelines Pipeline A et Pipeline B s'exécutent en parallèle et écrivent les résultats dans des tables distinctes. Vous enregistrez le temps t en tant qu'horodatage de la fenêtre complète la plus ancienne traitée par le pipeline B.

3. Lorsque le filigrane du Pipeline A dépasse le temps t, drainez le Pipeline A. Lorsque vous drainez le pipeline, toutes les fenêtres ouvertes se ferment et le traitement des données en cours se termine. Si le pipeline contient des fenêtres et que des fenêtres complètes sont importantes (en supposant qu'elles ne contiennent aucune donnée tardive), laissez les deux pipelines s'exécuter jusqu'au chevauchement des fenêtres complètes avant de drainer le Pipeline A. Arrêtez le job de traitement en flux continu du Pipeline A une fois que toutes les données en cours ont été traitées et écrites dans la Table A. Le diagramme suivant illustre ce stade.

   

4. À ce stade, seul le pipeline B est en cours d'exécution. Vous pouvez effectuer une requête à partir d'une vue BigQuery (vue de façade), qui sert de façade pour masquer la table A et la table B. Pour les lignes ayant le même horodatage dans les deux tables, configurez la vue de façon à renvoyer les lignes de la Table B ou revenir à la Table A si les lignes n'existent pas dans la Table B. Le diagramme suivant montre la vue (vue de façade) lue à la fois dans la table A et la table B.

   

   À ce stade, vous pouvez supprimer l'abonnement Subscription A.

Lorsque des problèmes sont détectés avec un nouveau déploiement de pipeline, l'utilisation de pipelines en parallèle peut simplifier le rollback. Dans cet exemple, vous souhaiterez peut-être maintenir l'exécution du Pipeline A pendant que vous surveillez le bon fonctionnement du Pipeline B. En cas de problème avec le Pipeline B, vous pouvez effectuer un rollback et restaurer le Pipeline A.

Gérer les mutations de schéma

Les systèmes de gestion de données doivent souvent s'adapter aux mutations de schéma au fil du temps, parfois en raison de l'évolution des besoins métier ou pour des raisons techniques. L'application de mises à jour de schéma nécessite généralement une planification et une exécution minutieuses pour éviter de perturber les systèmes d'information de l'entreprise.

Prenons l'exemple d'un pipeline qui lit les messages contenant des charges utiles JSON à partir d'un sujet Pub/Sub. Le pipeline convertit chaque message en instance TableRow, puis écrit les lignes dans une table BigQuery. Le schéma de la table de sortie est semblable aux messages traités par le pipeline. Dans le diagramme suivant, le schéma est nommé Schéma A.

Au fil du temps, le schéma du message peut subir des mutations complexes. Par exemple, des champs sont ajoutés, supprimés ou remplacés. Le Schéma A évolue en un nouveau schéma. Dans la discussion qui suit, le nouveau schéma est nommé Schéma B. Dans ce cas, le pipeline A doit être mis à jour et le schéma de la table de sortie doit être compatible avec le schéma B.

Pour la table de sortie, vous pouvez effectuer certaines mutations de schéma sans temps d'arrêt. Pour donner un exemple, vous pouvez ajouter de nouveaux champs ou assouplir les modes de colonne, par exemple en remplaçant REQUIRED par NULLABLE, sans temps d'arrêt. Ces mutations n'ont généralement pas d'incidence sur les requêtes existantes. Cependant, les mutations de schéma qui modifient ou suppriment des champs de schéma existants interrompent les requêtes ou entraînent d'autres interruptions. L'approche suivante permet d'adapter les modifications sans nécessiter de temps d'arrêt.

Séparez les données écrites par le pipeline dans une table principale et dans une ou plusieurs tables de préproduction. La table principale stocke les données historiques écrites par le pipeline. Les tables de préproduction stockent les derniers résultats du pipeline. Vous pouvez définir une vue de façade BigQuery pour les tables principales et les tables de préproduction. Cela permet aux utilisateurs d'interroger aussi bien des données historiques que des données mises à jour.

Le diagramme suivant montre comment modifier le flux de pipeline précédent pour inclure une table de préproduction (Staging Table A), une table principale et une vue de façade.

Dans le flux révisé, le pipeline A traite les messages qui utilisent le schéma A et écrit le résultat dans la table de préproduction A, qui dispose d'un schéma compatible. La table principale contient des données historiques écrites par des versions précédentes du pipeline, ainsi que des résultats fusionnés périodiquement à partir de la table de préproduction. Les utilisateurs peuvent interroger des données à jour et des données historiques en temps réel, à l'aide de la vue de façade.

Lorsque le schéma du message évolue du schéma A au schéma B, vous pouvez mettre à jour le code du pipeline pour qu'il soit compatible avec les messages qui utilisent le schéma B. Le pipeline existant doit être mis à jour avec la nouvelle mise en œuvre. En exécutant des pipelines en parallèle, vous avez l'assurance que le traitement des flux de données se poursuit sans interruption. L'arrêt et le remplacement de pipelines entraînent une interruption du traitement, car aucun pipeline ne s'exécute pendant un certain temps.

Le pipeline mis à jour écrit dans une table de préproduction supplémentaire (table de préproduction B) qui utilise le schéma B. Vous pouvez utiliser un workflow orchestré pour créer la nouvelle table de préproduction avant de mettre à jour le pipeline. Mettez à jour la vue de façade de façon à inclure les résultats de la nouvelle table de préproduction, en utilisant éventuellement une étape de workflow associée.

Le diagramme suivant montre le flux mis à jour, la table de préproduction B avec le schéma B et la façon dont la vue de façade est mise à jour pour inclure le contenu de la table principale et des deux tables de préproduction.

Dans le cadre d'un processus distinct de la mise à jour du pipeline, vous pouvez fusionner les tables de préproduction dans la table principale, périodiquement ou selon les besoins. Le diagramme suivant montre la fusion de la table de préproduction A dans la table principale.

Étapes suivantes

• Découvrez les étapes détaillées permettant de mettre à jour un pipeline existant.