Réexécuter un pipeline

Vous pouvez réexécuter des tâches spécifiques à partir d'exécutions de pipelines de machine learning (ML) terminées, annulées ou ayant échoué. Lorsque vous lancez une réexécution, vous pouvez modifier les configurations au niveau des tâches ou choisir d'ignorer des tâches, puis créer une exécution basée sur la configuration mise à jour. La nouvelle exécution de pipeline conserve une référence à l'exécution de pipeline d'origine pour la traçabilité. Si une tâche a réussi lors d'une exécution précédente, Vertex AI Pipelines réutilise les résultats mis en cache pour cette tâche. Sinon, si l'étape a échoué, Vertex AI Pipelines l'exécute lors de la réexécution du pipeline.

Cela vous permet de résoudre efficacement les échecs de pipeline de ML en ajustant votre pipeline de ML sans le redémarrer entièrement. Vous pouvez ajuster une tâche ayant échoué, comparer les résultats avec différents ensembles de paramètres ou ignorer une tâche non essentielle qui échoue.

Réexécuter un pipeline est utile pour les professionnels du MLOps qui gèrent des pipelines de ML complexes en production. Voici quelques exemples de scénarios dans lesquels cela peut être utile :

  • Gérer les échecs partiels dans les processus parallèles : lorsqu'une partie d'un grand processus parallèle échoue, vous pouvez ignorer la tâche ayant échoué et laisser le reste du pipeline continuer à s'exécuter. Par exemple, si un pipeline de données échoue pour une tâche sur 100, vous pouvez l'ignorer.

  • Réexécuter une tâche avec des données d'entrée mises à jour : si une seule tâche doit être réexécutée avec des données mises à jour, vous pouvez réexécuter cette tâche spécifique.

  • Déboguer les problèmes de production sans avoir à modifier le code : réexécutez une tâche spécifique et toutes les tâches qui en dépendent sans impliquer l'auteur du code du pipeline.

Réexécuter un pipeline

Pour réexécuter un pipeline, utilisez le SDK Vertex AI pour Python.

Python

Utilisez l'exemple suivant pour réexécuter un pipeline en ignorant une tâche ayant échoué et en réexécutant une autre tâche avec des paramètres mis à jour à l'aide de la méthode PipelineJob.rerun() :

from google.cloud import aiplatform
from google.cloud.aiplatform.preview.pipelinejob.pipeline_jobs import (
  _PipelineJob as PipelineJob
)
from google.cloud.aiplatform_v1beta1.types.ui_pipeline_spec import RuntimeArtifact
from google.protobuf.struct_pb2 import Value
from google.cloud.aiplatform_v1beta1.types import PipelineTaskRerunConfig
aiplatform.init(project="PROJECT_ID", location="LOCATION")
job = aiplatform.PipelineJob.get(resource_name="PIPELINE_RUN_RESOURCE_NAME")
original_job_name = job.resource_name
rerun_task_id = None
skip_failed_task_id = None
task_inputs_override = PipelineTaskRerunConfig.Inputs(
    parameter_values={
      "TASK_PARAMETER_1": Value(TASK_PARAMETER_1_VALUE),
      "TASK_PARAMETER_2": Value(TASK_PARAMETER_2_VALUE)
    }
)
for task in job.task_details:
    if task.task_name == RERUN_TASK_NAME:
        rerun_task_id = task.task_id
    if task.task_name == SKIP_FAILED_TASK_NAME:
        skip_failed_task_id = task.task_id
pipeline_job.rerun(original_pipelinejob_name=original_job_name,
  pipeline_task_rerun_configs=[
    PipelineTaskRerunConfig(task_id = rerun_task_id,
      skip_task = False,
      inputs = PipelineTaskRerunConfig.Inputs(task_inputs_override)
    ),
    PipelineTaskRerunConfig(task_id = skip_failed_task_id,
      skip_task = True
    )
  ],
  parameter_values=PIPELINE_PARAMETER_VALUES,
  job_id=RERUN_PIPELINE_JOB_ID)

Remplacez les éléments suivants :

  • PROJECT_ID : projet Google Cloud contenant l'exécution du pipeline.
  • LOCATION : région dans laquelle se trouve l'exécution du pipeline. Pour en savoir plus sur les régions dans lesquelles Vertex AI Pipelines est disponible, consultez le Guide des emplacements Vertex AI.

  • PIPELINE_RUN_RESOURCE_NAME : nom complet de la ressource de l'exécution de pipeline terminée, ayant échoué ou annulée que vous souhaitez réexécuter. Saisissez le nom de la ressource au format projects/PROJECT_NUMBER/locations/LOCATION/pipelineJobs/PIPELINE_RUN_ID, où :

    • PROJECT_NUMBER : numéro de votre projet. Vous trouverez ce numéro de projet dans la console Google Cloud . Pour en savoir plus, consultez Trouver le nom, le numéro et l'ID du projet.
    • PIPELINE_RUN_ID avec l'ID unique de l'exécution du pipeline que vous souhaitez réexécuter. L'ID s'affiche dans l'onglet Exécutions de la page Pipelines de la console Google Cloud .
  • RERUN_TASK_NAME : nom de la tâche à relancer avec des paramètres mis à jour.
  • SKIP_FAILED_TASK_NAME : nom de la tâche ayant échoué à ignorer lors de la nouvelle exécution.
  • TASK_PARAMETER_1 et TASK_PARAMETER_2 : noms des paramètres de la tâche que vous souhaitez remplacer lors de la réexécution du pipeline.
  • TASK_PARAMETER_1_VALUE et TASK_PARAMETER_2_VALUE : nouvelles valeurs de TASK_PARAMETER_1 et TASK_PARAMETER_2 respectivement dans la réexécution du pipeline.
  • PIPELINE_PARAMETER_VALUES : facultatif. Valeurs des paramètres au niveau de l'exécution du pipeline à utiliser pour la réexécution du pipeline.
  • RERUN_PIPELINE_JOB_ID : facultatif. ID unique à attribuer à la nouvelle tâche de réexécution du pipeline.