Introducción a Vertex Pipelines

1. Descripción general

En este lab, aprenderás a crear y ejecutar canalizaciones de AA con Vertex Pipelines.

Qué aprenderá

Aprenderás a hacer lo siguiente:

• Aprenderás a usar el SDK de canalizaciones de Kubeflow para compilar canalizaciones de AA escalables.
• Aprenderás a crear y ejecutar una canalización de introducción en 3 pasos que admite entradas de texto.
• Aprenderás a crear y ejecutar una canalización que entrene, evalúe y, luego, implemente un modelo de clasificación de AutoML.
• Usa componentes compilados previamente para interactuar con los servicios de Vertex AI que se proporcionan a través de la biblioteca google_cloud_pipeline_components.
• Aprenderás a programar un trabajo de canalización con Cloud Scheduler.

El costo total de la ejecución de este lab en Google Cloud es de aproximadamente $25.

2. Introducción a Vertex AI

En este lab, se utiliza la oferta de productos de IA más reciente de Google Cloud. Vertex AI integra las ofertas de AA de Google Cloud en una experiencia de desarrollo fluida. Anteriormente, se podía acceder a los modelos personalizados y a los entrenados con AutoML mediante servicios independientes. La nueva oferta combina ambos en una sola API, junto con otros productos nuevos. También puedes migrar proyectos existentes a Vertex AI.

Además del entrenamiento de modelos y los servicios de implementación, Vertex AI también incluye una variedad de productos de MLOps, incluidos Vertex Pipelines (el producto en el que se enfoca este lab), Model Monitoring, Feature Store y mucho más. Puedes ver todas las ofertas de productos de Vertex AI en el diagrama que se muestra a continuación.

Si tienes comentarios, consulta la página de asistencia.

¿Por qué son útiles las canalizaciones de AA?

Antes de comenzar, primero debes comprender por qué deberías usar canalizaciones. Imagina que estás creando un flujo de trabajo de AA que incluye procesar datos, entrenar un modelo, ajustar hiperparámetros y realizar evaluaciones, así como implementar modelos. Es posible que cada uno de estos pasos tenga dependencias diferentes, lo que podría ser difícil de manejar si tratas todo el flujo de trabajo como una aplicación monolítica. A medida que empiezas a escalar tu proceso de AA, es posible que quieras compartir el flujo de trabajo de AA con otras personas de tu equipo para que puedan ejecutarlo y agregar más código. Pero esto puede ser complicado sin un proceso confiable y reproducible. Con las canalizaciones, cada paso en tu proceso de AA tiene su propio contenedor. Así, podrás desarrollar pasos de forma independiente y hacer un seguimiento de la entrada y salida en cada paso de manera reproducible. Además, puedes programar o activar ejecuciones para tu canalización en función de otros eventos de tu entorno de Cloud, como iniciar la ejecución de una canalización cuando hay nuevos datos de entrenamiento disponibles.

Resumen: Las canalizaciones te permiten automatizar y reproducir tu flujo de trabajo de AA.

3. Configura el entorno de Cloud

Para ejecutar este codelab, necesitarás un proyecto de Google Cloud Platform que tenga habilitada la facturación. Para crear un proyecto, sigue estas instrucciones.

Paso 1: Inicia Cloud Shell

En este lab, trabajarás con una sesión de Cloud Shell, que es un intérprete de comandos alojado en una máquina virtual que se ejecuta en la nube de Google. Podrías ejecutar fácilmente esta sección de forma local, en tu computadora, pero Cloud Shell brinda una experiencia reproducible en un entorno coherente para todo el mundo. Después de este lab, puedes volver a probar esta sección en tu computadora.

La primera vez que hagas una solicitud en Cloud Shell que requiera autorización, verás el mensaje “Autoriza Cloud Shell”. como el que se muestra a continuación, en el que puedes hacer clic con seguridad en el botón "Autorizar" .

Activar Cloud Shell

En la parte superior derecha de la consola de Cloud, haz clic en el siguiente botón para activar Cloud Shell:

Si nunca iniciaste Cloud Shell, aparecerá una pantalla intermedia (mitad inferior de la página) que describe de qué se trata. Si ese es el caso, haz clic en Continuar (y no volverás a verlo). Así es como se ve la pantalla única:

El aprovisionamiento y la conexión a Cloud Shell solo tomará unos minutos.

Esta máquina virtual está cargada con todas las herramientas de desarrollo que necesitas. Ofrece un directorio principal persistente de 5 GB y se ejecuta en Google Cloud, lo que permite mejorar considerablemente el rendimiento de la red y la autenticación. Gran parte de tu trabajo en este codelab, si no todo, se puede hacer simplemente con un navegador o tu Chromebook.

Una vez conectado a Cloud Shell, debería ver que ya se autenticó y que el proyecto ya se configuró con tu ID del proyecto.

En Cloud Shell, ejecuta el siguiente comando para confirmar que tienes la autenticación:

gcloud auth list

Deberías ver algo como esto en el resultado del comando:

Nota: gcloud es la herramienta de línea de comandos potente y unificada de Google Cloud. Ya está preinstalada en Cloud Shell. Notarás que esta función es compatible con la función de autocompletado con tabulador. Para obtener más información, consulta la descripción general de la herramienta de línea de comandos de gcloud.

Ejecuta el siguiente comando en Cloud Shell para confirmar que el comando de gcloud conoce tu proyecto:

gcloud config list project

Resultado del comando

[core]
project = <PROJECT_ID>

De lo contrario, puedes configurarlo con el siguiente comando:

gcloud config set project <PROJECT_ID>

Resultado del comando

Updated property [core/project].

Cloud Shell cuenta con algunas variables de entorno, incluida GOOGLE_CLOUD_PROJECT, que contiene el nombre de nuestro proyecto de Cloud actual. Lo usaremos en varios lugares a lo largo de este lab. Puede ejecutar el siguiente código para verla:

echo $GOOGLE_CLOUD_PROJECT

Paso 2: Habilitar las API

Más adelante, verás en qué momento se necesitan estos servicios y por qué. Por ahora, ejecuta este comando para que tu proyecto pueda acceder a los servicios de Compute Engine, Container Registry y Vertex AI:

gcloud services enable compute.googleapis.com         \
                       containerregistry.googleapis.com  \
                       aiplatform.googleapis.com  \
                       cloudbuild.googleapis.com \
                       cloudfunctions.googleapis.com

Si se realizó correctamente, se mostrará un mensaje similar a este:

Operation "operations/acf.cc11852d-40af-47ad-9d59-477a12847c9e" finished successfully.

Paso 3: Crea un bucket de Cloud Storage

Para ejecutar un trabajo de entrenamiento en Vertex AI, necesitaremos un bucket de almacenamiento para almacenar los elementos del modelo guardados. El bucket debe ser regional. Usaremos us-central aquí, pero puedes usar otra región (solo debes reemplazarla durante el lab). Si ya tiene un bucket, puede omitir este paso.

Ejecuta los siguientes comandos en la terminal de Cloud Shell para crear un bucket:

BUCKET_NAME=gs://$GOOGLE_CLOUD_PROJECT-bucket
gsutil mb -l us-central1 $BUCKET_NAME

A continuación, le daremos a la cuenta de servicio de Compute acceso a este bucket. Esto garantizará que Vertex Pipelines tenga los permisos necesarios para escribir archivos en el bucket. Ejecute el siguiente comando para agregar este permiso:

gcloud projects describe $GOOGLE_CLOUD_PROJECT > project-info.txt
PROJECT_NUM=$(cat project-info.txt | sed -nre 's:.*projectNumber\: (.*):\1:p')
SVC_ACCOUNT="${PROJECT_NUM//\'/}[e-mail]"
gcloud projects add-iam-policy-binding $GOOGLE_CLOUD_PROJECT --member serviceAccount:$SVC_ACCOUNT --role roles/storage.objectAdmin

Paso 4: Crea una instancia de Vertex AI Workbench

En la sección Vertex AI de Cloud Console, haz clic en Workbench:

Allí, en Notebooks administrados por el usuario, haz clic en Nuevo notebook:

Luego, selecciona el tipo de instancia TensorFlow Enterprise 2.3 (with LTS) without GPUs:

Usa las opciones predeterminadas y, luego, haz clic en Crear.

Paso 5: Abre tu notebook

Una vez que se crea la instancia, selecciona Abrir JupyterLab:

4. Configuración de Vertex Pipelines

Existen algunas bibliotecas adicionales que debemos instalar para usar Vertex Pipelines:

• Kubeflow Pipelines: Este es el SDK que usaremos para compilar nuestra canalización. Vertex Pipelines admite canalizaciones en ejecución compiladas con Kubeflow Pipelines o TFX.
• Componentes de canalizaciones de Google Cloud: Esta biblioteca proporciona componentes predefinidos que facilitan la interacción con los servicios de Vertex AI durante los pasos de tu canalización.

Paso 1: Crea un notebook de Python y, luego, instala bibliotecas

Primero, en el menú Selector de tu instancia de notebook, selecciona Python 3 para crear un notebook:

Para acceder al menú Selector, haz clic en Acceder + en la esquina superior izquierda de tu instancia de notebook.

Para instalar los dos servicios que usaremos en este lab, primero hay que establecer la marca de usuario en una celda del notebook:

USER_FLAG = "--user"

Luego, ejecuta el siguiente comando en tu notebook:

!pip3 install {USER_FLAG} google-cloud-aiplatform==1.7.0 --upgrade
!pip3 install {USER_FLAG} kfp==1.8.9 google-cloud-pipeline-components==0.2.0

Luego de instalar estos paquetes, deberá reiniciar el kernel:

import os

if not os.getenv("IS_TESTING"):
    # Automatically restart kernel after installs
    import IPython

    app = IPython.Application.instance()
    app.kernel.do_shutdown(True)

Finalmente, verifique que instaló los paquetes de forma correcta. La versión del SDK de KFP debe ser igual o superior a 1.8:

!python3 -c "import kfp; print('KFP SDK version: {}'.format(kfp.__version__))"
!python3 -c "import google_cloud_pipeline_components; print('google_cloud_pipeline_components version: {}'.format(google_cloud_pipeline_components.__version__))"

Paso 2: Configura tu ID del proyecto y bucket

Durante este lab, podrá hacer referencia al ID del proyecto de Cloud y al bucket que creó anteriormente. A continuación, crearemos variables para cada uno de ellos.

Si desconoce el ID de su proyecto, probablemente logre obtenerlo mediante la ejecución del siguiente comando:

import os
PROJECT_ID = ""

# Get your Google Cloud project ID from gcloud
if not os.getenv("IS_TESTING"):
    shell_output=!gcloud config list --format 'value(core.project)' 2>/dev/null
    PROJECT_ID = shell_output[0]
    print("Project ID: ", PROJECT_ID)

De lo contrario, configúralo aquí:

if PROJECT_ID == "" or PROJECT_ID is None:
    PROJECT_ID = "your-project-id"  # @param {type:"string"}

Luego, cree una variable para almacenar el nombre de su bucket. Si lo creó en este lab, lo siguiente funcionará. De lo contrario, deberá configurarlo de forma manual:

BUCKET_NAME="gs://" + PROJECT_ID + "-bucket"

Paso 3: Importa bibliotecas

Agrega lo siguiente para importar las bibliotecas que usaremos durante este codelab:

import kfp

from kfp.v2 import compiler, dsl
from kfp.v2.dsl import component, pipeline, Artifact, ClassificationMetrics, Input, Output, Model, Metrics

from google.cloud import aiplatform
from google_cloud_pipeline_components import aiplatform as gcc_aip
from typing import NamedTuple

Paso 4: Define las constantes

Lo último que debemos hacer antes de crear nuestra canalización es definir algunas variables constantes. PIPELINE_ROOT es la ruta de acceso de Cloud Storage en la que se escribirán los artefactos que cree nuestra canalización. Aquí estamos usando us-central1 como la región, pero si usaste una región diferente cuando creaste tu bucket, actualiza la variable REGION en el siguiente código:

PATH=%env PATH
%env PATH={PATH}:/home/jupyter/.local/bin
REGION="us-central1"

PIPELINE_ROOT = f"{BUCKET_NAME}/pipeline_root/"
PIPELINE_ROOT

Luego de ejecutar el código anterior, debería ver impreso el directorio raíz de su canalización. Esta es la ubicación de Cloud Storage en la que se escribirán los artefactos de su canalización, y tendrá el siguiente formato: gs://YOUR-BUCKET-NAME/pipeline_root/.

5. Cree su primera canalización

Para familiarizarnos con el funcionamiento de Vertex Pipelines, primero crearemos una canalización corta con el SDK de KFP. Esta canalización no ejecuta ninguna acción relacionada con el AA (no se preocupe, ya llegaremos a ese paso). Mediante este ejercicio, buscamos que aprenda los siguientes procesos:

• Cómo crear componentes personalizados en el SDK de KFP
• Cómo ejecutar y supervisar una canalización en Vertex Pipelines

Crearemos una canalización que imprima una oración mediante dos salidas: el nombre de un producto y una descripción en forma de emoji. Esta canalización consistirá en tres componentes:

• product_name: Este componente tomará un nombre de producto (o cualquier sustantivo que desees) como entrada y mostrará esa cadena como salida.
• emoji: Este componente tomará la descripción de texto de un emoji y lo convertirá en un emoji. Por ejemplo, el código de texto para ✨ es “sparkles”. Este componente usa una biblioteca de emojis para mostrarte cómo administrar dependencias externas en tu canalización
• build_sentence: Este componente final consumirá el resultado de los dos anteriores para crear una oración que use el emoji. Por ejemplo, la salida sería “Vertex Pipelines es ✨”.

Comencemos a programar.

Paso 1: Crea un componente basado en una función de Python

Con el SDK de KFP, podemos crear componentes basados en las funciones de Python. Lo usaremos para los 3 componentes de nuestra canalización. Primero, compilaremos el componente product_name, que simplemente toma una cadena como entrada y la muestra. Agregue el siguiente comando a su notebook:

@component(base_image="python:3.9", output_component_file="first-component.yaml")
def product_name(text: str) -> str:
    return text

Analicemos con mayor detalle esta sintaxis:

• El decorador @component compila esta función en un componente cuando se ejecuta la canalización. Lo utilizarás cada vez que escribas un componente personalizado.
• El parámetro base_image especifica la imagen de contenedor que usará este componente.
• El parámetro output_component_file es opcional y especifica el archivo yaml en el que se escribirá el componente compilado. Luego de ejecutar la celda, deberías ver que ese archivo se escribió en tu instancia de notebook. Si quieres compartir este componente con otra persona, puedes enviar el archivo yaml generado para que lo cargue con el siguiente comando:

product_name_component = kfp.components.load_component_from_file('./first-component.yaml')

• El -> str que aparece después de la definición de la función especifica el tipo de salida de este componente.

Paso 2: Crea dos componentes adicionales

Crearemos dos componentes más para completar nuestra canalización. El primero toma una string como entrada y la convierte en su emoji correspondiente, si es que tiene alguno. De esta forma, devolverá una tupla con el texto de entrada proporcionado, y el emoji resultante:

@component(packages_to_install=["emoji"])
def emoji(
    text: str,
) -> NamedTuple(
    "Outputs",
    [
        ("emoji_text", str),  # Return parameters
        ("emoji", str),
    ],
):
    import emoji

    emoji_text = text
    emoji_str = emoji.emojize(':' + emoji_text + ':', language='alias')
    print("output one: {}; output_two: {}".format(emoji_text, emoji_str))
    return (emoji_text, emoji_str)

Este componente es un poco más complejo que el anterior. Desglosémoslo de nuevo:

• El parámetro packages_to_install le indica al componente si hay dependencias de bibliotecas externas para este contenedor. En este caso, usaremos una biblioteca llamada emoji.
• Este componente muestra un NamedTuple llamado Outputs. Observa que cada cadena de esta tupla tiene claves: emoji_text y emoji. Utilizaremos estas claves en el próximo componente para acceder al resultado.

El componente final de esta canalización consumirá el resultado de las dos primeras y los combinará para mostrar una string:

@component
def build_sentence(
    product: str,
    emoji: str,
    emojitext: str
) -> str:
    print("We completed the pipeline, hooray!")
    end_str = product + " is "
    if len(emoji) > 0:
        end_str += emoji
    else:
        end_str += emojitext
    return(end_str)

Quizás te estés preguntando, ¿cómo sabe este componente que debe utilizar el resultado de los pasos anteriores que definió? Buena pregunta. Veremos esto en el próximo paso.

Paso 3: Junta los componentes en una canalización

Las definiciones indicadas de los componentes crearon funciones de fábrica que se pueden utilizar en una definición de canalización para crear pasos. Para configurar una canalización, usa el decorador @pipeline, asígnale un nombre y una descripción a la canalización, y proporciona la ruta raíz en la que se deben escribir los artefactos de tu canalización. Con los artefactos, hacemos referencia a los archivos de salida que genera su canalización. Si bien esta canalización de introducción no genera ninguno, la siguiente lo hará.

En el siguiente bloque de código, definimos una función intro_pipeline. Aquí es donde especificamos las entradas a nuestros pasos iniciales de la canalización y cómo los pasos se conectan entre sí:

• product_task toma un nombre de producto como entrada. Aquí transferimos “Vertex Pipelines” pero puedes cambiarlo según tus necesidades.
• emoji_task toma el código de texto de un emoji como entrada. También puedes cambiarlo según tus necesidades. Por ejemplo, “party_face” hace referencia al emoji 🥳. Ten en cuenta que, como ni este ni el componente product_task tienen pasos que les otorguen entradas, debemos especificarlas de forma manual cuando definimos nuestra canalización.
• El último paso de nuestra canalización, consumer_task, tiene tres parámetros de entrada:
  • El resultado de product_task Dado que en este paso solo se produce un resultado, podemos hacer referencia a él a través de product_task.output.
  • El resultado emoji de nuestro paso emoji_task. Consulta el componente emoji definido anteriormente en el que nombramos los parámetros de salida.
  • De manera similar, el resultado con nombre emoji_text del componente emoji. En caso de que a nuestra canalización se le transfiera texto que no corresponde con el emoji, lo utilizará para construir una oración.

@pipeline(
    name="hello-world",
    description="An intro pipeline",
    pipeline_root=PIPELINE_ROOT,
)

# You can change the `text` and `emoji_str` parameters here to update the pipeline output
def intro_pipeline(text: str = "Vertex Pipelines", emoji_str: str = "sparkles"):
    product_task = product_name(text)
    emoji_task = emoji(emoji_str)
    consumer_task = build_sentence(
        product_task.output,
        emoji_task.outputs["emoji"],
        emoji_task.outputs["emoji_text"],
    )

Paso 4: Compila y ejecuta la canalización

Luego de que definas la canalización, la podrás compilar. El siguiente comando generará un archivo JSON que utilizará para ejecutarla:

compiler.Compiler().compile(
    pipeline_func=intro_pipeline, package_path="intro_pipeline_job.json"
)

Luego, crea una variable TIMESTAMP. Lo usaremos en nuestro ID de trabajo:

from datetime import datetime

TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")

Luego, define tu trabajo de canalización:

job = aiplatform.PipelineJob(
    display_name="hello-world-pipeline",
    template_path="intro_pipeline_job.json",
    job_id="hello-world-pipeline-{0}".format(TIMESTAMP),
    enable_caching=True
)

Por último, ejecuta el trabajo para crear una ejecución de canalización nueva:

job.submit()

Después de ejecutar esta celda, deberías ver registros con un vínculo para ver la ejecución de la canalización en tu consola:

Navega a ese vínculo. Tu canalización debería verse así cuando se complete:

Esta canalización demorará de 5 a 6 minutos en ejecutarse. Cuando se complete, puedes hacer clic en el componente build-sentence para ver el resultado final:

Ahora que sabes cómo funcionan el SDK de KFP y Vertex Pipelines, ya puedes compilar una canalización que cree y, luego, implemente un modelo de AA con otros servicios de Vertex AI. ¡Comencemos!

6. Crea una canalización de AA de extremo a extremo

Llegó la hora de compilar su primera canalización de AA. En esta canalización, usaremos el conjunto de datos de frijoles secos de UCI Machine Learning de KOKLU, M. y OZKAN, I.A., (2020), "Multiclass Classification of Dry Beans Using Computer Vision and Machine Learning Techniques.""En Computers and Electronics in Agriculture, 174, 105507. DOI.

Este es un conjunto de datos tabular que, en nuestra canalización, lo usaremos para entrenar, evaluar y, luego, implementar un modelo de AutoML que clasifique frijoles en uno de 7 tipos, según sus características.

Con esta canalización podrás realizar las siguientes acciones:

• Crea un conjunto de datos en
• Entrena un modelo de clasificación tabular con AutoML
• Obtener métricas de evaluación sobre este modelo
• Según las métricas de evaluación, decidir si quieres implementar el modelo mediante una lógica condicional en Vertex Pipelines
• Implementar el modelo en un extremo con Vertex Prediction

Cada paso descrito será un componente. La mayoría de los pasos de la canalización usarán componentes predefinidos para los servicios de Vertex AI a través de la biblioteca google_cloud_pipeline_components que importamos antes en este codelab. En esta sección, primero definiremos un componente personalizado. Luego, definiremos el resto de los pasos de la canalización mediante componentes predefinidos. Estos componentes facilitan el acceso a los servicios de Vertex AI, como el entrenamiento y la implementación de modelos.

En este paso, la mayor parte del tiempo se dedicará a la pieza de entrenamiento de AutoML de esta canalización, que demorará aproximadamente una hora.

Paso 1: Un componente personalizado para la evaluación de modelos

El componente personalizado que definiremos se usará al final de nuestra canalización, una vez que se complete el entrenamiento del modelo. Este componente realizará algunas acciones:

• Obtener las métricas de evaluación del modelo de clasificación entrenado de AutoML
• Analizar las métricas y renderizarlas en la IU de Vertex Pipelines
• Comparar las métricas con un umbral para determinar si se debe implementar el modelo

Antes de definir el componente, conozcamos sus parámetros de entrada y salida. Como entrada, esta canalización toma algunos metadatos sobre nuestro proyecto de Cloud, el modelo entrenado resultante (más adelante definiremos este componente), las métricas de evaluación del modelo y un thresholds_dict_str. El thresholds_dict_str es algo que definiremos cuando ejecutemos nuestra canalización. En el caso de este modelo de clasificación, será el valor del área bajo la curva ROC para el que deberíamos implementar el modelo. Por ejemplo, si pasamos 0.95, significa que solo queremos que nuestra canalización implemente el modelo si esta métrica es superior al 95%.

Nuestro componente de evaluación devuelve una cadena que indica si implementar o no el modelo. Agrega el siguiente comando a una celda de notebook para crear este componente personalizado:

@component(
    base_image="gcr.io/deeplearning-platform-release/tf2-cpu.2-3:latest",
    output_component_file="tabular_eval_component.yaml",
    packages_to_install=["google-cloud-aiplatform"],
)
def classification_model_eval_metrics(
    project: str,
    location: str,  # "us-central1",
    api_endpoint: str,  # "us-central1-aiplatform.googleapis.com",
    thresholds_dict_str: str,
    model: Input[Artifact],
    metrics: Output[Metrics],
    metricsc: Output[ClassificationMetrics],
) -> NamedTuple("Outputs", [("dep_decision", str)]):  # Return parameter.

    import json
    import logging

    from google.cloud import aiplatform as aip

    # Fetch model eval info
    def get_eval_info(client, model_name):
        from google.protobuf.json_format import MessageToDict

        response = client.list_model_evaluations(parent=model_name)
        metrics_list = []
        metrics_string_list = []
        for evaluation in response:
            print("model_evaluation")
            print(" name:", evaluation.name)
            print(" metrics_schema_uri:", evaluation.metrics_schema_uri)
            metrics = MessageToDict(evaluation._pb.metrics)
            for metric in metrics.keys():
                logging.info("metric: %s, value: %s", metric, metrics[metric])
            metrics_str = json.dumps(metrics)
            metrics_list.append(metrics)
            metrics_string_list.append(metrics_str)

        return (
            evaluation.name,
            metrics_list,
            metrics_string_list,
        )

    # Use the given metrics threshold(s) to determine whether the model is
    # accurate enough to deploy.
    def classification_thresholds_check(metrics_dict, thresholds_dict):
        for k, v in thresholds_dict.items():
            logging.info("k {}, v {}".format(k, v))
            if k in ["auRoc", "auPrc"]:  # higher is better
                if metrics_dict[k] < v:  # if under threshold, don't deploy
                    logging.info("{} < {}; returning False".format(metrics_dict[k], v))
                    return False
        logging.info("threshold checks passed.")
        return True

    def log_metrics(metrics_list, metricsc):
        test_confusion_matrix = metrics_list[0]["confusionMatrix"]
        logging.info("rows: %s", test_confusion_matrix["rows"])

        # log the ROC curve
        fpr = []
        tpr = []
        thresholds = []
        for item in metrics_list[0]["confidenceMetrics"]:
            fpr.append(item.get("falsePositiveRate", 0.0))
            tpr.append(item.get("recall", 0.0))
            thresholds.append(item.get("confidenceThreshold", 0.0))
        print(f"fpr: {fpr}")
        print(f"tpr: {tpr}")
        print(f"thresholds: {thresholds}")
        metricsc.log_roc_curve(fpr, tpr, thresholds)

        # log the confusion matrix
        annotations = []
        for item in test_confusion_matrix["annotationSpecs"]:
            annotations.append(item["displayName"])
        logging.info("confusion matrix annotations: %s", annotations)
        metricsc.log_confusion_matrix(
            annotations,
            test_confusion_matrix["rows"],
        )

        # log textual metrics info as well
        for metric in metrics_list[0].keys():
            if metric != "confidenceMetrics":
                val_string = json.dumps(metrics_list[0][metric])
                metrics.log_metric(metric, val_string)
        # metrics.metadata["model_type"] = "AutoML Tabular classification"

    logging.getLogger().setLevel(logging.INFO)
    aip.init(project=project)
    # extract the model resource name from the input Model Artifact
    model_resource_path = model.metadata["resourceName"]
    logging.info("model path: %s", model_resource_path)

    client_options = {"api_endpoint": api_endpoint}
    # Initialize client that will be used to create and send requests.
    client = aip.gapic.ModelServiceClient(client_options=client_options)
    eval_name, metrics_list, metrics_str_list = get_eval_info(
        client, model_resource_path
    )
    logging.info("got evaluation name: %s", eval_name)
    logging.info("got metrics list: %s", metrics_list)
    log_metrics(metrics_list, metricsc)

    thresholds_dict = json.loads(thresholds_dict_str)
    deploy = classification_thresholds_check(metrics_list[0], thresholds_dict)
    if deploy:
        dep_decision = "true"
    else:
        dep_decision = "false"
    logging.info("deployment decision is %s", dep_decision)

    return (dep_decision,)

Paso 2: Agrega componentes predefinidos de Google Cloud

En este paso, definiremos el resto de los componentes de nuestra canalización y veremos cómo funcionan todos juntos. Primero, define el nombre visible de la ejecución de tu canalización con una marca de tiempo:

import time
DISPLAY_NAME = 'automl-beans{}'.format(str(int(time.time())))
print(DISPLAY_NAME)

Luego, copia el siguiente comando en una celda de notebook nueva:

@pipeline(name="automl-tab-beans-training-v2",
                  pipeline_root=PIPELINE_ROOT)
def pipeline(
    bq_source: str = "bq://aju-dev-demos.beans.beans1",
    display_name: str = DISPLAY_NAME,
    project: str = PROJECT_ID,
    gcp_region: str = "us-central1",
    api_endpoint: str = "us-central1-aiplatform.googleapis.com",
    thresholds_dict_str: str = '{"auRoc": 0.95}',
):
    dataset_create_op = gcc_aip.TabularDatasetCreateOp(
        project=project, display_name=display_name, bq_source=bq_source
    )

    training_op = gcc_aip.AutoMLTabularTrainingJobRunOp(
        project=project,
        display_name=display_name,
        optimization_prediction_type="classification",
        budget_milli_node_hours=1000,
        column_transformations=[
            {"numeric": {"column_name": "Area"}},
            {"numeric": {"column_name": "Perimeter"}},
            {"numeric": {"column_name": "MajorAxisLength"}},
            {"numeric": {"column_name": "MinorAxisLength"}},
            {"numeric": {"column_name": "AspectRation"}},
            {"numeric": {"column_name": "Eccentricity"}},
            {"numeric": {"column_name": "ConvexArea"}},
            {"numeric": {"column_name": "EquivDiameter"}},
            {"numeric": {"column_name": "Extent"}},
            {"numeric": {"column_name": "Solidity"}},
            {"numeric": {"column_name": "roundness"}},
            {"numeric": {"column_name": "Compactness"}},
            {"numeric": {"column_name": "ShapeFactor1"}},
            {"numeric": {"column_name": "ShapeFactor2"}},
            {"numeric": {"column_name": "ShapeFactor3"}},
            {"numeric": {"column_name": "ShapeFactor4"}},
            {"categorical": {"column_name": "Class"}},
        ],
        dataset=dataset_create_op.outputs["dataset"],
        target_column="Class",
    )
    model_eval_task = classification_model_eval_metrics(
        project,
        gcp_region,
        api_endpoint,
        thresholds_dict_str,
        training_op.outputs["model"],
    )

    with dsl.Condition(
        model_eval_task.outputs["dep_decision"] == "true",
        name="deploy_decision",
    ):

        endpoint_op = gcc_aip.EndpointCreateOp(
            project=project,
            location=gcp_region,
            display_name="train-automl-beans",
        )

        gcc_aip.ModelDeployOp(
            model=training_op.outputs["model"],
            endpoint=endpoint_op.outputs["endpoint"],
            dedicated_resources_min_replica_count=1,
            dedicated_resources_max_replica_count=1,
            dedicated_resources_machine_type="n1-standard-4",
        )

Veamos qué sucede en este código:

• Primero, al igual que en nuestra canalización anterior, definimos los parámetros de entrada que utiliza esta canalización. Debemos establecerlos de forma manual, ya que no dependen del resultado de otros pasos.
• El resto de la canalización usa algunos componentes compilados previamente para interactuar con los servicios de Vertex AI:
  • TabularDatasetCreateOp crea un conjunto de datos tabular en Vertex AI a partir de una fuente de conjunto de datos en Cloud Storage o BigQuery. En esta canalización, pasamos los datos a través de una URL de tabla de BigQuery.
  • AutoMLTabularTrainingJobRunOp inicia un trabajo de entrenamiento de AutoML para un conjunto de datos tabular. Pasamos algunos parámetros de configuración a este componente, incluidos el tipo de modelo (en este caso, la clasificación), algunos datos en las columnas, por cuánto tiempo nos gustaría ejecutar el entrenamiento y un puntero para el conjunto de datos. Ten en cuenta que, para pasar el conjunto de datos a este componente, proporcionamos el resultado del componente anterior mediante dataset_create_op.outputs["dataset"].
  • EndpointCreateOp crea un extremo en Vertex AI. El extremo creado en este paso se pasará como entrada al siguiente componente
  • ModelDeployOp implementa un modelo determinado en un extremo en Vertex AI. En este caso, usamos el extremo creado en el paso anterior. Aunque existen opciones de configuración adicionales, aquí proporcionamos el modelo y el tipo de máquina del extremo que queremos implementar. Pasamos el modelo mediante el acceso a los resultados del paso de entrenamiento en nuestra canalización.
• Esta canalización también hace uso de una lógica condicional, una función de Vertex Pipelines que te permite definir una condición, junto con diferentes ramas basadas en el resultado de esa condición. Recuerda que cuando definimos nuestra canalización, pasamos un parámetro thresholds_dict_str. Es el umbral de precisión que usamos para determinar si debemos implementar el modelo en un extremo. Para implementar esto, usamos la clase Condition del SDK de KFP. La condición que pasamos es el resultado del componente de evaluación personalizado que definimos antes en este codelab. Si esta condición es verdadera, la canalización seguirá ejecutando el componente deploy_op. Si la precisión no cumple con el umbral predefinido, se detendrá la canalización y no se implementará el modelo.

Paso 3: Compila y ejecuta la canalización de AA de extremo a extremo

Después de definir la canalización completa, es momento de compilarla:

compiler.Compiler().compile(
    pipeline_func=pipeline, package_path="tab_classif_pipeline.json"
)

Luego, define el trabajo:

ml_pipeline_job = aiplatform.PipelineJob(
    display_name="automl-tab-beans-training",
    template_path="tab_classif_pipeline.json",
    pipeline_root=PIPELINE_ROOT,
    parameter_values={"project": PROJECT_ID, "display_name": DISPLAY_NAME},
    enable_caching=True
)

Por último, ejecuta el trabajo:

ml_pipeline_job.submit()

Navega al vínculo que se muestra en los registros después de ejecutar la celda anterior para ver tu canalización en la consola. Esta canalización demorará alrededor de una hora en ejecutarse. La mayor parte del tiempo se dedica a la etapa de entrenamiento de AutoML. La canalización completa se verá de forma similar a esta:

Si presionas el botón de activación “Expand artifacts” que aparece en la parte superior, podrás ver los detalles de los diferentes artefactos que creó tu canalización. Por ejemplo, si haces clic en el artefacto dataset, verás detalles en el conjunto de datos de Vertex AI que se creó. Puedes hacer clic en este vínculo para ir a la página de ese conjunto de datos:

De manera similar, para conocer las visualizaciones de las métricas resultantes de nuestro componente de evaluación personalizado, haz clic en el artefacto llamado metricsc. En el lado derecho del panel, podrás ver la matriz de confusión de este modelo:

Para ver el modelo y el extremo creados a partir de la ejecución de esta canalización, ve a la sección de modelos y haz clic en el modelo llamado automl-beans. Ahí encontrarás este modelo implementado en un extremo:

También puede acceder a esta página si hace clic en el artefacto endpoint del gráfico de su canalización.

Además de mirar el gráfico de la canalización en la consola, también puedes utilizar Vertex Pipelines para hacer un seguimiento del linaje. Cuando hablamos de seguimiento del linaje, nos referimos a hacer un seguimiento de los artefactos creados durante la canalización. Esto puede ayudarnos a comprender dónde se crearon los artefactos y cómo se usan en un flujo de trabajo de AA. Por ejemplo, para ver el seguimiento del linaje del conjunto de datos que se creó en esta canalización, haga clic en el artefacto del conjunto de datos y, luego, en Ver linaje:

Aquí se muestran todos los lugares en los que se está usando este artefacto:

Paso 4: Compare las métricas en las ejecuciones de las canalizaciones

Si ejecutas esta canalización varias veces, es probable que quieras comparar las métricas de las ejecuciones. Puedes usar el método aiplatform.get_pipeline_df() para acceder a los metadatos de la ejecución. En este paso, obtendrás los metadatos de todas las ejecuciones de esta canalización y los cargarás al DataFrame de Pandas:

pipeline_df = aiplatform.get_pipeline_df(pipeline="automl-tab-beans-training-v2")
small_pipeline_df = pipeline_df.head(2)
small_pipeline_df

Con eso, terminaste el lab.

🎉 ¡Felicitaciones! 🎉

Aprendiste a usar Vertex AI para hacer lo siguiente:

• Usar el SDK de Kubeflow Pipelines para compilar canalizaciones con componentes personalizados
• Ejecuta tus canalizaciones en Vertex Pipelines y, luego, inicia las ejecuciones de estas con el SDK
• Visualiza y analiza tu grafo de Vertex Pipelines en la consola
• Usar componentes de canalizaciones compilados previamente para agregar servicios de Vertex AI a tu canalización
• Programar trabajos de canalización recurrentes

Para obtener más información sobre las distintas partes de Vertex, consulte la documentación.

7. Limpieza

Para que no se te cobre, se recomienda borrar los recursos que creaste durante este lab.

Paso 1: Detén o borra la instancia de Notebooks

Si quieres continuar usando el notebook que creaste en este lab, te recomendamos que lo desactives cuando no lo utilices. En la IU de Notebooks de la consola de Cloud, selecciona el notebook y, luego, haz clic en Detener. Si quieres borrar la instancia por completo, selecciona Borrar (Delete):

Paso 2: Borra el extremo

Para borrar el extremo que implementaste, navega a la sección Extremos de la consola de Vertex AI y haz clic en el ícono de borrar:

Luego, haz clic en Anular implementación desde el siguiente mensaje:

Por último, navega a la sección Modelos en la consola, busca ese modelo y, en el menú de tres puntos de la derecha, haz clic en Borrar modelo:

Paso 3: Borra el bucket de Cloud Storage

Para borrar el bucket de almacenamiento, en el menú de navegación de la consola de Cloud, navega a Almacenamiento, selecciona tu bucket y haz clic en Borrar (Delete):