Criar um pipeline

O Vertex AI Pipelines permite orquestrar fluxos de trabalho de machine learning (ML) sem servidor. Antes que os pipelines do Vertex AI possam orquestrar seu fluxo de trabalho de ML, você precisa descrever seu fluxo de trabalho como um pipeline. Os pipelines de ML são fluxos de trabalho de ML portáteis e escalonáveis baseados em contêineres e serviços do Google Cloud.

Neste guia, descrevemos como começar a criar pipelines de ML.

Que SDK de pipelines devo usar?

O Vertex AI Pipelines pode executar pipelines criados usando qualquer um dos SDKs a seguir:

  • SDK do Kubeflow Pipelines v1.8 ou posterior (a v2 é recomendada)

  • TensorFlow Extended v0.30.0 ou mais recente

Se você usa o TensorFlow em um fluxo de trabalho de ML que processa terabytes de dados estruturados ou de texto, recomendamos criar o pipeline usando o TFX.

Para outros casos de uso, recomendamos que você crie seu pipeline usando o SDK do Kubeflow Pipelines. Ao criar um pipeline com o SDK do Kubeflow Pipelines, é possível implementar seu fluxo de trabalho criando componentes personalizados ou reutilizando componentes pré-criados, como os componentes do pipeline do Google Cloud. Os componentes do pipeline do Google Cloud facilitam o uso de serviços de Vertex IA, como o AutoML, no seu pipeline.

Neste guia, descrevemos como criar pipelines usando o SDK do Kubeflow Pipelines.

Antes de começar

Antes de criar e executar os pipelines, use as instruções a seguir para configurar o projeto do Google Cloud e o ambiente de desenvolvimento.

  1. Para preparar o projeto do Google Cloud para executar pipelines de ML, siga as instruções no guia Como configurar o projeto do Google Cloud.

  2. Para criar seu pipeline usando o SDK do Kubeflow Pipelines, instale o SDK do Kubeflow Pipelines v1.8 ou posterior.

  3. Para usar o cliente Vertex AI Python nos pipelines, instale as bibliotecas de cliente Vertex AI v1.7 ou mais recente.

  4. Para usar os serviços do Vertex AI nos pipelines, instale os componentes de pipeline do Google Cloud.

Primeiros passos para criar um pipeline

Para orquestrar seu fluxo de trabalho de ML no Pipelines do Vertex AI, primeiro é preciso descrever seu fluxo de trabalho como um pipeline. A amostra a seguir demonstra como usar componentes de pipeline do Google Cloud para que a Vertex AI crie um conjunto de dados, treine um modelo usando AutoML e o implante para fazer previsões.

Antes de executar o código de exemplo a seguir, é preciso configurar a autenticação.

Como configurar a autenticação

Para configurar a autenticação, crie uma chave de conta de serviço e defina uma variável de ambiente para o caminho dessa chave.

  1. Crie uma conta de serviço:

    1. No Console do Google Cloud, acesse a página Criar conta de serviço.

      Acesse "Criar conta de serviço"

    2. No campo Nome da conta de serviço, insira um nome.
    3. Opcional: no campo Descrição da conta de serviço, digite uma descrição.
    4. Clique em Criar.
    5. Clique no campo Selecionar um papel. Em Todos os papéis, selecione Vertex AI > Usuário da Vertex AI.

    6. Clique em Concluído para criar a conta de serviço.

      Não feche a janela do navegador. Você vai usá-la na próxima etapa.

  2. Crie uma chave da conta de serviço para autenticação:

    1. No console do Google Cloud, clique no endereço de e-mail da conta de serviço que você criou.
    2. Clique em Chaves.
    3. Clique em Adicionar chave e, depois, em Criar nova chave.
    4. Clique em Criar. O download de um arquivo de chave JSON é feito no seu computador.
    5. Clique em Fechar.
  3. Conceda à nova conta de serviço acesso à conta de serviço usada para executar pipelines.
    1. Clique em para retornar à lista de contas de serviço.

    2. Clique no nome da conta de serviço que você usa para executar pipelines. A página Detalhes da conta de serviço é exibida.

      Se você tiver seguido as instruções no guia para configurar o projeto para o Vertex AI Pipelines, essa é a mesma conta de serviço criada em Configurar uma conta de serviço com permissões granulares. Caso contrário, o Vertex AI usa a conta de serviço padrão do Compute Engine para executar pipelines. O nome da conta de serviço padrão do Compute Engine tem o seguinte formato: PROJECT_NUMBER-compute@developer.gserviceaccount.com

    3. Clique na guia Permissões..
    4. Clique em Conceder acesso. O painel Adicionar principais será exibido.
    5. Na caixa Novos principais, digite o endereço de e-mail da conta de serviço que você criou em uma etapa anterior.
    6. Na lista suspensa Papel, selecione Contas de serviço > Usuário da conta de serviço.
    7. Clique em Salvar

  4. Defina a variável de ambiente GOOGLE_APPLICATION_CREDENTIALS como o caminho do arquivo JSON que contém a chave da conta de serviço: Essa variável aplica-se apenas à sessão de shell atual. Portanto, se você abrir uma nova sessão, defina a variável novamente.

    Exemplo: Linux ou macOS

    Substitua [PATH] pelo caminho do arquivo JSON que contém a chave da conta de serviço.

    export GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

    Exemplo:

    export GOOGLE_APPLICATION_CREDENTIALS="/home/user/Downloads/service-account-file.json"

    Exemplo: Windows

    Substitua [PATH] pelo caminho do arquivo JSON que contém a chave da conta de serviço e [FILE_NAME] pelo nome de arquivo.

    Com o PowerShell:

    $env:GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

    Exemplo:

    $env:GOOGLE_APPLICATION_CREDENTIALS="C:\Users\username\Downloads\[FILE_NAME].json"

    Com prompt de comando:

    set GOOGLE_APPLICATION_CREDENTIALS=[PATH]

Definir seu fluxo de trabalho usando o pacote DSL do Kubeflow Pipelines

O pacote kfp.dsl contém a linguagem específica de domínio (DSL, na sigla em inglês) que pode ser usada para definir e interagir com pipelines e componentes.

Os componentes do pipeline do Kubeflow são funções de fábrica que criam etapas de pipeline. Cada componente descreve as entradas, saídas e a implementação do componente. Por exemplo, no exemplo de código abaixo, ds_op é um componente.

Os componentes são usados para criar etapas de pipeline. Quando um pipeline é executado, as etapas são executadas à medida que os dados de que dependem ficam disponíveis. Por exemplo, um componente de treinamento pode usar um arquivo CSV como entrada e usá-lo para treinar um modelo.

import kfp
from google.cloud import aiplatform
from google_cloud_pipeline_components.v1.dataset import ImageDatasetCreateOp
from google_cloud_pipeline_components.v1.automl.training_job import AutoMLImageTrainingJobRunOp
from google_cloud_pipeline_components.v1.endpoint import EndpointCreateOp, ModelDeployOp

project_id = PROJECT_ID
pipeline_root_path = PIPELINE_ROOT

# Define the workflow of the pipeline.
@kfp.dsl.pipeline(
    name="automl-image-training-v2",
    pipeline_root=pipeline_root_path)
def pipeline(project_id: str):
    # The first step of your workflow is a dataset generator.
    # This step takes a Google Cloud Pipeline Component, providing the necessary
    # input arguments, and uses the Python variable `ds_op` to define its
    # output. Note that here the `ds_op` only stores the definition of the
    # output but not the actual returned object from the execution. The value
    # of the object is not accessible at the dsl.pipeline level, and can only be
    # retrieved by providing it as the input to a downstream component.
    ds_op = ImageDatasetCreateOp(
        project=project_id,
        display_name="flowers",
        gcs_source="gs://cloud-samples-data/vision/automl_classification/flowers/all_data_v2.csv",
        import_schema_uri=aiplatform.schema.dataset.ioformat.image.single_label_classification,
    )

    # The second step is a model training component. It takes the dataset
    # outputted from the first step, supplies it as an input argument to the
    # component (see `dataset=ds_op.outputs["dataset"]`), and will put its
    # outputs into `training_job_run_op`.
    training_job_run_op = AutoMLImageTrainingJobRunOp(
        project=project_id,
        display_name="train-iris-automl-mbsdk-1",
        prediction_type="classification",
        model_type="CLOUD",
        dataset=ds_op.outputs["dataset"],
        model_display_name="iris-classification-model-mbsdk",
        training_fraction_split=0.6,
        validation_fraction_split=0.2,
        test_fraction_split=0.2,
        budget_milli_node_hours=8000,
    )

    # The third and fourth step are for deploying the model.
    create_endpoint_op = EndpointCreateOp(
        project=project_id,
        display_name = "create-endpoint",
    )

    model_deploy_op = ModelDeployOp(
        model=training_job_run_op.outputs["model"],
        endpoint=create_endpoint_op.outputs['endpoint'],
        automatic_resources_min_replica_count=1,
        automatic_resources_max_replica_count=1,
    )

Substitua:

  • PROJECT_ID: o projeto do Google Cloud em que este pipeline é executado.

  • PIPELINE_ROOT_PATH: especifique um URI do Cloud Storage que sua conta de serviço de pipelines possa acessar. Os artefatos das execuções de pipeline são armazenados na raiz do pipeline.

    A raiz do pipeline pode ser definida como um argumento da @kfp.dsl.pipeline anotação na função do pipeline ou quando você chama create_run_from_job_spec para criar uma execução de pipeline.

Compilar o pipeline em um arquivo YAML

Depois que o fluxo de trabalho for definido, será possível compilar o pipeline no formato YAML. O arquivo YAML inclui todas as informações para executar o pipeline no Vertex AI Pipelines.

from kfp import compiler

compiler.Compiler().compile(
    pipeline_func=pipeline,
    package_path='image_classif_pipeline.yaml'
)

Enviar a execução do pipeline

Depois que o fluxo de trabalho do pipeline é compilado no formato JSON, é possível usar o cliente Python da Vertex AI para enviar e executar seu pipeline.

import google.cloud.aiplatform as aip

# Before initializing, make sure to set the GOOGLE_APPLICATION_CREDENTIALS
# environment variable to the path of your service account.
aip.init(
    project=project_id,
    location=PROJECT_REGION,
)

# Prepare the pipeline job
job = aip.PipelineJob(
    display_name="automl-image-training-v2",
    template_path="image_classif_pipeline.yaml",
    pipeline_root=pipeline_root_path,
    parameter_values={
        'project_id': project_id
    }
)

job.submit()

Substitua:

  • PROJECT_REGION: a região em que este pipeline é executado.

No exemplo anterior:

  1. Um pipeline do Kubeflow é definido como uma função do Python. A função é anotada com o decorador @kfp.dsl.pipeline, que especifica o nome do pipeline e o caminho raiz. O caminho raiz do pipeline é o local onde os artefatos do pipeline são armazenados.
  2. As etapas do fluxo de trabalho do pipeline são criadas usando os componentes do pipeline do Google Cloud. Ao usar as saídas de um componente como entrada de outro componente, defina o fluxo de trabalho do pipeline como um gráfico. Por exemplo: training_job_run_op depende da saída dataset de ds_op.
  3. Você compila o pipeline usando kfp.compiler.Compiler.
  4. Crie um pipeline executado no Vertex AI Pipelines usando o cliente Python Vertex AI. Ao executar um pipeline, é possível substituir o nome do pipeline e o caminho raiz do pipeline. As execuções de pipeline podem ser agrupadas usando o nome do pipeline. A modificação do nome do pipeline pode ajudar você a distinguir entre as execuções de pipeline de produção e experimentais.

Para saber mais sobre a criação de pipelines, leia a seção Como criar pipelines do Kubeflow e siga as amostras e tutoriais.

Testar um pipeline localmente (opcional)

Depois de definir os pipelines e os componentes, teste o código do componente executando o código no ambiente de criação local. Ao executar o pipeline ou um componente localmente, identifique e depure possíveis problemas antes de criar uma execução de pipeline em um ambiente remoto, como o Vertex AI Pipelines. Para mais informações sobre a execução local de pipelines e componentes, consulte Execução local na documentação do KFP.

Nesta página, mostramos como definir e executar um pipeline que consiste em duas tarefas.

Configurar seu ambiente local

  1. Instale a versão secundária mais recente do SDK do Kubeflow Pipelines (KFP) v2:

    pip install --upgrade kfp>=2,<3

  2. Opcional: instale o Docker.

  3. Use o exemplo de código a seguir para definir um pipeline simples:

    from kfp import dsl

# Define a component to add two numbers.
@dsl.component
def add(a: int, b: int) -> int:
    return a + b

# Define a simple pipeline using the component.
@dsl.pipeline
def addition_pipeline(x: int, y: int, z: int) -> int:
    task1 = add(a=x, b=y)
    task2 = add(a=task1.output, b=z)
    return task2.output

Invocar uma execução local

Inicialize uma sessão local usando a função local.init(). Quando você usa local.init(), o SDK do KFP executa localmente seus pipelines e componentes quando você os chama.

Ao usar local.init(), é necessário especificar um tipo de executor. O tipo de executor indica como o KFP executará cada tarefa.

Use o exemplo a seguir para especificar o tipo de executor DockerRunner para executar cada tarefa em um contêiner. Para mais informações sobre os executores locais compatíveis com o KFP, consulte Executores locais na documentação do KFP.

from kfp import local

local.init(runner=local.DockerRunner())

pipeline_task = addition_pipeline(x=1, y=2, z=3)

Use o código a seguir para visualizar a saída da tarefa do pipeline na execução local:

print(f'Result: {pipeline_task.output}')

Como criar pipelines do Kubeflow

Use o processo a seguir para criar um pipeline.

  1. Projetar seu pipeline como uma série de componentes. Para promover a reutilização, cada componente precisa ter uma única responsabilidade. Sempre que possível, projete o pipeline para reutilizar componentes comprovados, como os componentes do Google Cloud.

    Saiba mais sobre a criação de pipelines.

  2. Crie qualquer componente personalizado necessário para implementar seu fluxo de trabalho de ML usando o SDK do Kubeflow Pipelines. Componentes são conjuntos autocontidos de código que executam uma etapa no fluxo de trabalho de ML. Use as opções a seguir para criar os componentes do pipeline.

  3. Crie o pipeline como uma função do Python.

    Saiba mais sobre como definir o pipeline como uma função do Python.

  4. Use o compilador do SDK do Kubeflow Pipelines para compilar o pipeline.

    from kfp import compiler

compiler.Compiler().compile(
    pipeline_func=PIPELINE_FUNCTION,
    package_path=PIPELINE_PACKAGE_PATH)

    Substitua:

    • PIPELINE_FUNCTION: o nome da função do pipeline.
    • PIPELINE_PACKAGE_PATH: o caminho para armazenar o pipeline compilado.

  5. Execute o pipeline usando o Console do Google Cloud ou o Python.

Como acessar recursos do Google Cloud em um pipeline

Se você não especificar uma conta de serviço ao executar um pipeline, o Vertex AI Pipelines usará a conta de serviço padrão do Compute Engine para executar o pipeline. O Vertex AI Pipelines também usa a conta de serviço de uma execução de pipeline para autorizar o pipeline a acessar os recursos do Google Cloud. A conta de serviço padrão do Compute Engine tem o papel Editor do projeto por padrão. Isso pode conceder aos pipelines acesso extra aos recursos do Google Cloud no seu projeto do Google Cloud.

Recomendamos que você crie uma conta de serviço para executar os pipelines e, em seguida, conceda a ela permissões granulares para os recursos do Google Cloud necessários para executar o pipeline.

Saiba mais sobre como usar o gerenciamento de identidade e acesso para criar uma conta de serviço e gerenciar o acesso concedido a uma conta de serviço.

Mantenha seus pipelines atualizados

Os clientes do SDK e as imagens de contêiner usadas para criar e executar pipelines são atualizados periodicamente para novas versões a fim de corrigir vulnerabilidades de segurança e adicionar novas funcionalidades. Para manter seus pipelines atualizados com a versão mais recente, recomendamos que você faça o seguinte:

A seguir