Executar um pipeline novamente

É possível executar novamente tarefas específicas de execuções de pipeline de machine learning (ML) concluídas, canceladas ou com falha. Ao iniciar uma nova execução, é possível modificar as configurações no nível da tarefa ou pular tarefas e criar uma execução com base na configuração atualizada. A nova execução de pipeline mantém uma referência à execução original para rastreabilidade. Se uma tarefa foi concluída em uma execução anterior, os pipelines da Gemini Enterprise Agent Platform reutilizam os resultados armazenados em cache dessa tarefa. Caso contrário, se a etapa tiver falhado, os pipelines do Agent Platform vão executá-la durante a nova execução do pipeline.

Isso permite resolver falhas no pipeline de ML de maneira eficiente, fazendo ajustes sem precisar reiniciar todo o pipeline. É possível ajustar uma tarefa com falha, comparar resultados com diferentes conjuntos de parâmetros ou pular uma tarefa não essencial que está falhando.

Executar um pipeline novamente é útil para profissionais de MLOps que gerenciam pipelines de ML complexos em produção. Exemplos de cenários em que isso é útil:

  • Como lidar com falhas parciais em processos paralelos: quando uma parte de um processo paralelo grande falha, é possível pular a tarefa com falha e deixar o restante do pipeline continuar sendo executado. Por exemplo, se um pipeline de dados para uma de 100 tarefas falhar, você poderá ignorá-lo.

  • Executar novamente uma tarefa com dados de entrada atualizados: se uma única tarefa precisar ser executada novamente com dados atualizados, você poderá fazer isso.

  • Depurar problemas de produção sem exigir mudanças no código: execute novamente uma tarefa específica e todas as tarefas que dependem dela sem envolver o autor do código do pipeline.

Executar um pipeline novamente

Para executar um pipeline novamente, use o SDK da Agent Platform para Python.

Python

Use o exemplo a seguir para executar novamente um pipeline pulando uma tarefa com falha e executando outra tarefa com parâmetros atualizados usando o 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)

Substitua:

  • PROJECT_ID: o Google Cloud projeto que contém a execução do pipeline.
  • LOCATION: a região em que a execução do pipeline está localizada. Para mais informações sobre as regiões em que o Agent Platform Pipelines está disponível, consulte o Guia de locais da plataforma de agentes do Gemini Enterprise.

  • PIPELINE_RUN_RESOURCE_NAME: o nome de recurso totalmente qualificado da execução de pipeline concluída, com falha ou cancelada que você quer executar novamente. Insira o nome do recurso no formato projects/PROJECT_NUMBER/locations/LOCATION/pipelineJobs/PIPELINE_RUN_ID, em que:

    • PROJECT_NUMBER: o número do projeto. Você pode localizar esse número do projeto no console Google Cloud . Para mais informações, consulte Encontrar o nome, o número e o ID do projeto.
    • PIPELINE_RUN_ID com o ID exclusivo da execução do pipeline que você quer executar novamente. O ID é exibido na guia Execuções da página Pipelines no console do Google Cloud .
  • RERUN_TASK_NAME: o nome da tarefa a ser executada novamente com parâmetros atualizados.
  • SKIP_FAILED_TASK_NAME: o nome da tarefa com falha a ser ignorada durante a nova execução.
  • TASK_PARAMETER_1 e TASK_PARAMETER_2: os nomes dos parâmetros da tarefa que você quer substituir na nova execução do pipeline.
  • TASK_PARAMETER_1_VALUE e TASK_PARAMETER_2_VALUE: Os novos valores de TASK_PARAMETER_1 e TASK_PARAMETER_2 respectivamente na nova execução do pipeline.
  • PIPELINE_PARAMETER_VALUES: opcional. Os valores de parâmetro atualizados no nível da execução do pipeline a serem usados para a nova execução.
  • RERUN_PIPELINE_JOB_ID: opcional. Um ID exclusivo a ser atribuído ao novo job de nova execução do pipeline.