Construir un canal

Vertex AI Pipelines te permite orquestar tus flujos de trabajo de aprendizaje automático (AA) sin servidor. Para que Vertex AI Pipelines pueda orquestar tu flujo de trabajo de aprendizaje automático, debes describirlo como un flujo de procesamiento. Los flujos de procesamiento de AA son flujos de trabajo de AA portátiles y escalables basados en contenedores y Google Cloud servicios.

En esta guía se describe cómo empezar a crear canalizaciones de aprendizaje automático.

¿Qué SDK de Pipelines debo usar?

Vertex AI Pipelines puede ejecutar las pipelines creadas con cualquiera de los siguientes SDKs:

  • SDK de Kubeflow Pipelines v2.0 o posterior

  • TensorFlow Extended v0.30.0 o una versión posterior

Si usas TensorFlow en un flujo de trabajo de aprendizaje automático que procesa terabytes de datos estructurados o de texto, te recomendamos que crees tu flujo de procesamiento con TFX.

Para otros casos prácticos, te recomendamos que crees tu canalización con el SDK de Kubeflow Pipelines. Si creas un flujo de procesamiento con el SDK de Kubeflow Pipelines, puedes implementar tu flujo de trabajo creando componentes personalizados o reutilizando componentes precompilados, como los Google Cloud componentes de flujo de procesamiento. Google Cloud Los componentes de flujo de procesamiento facilitan el uso de servicios de Vertex AI, como AutoML, en tu flujo de procesamiento.

En esta guía se describe cómo crear pipelines con el SDK de Kubeflow Pipelines.

Antes de empezar

Antes de compilar y ejecutar tus canalizaciones, sigue estas instrucciones para configurar tu proyecto y tu entorno de desarrollo. Google Cloud

  1. Para preparar tu Google Cloud proyecto para ejecutar las canalizaciones de aprendizaje automático, sigue las instrucciones de la guía para configurar tu Google Cloud proyecto.

  2. Instala la versión 2 o una posterior del SDK de Kubeflow Pipelines.

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

  1. Para usar el cliente de Python de Vertex AI en tus pipelines, instala las bibliotecas de cliente de Vertex AI v1.7 o versiones posteriores.

  2. Para usar los servicios de Vertex AI en tus flujos de trabajo, instala elGoogle Cloud SDK.

Empezar a crear un flujo de procesamiento

Para orquestar tu flujo de trabajo de aprendizaje automático en Vertex AI Pipelines, primero debes describir tu flujo de trabajo como un flujo de procesamiento. En el siguiente ejemplo se muestra cómo usar los Google Cloud componentes de flujo de trabajo con Vertex AI para crear un conjunto de datos, entrenar un modelo con AutoML e implementar el modelo entrenado para obtener predicciones.

Antes de ejecutar el siguiente código de ejemplo, debes configurar la autenticación.

Cómo configurar la autenticación

Para configurar la autenticación, debes crear una clave de cuenta de servicio y definir una variable de entorno para la ruta a la clave de cuenta de servicio.

  1. Crea una cuenta de servicio:

    1. En la Google Cloud consola, ve a la página Crear cuenta de servicio.

      Ir a Crear cuenta de servicio

    2. En el campo Nombre de cuenta de servicio, escribe un nombre.
    3. Opcional: En el campo Descripción de la cuenta de servicio, escribe una descripción.
    4. Haz clic en Crear.
    5. Haz clic en el campo Selecciona un rol. En Todos los roles, selecciona Vertex AI > Usuario de Vertex AI.

    6. Haz clic en Hecho para crear la cuenta de servicio.

      No cierres la ventana del navegador. Lo usarás en el siguiente paso.

  2. Crea una clave de cuenta de servicio para la autenticación:

    1. En la Google Cloud consola, haz clic en la dirección de correo de la cuenta de servicio que has creado.
    2. Haz clic en Teclas.
    3. Haz clic en Añadir clave y, a continuación, en Crear clave.
    4. Haz clic en Crear. Se descargará un archivo de clave JSON en tu ordenador.
    5. Haz clic en Cerrar.
  3. Concede a tu nueva cuenta de servicio acceso a la cuenta de servicio que usas para ejecutar las canalizaciones.
    1. Haz clic en para volver a la lista de cuentas de servicio.

    2. Haz clic en el nombre de la cuenta de servicio que usas para ejecutar las canalizaciones. Se abrirá la página Detalles de la cuenta de servicio.

      Si has seguido las instrucciones de la guía para configurar tu proyecto para Vertex AI Pipelines, esta es la misma cuenta de servicio que creaste en la sección Configurar una cuenta de servicio con permisos granulares. De lo contrario, Vertex AI usará la cuenta de servicio predeterminada de Compute Engine para ejecutar los flujos de trabajo. La cuenta de servicio predeterminada de Compute Engine tiene un nombre como el siguiente: PROJECT_NUMBER-compute@developer.gserviceaccount.com

    3. Haz clic en la pestaña Permisos.
    4. Haz clic en Conceder acceso. Aparecerá el panel Añadir principales.
    5. En el cuadro Nuevos principales, introduce la dirección de correo de la cuenta de servicio que has creado en un paso anterior.
    6. En la lista desplegable Rol, selecciona Cuentas de servicio > Usuario de cuenta de servicio.
    7. Haz clic en Guardar.

  4. Define la variable de entorno GOOGLE_APPLICATION_CREDENTIALS en la ruta del archivo JSON que contiene la clave de tu cuenta de servicio. Esta variable solo se aplica a la sesión de shell actual, por lo que, si abres una nueva sesión, tendrás que volver a definirla.

    Ejemplo: Linux o macOS

    Sustituye [PATH] por la ruta del archivo JSON que contiene la clave de tu cuenta de servicio. 

    export GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

    Por ejemplo:

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

    Ejemplo: Windows

    Sustituye [PATH] por la ruta del archivo JSON que contiene la clave de tu cuenta de servicio y [FILE_NAME] por el nombre del archivo.

    Con PowerShell:

    $env:GOOGLE_APPLICATION_CREDENTIALS="[PATH]"

    Por ejemplo:

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

    Con el símbolo del sistema:

    set GOOGLE_APPLICATION_CREDENTIALS=[PATH]

Definir el flujo de trabajo con el paquete DSL de Kubeflow Pipelines

El paquete kfp.dsl contiene el lenguaje específico del dominio (DSL) que puedes usar para definir e interactuar con las canalizaciones y los componentes.

Los componentes de los flujos de procesamiento de Kubeflow son funciones de fábrica que crean pasos de flujos de procesamiento. En cada componente se describen las entradas, las salidas y la implementación del componente. Por ejemplo, en el código de muestra que se incluye a continuación, ds_op es un componente.

Los componentes se usan para crear pasos de la canalización. Cuando se ejecuta una canalización, los pasos se ejecutan a medida que los datos de los que dependen están disponibles. Por ejemplo, un componente de entrenamiento podría tomar un archivo CSV como entrada y usarlo para entrenar un 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,
    )

Haz los cambios siguientes:

  • PROJECT_ID: el Google Cloud proyecto en el que se ejecuta esta canalización.

  • PIPELINE_ROOT_PATH: especifica un URI de Cloud Storage al que pueda acceder la cuenta de servicio de tu canalización. Los artefactos de tus ejecuciones de canalización se almacenan en la raíz de la canalización. El URI de Cloud Storage debe empezar por gs://.

    La raíz de la canalización se puede definir como un argumento de la anotación @kfp.dsl.pipeline en la función de canalización o se puede definir al llamar a create_run_from_job_spec para crear una ejecución de canalización.

Compilar una canalización en un archivo YAML

Una vez que hayas definido el flujo de trabajo de tu canalización, puedes compilarla en formato YAML. El archivo YAML incluye toda la información necesaria para ejecutar tu canalización en Vertex AI Pipelines.

from kfp import compiler

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

Enviar la ejecución de un flujo de procesamiento

Una vez que el flujo de trabajo de tu canalización se haya compilado en formato YAML, puedes usar el cliente de Python de Vertex AI para enviar y ejecutar tu canalización.

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()

Haz los cambios siguientes:

  • PROJECT_REGION: la región en la que se ejecuta este flujo de procesamiento.

En el ejemplo anterior:

  1. Un flujo de trabajo de Kubeflow se define como una función de Python. La función se anota con el decorador @kfp.dsl.pipeline, que especifica el nombre y la ruta raíz de la canalización. La ruta raíz de la canalización es la ubicación en la que se almacenan los artefactos de la canalización.
  2. Los pasos del flujo de trabajo de la canalización se crean con los Google Cloud componentes de canalización. Al usar las salidas de un componente como entrada de otro, defines el flujo de trabajo de la canalización como un gráfico. Por ejemplo, training_job_run_op depende del dataset resultado de ds_op.
  3. Compila la canalización con kfp.compiler.Compiler.
  4. Crea una ejecución de flujo de procesamiento en Vertex AI Pipelines con el cliente de Python de Vertex AI. Cuando ejecutas una canalización, puedes anular el nombre y la ruta raíz de la canalización. Los pases de flujo de trabajo se pueden agrupar por nombre. Si anulas el nombre de la canalización, podrás distinguir entre las ejecuciones de canalizaciones de producción y experimentales.

Para obtener más información sobre cómo crear pipelines, consulta la sección Crear pipelines de Kubeflow y sigue los ejemplos y los tutoriales.

Probar una canalización de forma local (opcional)

Una vez que hayas definido las canalizaciones y los componentes, podrás probar el código de los componentes ejecutándolo en tu entorno de creación local. Si ejecutas tu flujo de procesamiento o un componente de forma local, puedes identificar y depurar posibles problemas antes de crear una ejecución de flujo de procesamiento en un entorno remoto, como Vertex AI Pipelines. Para obtener más información sobre la ejecución local de canalizaciones y componentes, consulta Ejecución local en la documentación de KFP.

En esta página se explica cómo definir y ejecutar una canalización que consta de dos tareas.

Configurar el entorno local

  1. Opcional: instala Docker.

  2. Usa el siguiente código de ejemplo para definir una canalización sencilla:

    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 una ejecución local

Inicializa una sesión local con la función local.init(). Cuando usas local.init(), el SDK de KFP ejecuta localmente tus pipelines y componentes cuando los llamas.

Cuando usas local.init(), debes especificar un tipo de ejecutor. El tipo de runner indica cómo debe ejecutar KFP cada tarea.

Usa el siguiente ejemplo para especificar el tipo de DockerRunner runner para ejecutar cada tarea en un contenedor. Para obtener más información sobre los ejecutores locales compatibles con KFP, consulta la sección Ejecutores locales de la documentación de KFP.

from kfp import local

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

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

Usa el siguiente código para ver el resultado de la tarea de la canalización al ejecutarla de forma local:

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

Crear flujos de procesamiento de Kubeflow

Sigue este proceso para crear una canalización.

  1. Diseña tu flujo de procesamiento como una serie de componentes. Para fomentar la reutilización, cada componente debe tener una única responsabilidad. Siempre que sea posible, diseña tu flujo de procesamiento para reutilizar componentes probados, como los Google Cloud componentes de flujo de procesamiento.

    Más información sobre cómo diseñar flujos de trabajo

  2. Crea los componentes personalizados que necesites para implementar tu flujo de trabajo de aprendizaje automático con el SDK de Kubeflow Pipelines. Los componentes son conjuntos de código independientes que realizan un paso en tu flujo de trabajo de aprendizaje automático. Usa las siguientes opciones para crear los componentes de la canalización.

  3. Crea tu flujo de procesamiento como una función de Python.

    Consulta más información sobre cómo definir tu flujo de trabajo como una función de Python.

  4. Usa el compilador del SDK de Kubeflow Pipelines para compilar tu flujo de trabajo.

    from kfp import compiler

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

    Haz los cambios siguientes:

    • PIPELINE_FUNCTION: el nombre de la función de tu flujo de procesamiento.
    • PIPELINE_PACKAGE_PATH: la ruta a la ubicación donde se almacenará la canalización compilada.

  5. Ejecuta tu flujo de procesamiento con la consola Google Cloud o Python.

Acceder a Google Cloud recursos en una canalización

Si no especificas una cuenta de servicio al ejecutar un flujo de procesamiento, Vertex AI Pipelines usará la cuenta de servicio predeterminada de Compute Engine para ejecutarlo. Vertex AI Pipelines también usa la cuenta de servicio de una ejecución de flujo de procesamiento para autorizar el acceso de tu flujo de procesamiento a losGoogle Cloud recursos. La cuenta de servicio predeterminada de Compute Engine tiene el rol Editor del proyecto de forma predeterminada. Esto puede dar a tus pipelines un acceso excesivo a los recursos de Google Cloud en tu proyecto de Google Cloud .

Te recomendamos que crees una cuenta de servicio para ejecutar tus flujos de procesamiento y, a continuación, concedas a esta cuenta permisos específicos para los Google Cloud recursos que necesites para ejecutar tu flujo de procesamiento.

Consulta más información sobre cómo usar Gestión de Identidades y Accesos para crear una cuenta de servicio y gestionar el acceso concedido a una cuenta de servicio.

Mantener las canalizaciones actualizadas

Los clientes del SDK y las imágenes de contenedor que usas para compilar y ejecutar las canalizaciones se actualizan periódicamente a nuevas versiones para corregir vulnerabilidades de seguridad y añadir nuevas funciones. Para que tus pipelines estén actualizados a la versión más reciente, te recomendamos que hagas lo siguiente:

Siguientes pasos