Réexécuter un pipeline

Vous pouvez relancer des tâches spécifiques à partir d'exécutions de pipeline de machine learning (ML) terminées, annulées ou ayant échoué. Lorsque vous lancez une nouvelle exécution, vous pouvez modifier les configurations au niveau de la tâche ou choisir d'ignorer des tâches, puis créer une exécution basée sur la configuration mise à jour. La nouvelle exécution du pipeline conserve une référence à l'exécution d'origine pour assurer 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 nouvelle 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.

La relance d'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 cette fonctionnalité est utile :

  • Gestion des échecs partiels dans les processus parallèles : lorsqu'une partie d'un processus parallèle volumineux é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 pour une tâche sur 100 échoue, vous pouvez l'ignorer.

  • Relance d'une tâche avec des données d'entrée mises à jour : si une seule tâche doit être relancée avec des données mises à jour, vous pouvez la relancer spécifiquement.

  • Débogage des problèmes de production sans nécessiter de modifications de code : relancez une tâche spécifique et toutes les tâches qui en dépendent sans impliquer l'auteur du code du pipeline.

Relancer un pipeline

Pour relancer un pipeline, utilisez le SDK Vertex AI pour Python.

Python

Utilisez l'exemple suivant pour relancer un pipeline en ignorant une tâche ayant échoué et en relançant 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 contenant l'exécution du pipeline. Google Cloud
  • 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 du pipeline terminée, ayant échoué ou annulée que vous souhaitez relancer. 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 Google Cloud console. Pour en savoir plus, consultez Trouver le nom, le numéro et l'ID du projet.
    • PIPELINE_RUN_ID : ID unique de l'exécution du pipeline que vous souhaitez relancer. L'ID s'affiche dans l'onglet Exécutions de la page Pipelines de la Google Cloud console.
  • 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 nouvelle exécution du pipeline.
  • TASK_PARAMETER_1_VALUE et TASK_PARAMETER_2_VALUE : nouvelles valeurs de TASK_PARAMETER_1 et TASK_PARAMETER_2 respectivement lors de la nouvelle exécution du pipeline.
  • PIPELINE_PARAMETER_VALUES : facultatif. Valeurs de paramètres mises à jour au niveau de l'exécution du pipeline à utiliser pour la nouvelle exécution du pipeline.
  • RERUN_PIPELINE_JOB_ID : facultatif. ID unique à attribuer à la nouvelle tâche de nouvelle exécution du pipeline.