Volver a ejecutar un flujo de procesamiento

Puedes volver a ejecutar tareas específicas de ejecuciones de canalizaciones de aprendizaje automático (ML) completadas, canceladas o fallidas. Cuando inicias una repetición, puedes modificar las configuraciones a nivel de tarea o elegir omitir tareas y, a continuación, crear una ejecución basada en la configuración actualizada. La nueva ejecución de flujo de procesamiento mantiene una referencia a la ejecución de flujo de procesamiento original para la trazabilidad. Si una tarea se ha completado correctamente en una ejecución anterior, Vertex AI Pipelines reutiliza los resultados almacenados en caché de esa tarea. De lo contrario, si el paso ha fallado, Vertex AI Pipelines lo ejecuta durante la repetición del flujo de procesamiento.

De esta forma, puedes solucionar de forma eficiente los errores del flujo de procesamiento de aprendizaje automático haciendo ajustes en él sin tener que reiniciarlo por completo. Puedes ajustar una tarea fallida, comparar los resultados con diferentes conjuntos de parámetros o saltarte una tarea no esencial que esté fallando.

Volver a ejecutar un flujo de procesamiento es útil para los profesionales de MLOps que gestionan flujos de procesamiento de aprendizaje automático complejos en producción. Estos son algunos ejemplos de situaciones en las que resulta útil:

  • Gestionar fallos parciales en procesos paralelos: cuando falla una parte de un proceso paralelo grande, puedes omitir la tarea fallida y dejar que continúe el resto del flujo de procesamiento. Por ejemplo, si falla una de las 100 tareas de una canalización de datos, puedes omitirla.

  • Volver a ejecutar una tarea con datos de entrada actualizados: si es necesario volver a ejecutar una tarea con datos actualizados, puedes hacerlo.

  • Depurar problemas de producción sin necesidad de cambiar el código: vuelve a ejecutar una tarea específica y todas las tareas que dependan de ella sin involucrar al autor del código de la canalización.

Volver a ejecutar un flujo de procesamiento

Para volver a ejecutar una canalización, usa el SDK de Vertex AI para Python.

Python

Usa el siguiente ejemplo para volver a ejecutar una canalización omitiendo una tarea fallida y volviendo a ejecutar otra tarea con parámetros actualizados mediante el método 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)

Haz los cambios siguientes:

  • PROJECT_ID: el Google Cloud proyecto que contiene la ejecución del flujo de procesamiento.
  • LOCATION: la región en la que se encuentra la ejecución del flujo de procesamiento. Para obtener más información sobre las regiones en las que está disponible Vertex AI Pipelines, consulta la guía de ubicaciones de Vertex AI.

  • PIPELINE_RUN_RESOURCE_NAME: nombre de recurso completo de la ejecución de la canalización completada, fallida o cancelada que quieres volver a ejecutar. Introduce el nombre del recurso con el formato projects/PROJECT_NUMBER/locations/LOCATION/pipelineJobs/PIPELINE_RUN_ID, donde:

    • PROJECT_NUMBER: el número de tu proyecto. Puedes encontrar este número de proyecto en la Google Cloud consola. Para obtener más información, consulta Buscar el nombre, el número y el ID del proyecto.
    • PIPELINE_RUN_ID con el ID único de la ejecución de la canalización que quieras volver a ejecutar. El ID se muestra en la pestaña Ejecuciones de la página Pipelines de la consola Google Cloud .
  • RERUN_TASK_NAME: el nombre de la tarea que se va a volver a ejecutar con los parámetros actualizados.
  • SKIP_FAILED_TASK_NAME: el nombre de la tarea fallida que se va a omitir durante la repetición.
  • TASK_PARAMETER_1 y TASK_PARAMETER_2: los nombres de los parámetros de la tarea que quieras anular en la repetición de la canalización.
  • TASK_PARAMETER_1_VALUE y TASK_PARAMETER_2_VALUE: Los nuevos valores de TASK_PARAMETER_1 y TASK_PARAMETER_2 respectivamente en la repetición de la canalización.
  • PIPELINE_PARAMETER_VALUES: opcional. Los valores de los parámetros a nivel de ejecución de la canalización actualizados que se deben usar para volver a ejecutar la canalización.
  • RERUN_PIPELINE_JOB_ID: opcional. ID único que se asignará al nuevo trabajo de repetición de la canalización.