Entrena un modelo de TensorFlow con los datos de BigQuery

Nota: En este instructivo basado en notebooks, algunas muestras de código usan la sintaxis del notebook de Jupyter. Por ejemplo, los comandos de shell con el prefijo ! y el comando mágico de %%writefile. Debes ejecutar estas muestras en un entorno de Jupyter, como notebooks administrados por el usuario de Vertex AI Workbench o Colab. Estas muestras no se ejecutan de forma correcta en un entorno estándar de Python.

Ejecutar en Colab

Ver en GitHub

Abrir en Vertex AI Workbench

Descripción general

En este instructivo, se muestra cómo usar el SDK de Vertex AI para Python a fin de entrenar e implementar un modelo de clasificación tabular personalizado para la predicción en línea.

Objetivo

En este notebook, aprenderás a crear un modelo entrenado de forma personalizada a partir de una secuencia de comandos de Python en un contenedor de Docker con el SDK de Vertex AI para Python y, luego, obtener una predicción del modelo implementado mediante el envío de datos. Como alternativa, puedes crear modelos personalizados con la herramienta de línea de comandos de gcloud o en línea con la consola de Google Cloud.

En este instructivo, se usan los siguientes servicios y recursos de AA de Google Cloud:

• BigQuery
• Cloud Storage
• Conjuntos de datos administrados de Vertex AI
• Vertex AI Training
• Terminales Vertex AI

Los pasos realizados incluyen los siguientes:

• Crea un TrainingPipeline personalizado de Vertex AI para entrenar un modelo.
• Entrena un modelo de TensorFlow.
• Implementa el recurso Model en un recurso de entrega Endpoint.
• Realiza una predicción.
• Anula la implementación del recurso Model.

Conjunto de datos

El conjunto de datos que se usa para este instructivo es el conjunto de datos de pingüinos de conjuntos de datos públicos de BigQuery. En este instructivo, solo debes usar los campos culmen_length_mm, culmen_depth_mm, flipper_length_mm, body_mass_g del conjunto de datos para predecir las especies de pingüinos (species).

Costos

En este instructivo, se usan los siguientes componentes facturables de Google Cloud:

• Vertex AI
• Cloud Storage
• BigQuery

Obtén información sobre los precios de Vertex AI, los precios de Cloud Storage y los precios de BigQuery, y usa la calculadora de precios para generar una estimación de costos en función del uso previsto.

Instalación

Instala la versión más reciente del SDK de Cloud Storage, BigQuery y Vertex AI para Python.

# Install the packages
! pip3 install --upgrade google-cloud-aiplatform \
                        google-cloud-storage \
                        google-cloud-bigquery \
                        pyarrow

Solo Colab: Quita los comentarios de la siguiente celda para reiniciar el kernel.

# Automatically restart kernel after installs so that your environment can access the new packages
# import IPython

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

Antes de comenzar

Determina tu ID del proyecto

Si no conoces el ID del proyecto, intenta lo siguiente:

• Ejecuta gcloud config list.
• Ejecuta gcloud projects list.
• Consulta la página de asistencia: Busca el ID del proyecto

PROJECT_ID = "[your-project-id]"  # @param {type:"string"}

# Set the project id
! gcloud config set project {PROJECT_ID}

Región

También puedes cambiar la variable REGION que usa Vertex AI. Obtén más información sobre las regiones de Vertex AI.

REGION = "us-central1"  # @param {type: "string"}

Autentica la cuenta de Google Cloud

Según tu entorno de Jupyter, es posible que debas autenticarte de forma manual. Sigue las instrucciones correspondientes a continuación.

1. Vertex AI Workbench

• No hagas nada ya que estás autenticado.

2. Instancia de JupyterLab local, quita los comentarios y ejecuta:

# ! gcloud auth login

3. Colab, quita los comentarios y ejecuta lo siguiente:

# from google.colab import auth
# auth.authenticate_user()

4. Cuenta de servicio o alguna otra

• Consulta cómo otorgar permisos de Cloud Storage a tu cuenta de servicio en https://cloud.google.com/storage/docs/gsutil/commands/iam#ch-examples

Crea un bucket de Cloud Storage

Crea un bucket de almacenamiento para almacenar artefactos intermedios como los conjuntos de datos.

BUCKET_URI = "gs://your-bucket-name-unique"  # @param {type:"string"}

Solo si tu bucket aún no existe: Ejecuta la siguiente celda para crear tu bucket de Cloud Storage.

 gsutil mb -l $REGION -p $PROJECT_ID $BUCKET_URI

Inicializa el SDK de Vertex AI para Python

Inicializa el SDK de Vertex AI para Python en tu proyecto y el bucket correspondiente.

from google.cloud import aiplatform

# Initialize the Vertex AI SDK
aiplatform.init(project=PROJECT_ID, location=REGION, staging_bucket=BUCKET_URI)

Inicializa el cliente de BigQuery

Inicializa el cliente de Python de BigQuery para tu proyecto.

Para usar BigQuery, asegúrese de que su cuenta tenga el rol "Usuario de BigQuery".

from google.cloud import bigquery

# Set up BigQuery client
bq_client = bigquery.Client(project=PROJECT_ID)

Preprocesa los datos y divide los datos

Primero, debes descargar y procesar tus datos para el entrenamiento y la prueba.

• Convierte atributos categóricos en numéricos
• Quita las columnas sin usar
• Quita filas inutilizables
• Divide los datos de entrenamiento y prueba

import numpy as np
import pandas as pd

LABEL_COLUMN = "species"

# Define the BigQuery source dataset
BQ_SOURCE = "bigquery-public-data.ml_datasets.penguins"

# Define NA values
NA_VALUES = ["NA", "."]

# Download a table
table = bq_client.get_table(BQ_SOURCE)
df = bq_client.list_rows(table).to_dataframe()

# Drop unusable rows
df = df.replace(to_replace=NA_VALUES, value=np.NaN).dropna()

# Convert categorical columns to numeric
df["island"], _ = pd.factorize(df["island"])
df["species"], _ = pd.factorize(df["species"])
df["sex"], _ = pd.factorize(df["sex"])

# Split into a training and holdout dataset
df_train = df.sample(frac=0.8, random_state=100)
df_holdout = df[~df.index.isin(df_train.index)]

Crea un conjunto de datos tabular de Vertex AI desde el conjunto de datos de BigQuery

Crea un recurso de conjunto de datos tabular de Vertex AI a partir de los datos de entrenamiento de BigQuery.

Consulta más información aquí: https://cloud.google.com/vertex-ai/docs/training/using-managed-datasets

# Create BigQuery dataset
bq_dataset_id = f"{PROJECT_ID}.dataset_id_unique"
bq_dataset = bigquery.Dataset(bq_dataset_id)
bq_client.create_dataset(bq_dataset, exists_ok=True)

dataset = aiplatform.TabularDataset.create_from_dataframe(
    df_source=df_train,
    staging_path=f"bq://{bq_dataset_id}.table-unique",
    display_name="sample-penguins",
)

Entrenar un modelo

Existen dos formas de entrenar un modelo mediante una imagen de contenedor:

• Usa un contenedor compilado previamente de Vertex AI. Si usas un contenedor de entrenamiento compilado previamente, debes especificar un paquete de Python para instalarlo en la imagen de contenedor. Este paquete de Python contiene tu código de entrenamiento.

• Usa tu propia imagen de contenedor personalizada. Si usas tu propio contenedor, la imagen del contenedor debe contener el código de entrenamiento.

Para esta demostración, usarás un contenedor compilado previamente.

Define los argumentos de comando para la secuencia de comandos de entrenamiento

Prepara los argumentos de la línea de comandos para pasarlos a tu secuencia de comandos de entrenamiento.

• args: Los argumentos de línea de comandos que se pasarán al módulo de Python correspondiente. En este ejemplo, son los siguientes:
  • label_column: La columna de etiqueta de tus datos que se predecirá.
  • epochs: La cantidad de ciclos de entrenamiento.
  • batch_size: La cantidad de tamaño de lotes para el entrenamiento.

JOB_NAME = "custom_job_unique"

EPOCHS = 20
BATCH_SIZE = 10

CMDARGS = [
    "--label_column=" + LABEL_COLUMN,
    "--epochs=" + str(EPOCHS),
    "--batch_size=" + str(BATCH_SIZE),
]

Secuencia de comandos de entrenamiento

En la siguiente celda, escribe el contenido de la secuencia de comandos de entrenamiento, task.py. En resumen, la secuencia de comandos hace lo siguiente:

• Carga los datos de la tabla de BigQuery con la biblioteca cliente de Python de BigQuery.
• Compila un modelo con la API de modelos de TF.Keras.
• Compila el modelo (compile()).
• Establece una estrategia de distribución de entrenamiento según el argumento args.distribute.
• Entrena el modelo (fit()) con los ciclos de entrenamiento y el tamaño del lote según los argumentos args.epochs y args.batch_size.
• Obtiene el directorio donde se guardan los artefactos del modelo de la variable de entorno AIP_MODEL_DIR. Esta variable está configurada por el servicio de entrenamiento.
• Guarda el modelo entrenado en el directorio del modelo.

NOTA: Para mejorar el rendimiento del modelo, se recomienda normalizar las entradas al modelo antes del entrenamiento. Consulta el instructivo de TensorFlow en https://www.tensorflow.org/tutorials/structured_data/preprocessing_layers#numerical_columns para obtener más información.

NOTA: El siguiente código de entrenamiento requiere que otorgues a la cuenta de entrenamiento el rol “Usuario de sesión de lectura de BigQuery”. Consulta “https://cloud.google.com/vertex-ai/docs/general/access-control#service-agents para obtener detalles sobre cómo encontrar esta cuenta.

%%writefile task.py

import argparse
import numpy as np
import os

import pandas as pd
import tensorflow as tf

from google.cloud import bigquery
from google.cloud import storage

# Read environmental variables
training_data_uri = os.getenv("AIP_TRAINING_DATA_URI")
validation_data_uri = os.getenv("AIP_VALIDATION_DATA_URI")
test_data_uri = os.getenv("AIP_TEST_DATA_URI")

# Read args
parser = argparse.ArgumentParser()
parser.add_argument('--label_column', required=True, type=str)
parser.add_argument('--epochs', default=10, type=int)
parser.add_argument('--batch_size', default=10, type=int)
args = parser.parse_args()

# Set up training variables
LABEL_COLUMN = args.label_column

# See https://cloud.google.com/vertex-ai/docs/workbench/managed/executor#explicit-project-selection for issues regarding permissions.
PROJECT_NUMBER = os.environ["CLOUD_ML_PROJECT_ID"]
bq_client = bigquery.Client(project=PROJECT_NUMBER)

# Download a table
def download_table(bq_table_uri: str):
    # Remove bq:// prefix if present
    prefix = "bq://"
    if bq_table_uri.startswith(prefix):
        bq_table_uri = bq_table_uri[len(prefix) :]

    # Download the BigQuery table as a dataframe
    # This requires the "BigQuery Read Session User" role on the custom training service account.
    table = bq_client.get_table(bq_table_uri)
    return bq_client.list_rows(table).to_dataframe()

# Download dataset splits
df_train = download_table(training_data_uri)
df_validation = download_table(validation_data_uri)
df_test = download_table(test_data_uri)

def convert_dataframe_to_dataset(
    df_train: pd.DataFrame,
    df_validation: pd.DataFrame,
):
    df_train_x, df_train_y = df_train, df_train.pop(LABEL_COLUMN)
    df_validation_x, df_validation_y = df_validation, df_validation.pop(LABEL_COLUMN)

    y_train = np.asarray(df_train_y).astype("float32")
    y_validation = np.asarray(df_validation_y).astype("float32")

    # Convert to numpy representation
    x_train = np.asarray(df_train_x)
    x_test = np.asarray(df_validation_x)

    # Convert to one-hot representation
    num_species = len(df_train_y.unique())
    y_train = tf.keras.utils.to_categorical(y_train, num_classes=num_species)
    y_validation = tf.keras.utils.to_categorical(y_validation, num_classes=num_species)

    dataset_train = tf.data.Dataset.from_tensor_slices((x_train, y_train))
    dataset_validation = tf.data.Dataset.from_tensor_slices((x_test, y_validation))
    return (dataset_train, dataset_validation)

# Create datasets
dataset_train, dataset_validation = convert_dataframe_to_dataset(df_train, df_validation)

# Shuffle train set
dataset_train = dataset_train.shuffle(len(df_train))

def create_model(num_features):
    # Create model
    Dense = tf.keras.layers.Dense
    model = tf.keras.Sequential(
        [
            Dense(
                100,
                activation=tf.nn.relu,
                kernel_initializer="uniform",
                input_dim=num_features,
            ),
            Dense(75, activation=tf.nn.relu),
            Dense(50, activation=tf.nn.relu),
            Dense(25, activation=tf.nn.relu),
            Dense(3, activation=tf.nn.softmax),
        ]
    )

    # Compile Keras model
    optimizer = tf.keras.optimizers.RMSprop(lr=0.001)
    model.compile(
        loss="categorical_crossentropy", metrics=["accuracy"], optimizer=optimizer
    )

    return model

# Create the model
model = create_model(num_features=dataset_train._flat_shapes[0].dims[0].value)

# Set up datasets
dataset_train = dataset_train.batch(args.batch_size)
dataset_validation = dataset_validation.batch(args.batch_size)

# Train the model
model.fit(dataset_train, epochs=args.epochs, validation_data=dataset_validation)

tf.saved_model.save(model, os.getenv("AIP_MODEL_DIR"))

Entrene el modelo

Define tu TrainingPipeline personalizada en Vertex AI.

Usa la clase CustomTrainingJob para definir TrainingPipeline. La clase usa los siguientes parámetros:

• display_name: El nombre definido por el usuario de esta canalización de entrenamiento.
• script_path: La ruta de acceso local a la secuencia de comandos de entrenamiento.
• container_uri: El URI de la imagen del contenedor de entrenamiento.
• requirements: La lista de dependencias de paquetes de Python de la secuencia de comandos.
• model_serving_container_image_uri: El URI de un contenedor que puede entregar predicciones para tu modelo, ya sea un contenedor precompilado o un contenedor personalizado.

Usa la función run para comenzar el entrenamiento. La función toma los siguientes parámetros:

• args: Los argumentos de línea de comandos que se pasarán a la secuencia de comandos de Python
• replica_count: La cantidad de réplicas de trabajadores.
• model_display_name: El nombre visible de Model si la secuencia de comandos produce un Model administrado.
• machine_type: El tipo de máquina que se usará para el entrenamiento.
• accelerator_type: El tipo de acelerador de hardware.
• accelerator_count: La cantidad de aceleradores que se deben conectar a una réplica de trabajador.

La función run crea una canalización de entrenamiento que entrena y crea un objeto Model. Una vez que se completa la canalización de entrenamiento, la función run muestra el objeto Model.

job = aiplatform.CustomTrainingJob(
    display_name=JOB_NAME,
    script_path="task.py",
    container_uri="us-docker.pkg.dev/vertex-ai/training/tf-cpu.2-8:latest",
    requirements=["google-cloud-bigquery>=2.20.0", "db-dtypes"],
    model_serving_container_image_uri="us-docker.pkg.dev/vertex-ai/prediction/tf2-cpu.2-8:latest",
)

MODEL_DISPLAY_NAME = "penguins_model_unique"

# Start the training
model = job.run(
    dataset=dataset,
    model_display_name=MODEL_DISPLAY_NAME,
    bigquery_destination=f"bq://{PROJECT_ID}",
    args=CMDARGS,
)

Implementa el modelo

Antes de usar el modelo para hacer predicciones, debes implementarlo en un Endpoint. Para ello, llama a la función deploy en el recurso Model. Esto realiza dos acciones:

1. Crea un recurso Endpoint para implementar el recurso Model.
2. Implementa el recurso Model en el recurso Endpoint.

La función toma los siguientes parámetros:

• deployed_model_display_name: Un nombre legible para el modelo implementado.
• traffic_split: El porcentaje de tráfico en el extremo que va a este modelo, que se especifica como un diccionario de uno o más pares clave-valor.
  • Si solo hay un modelo, especifica { "0": 100 }, en el que “0” hace referencia a este modelo que se sube y 100 significa el 100% del tráfico.
  • Si hay modelos existentes en el extremo en los que se dividirá el tráfico, usa model_id para especificar { "0": percent, model_id: percent, ... }, donde model_id es el ID de un DeployedModel existente en el extremo. Los porcentajes deben sumar hasta 100.
• machine_type: El tipo de máquina que se usará para el entrenamiento.
• accelerator_type: El tipo de acelerador de hardware.
• accelerator_count: La cantidad de aceleradores que se deben conectar a una réplica de trabajador.
• starting_replica_count: La cantidad de instancias de procesamiento que se aprovisionarán inicialmente.
• max_replica_count: La cantidad máxima de instancias de procesamiento para escalar. En este instructivo, solo se aprovisiona una instancia.

División del tráfico

El parámetro traffic_split se especifica como un diccionario de Python. Puedes implementar más de una instancia de tu modelo en un extremo y, luego, establecer el porcentaje de tráfico que se dirige a cada instancia.

Puedes usar una división del tráfico para ingresar un modelo nuevo en producción de forma gradual. Por ejemplo, si tuvieras un modelo existente en producción con el 100% del tráfico, podrías implementar un modelo nuevo en el mismo extremo, dirigir el 10% del tráfico a él y reducir el tráfico del modelo original al 90%. Esto te permite supervisar el rendimiento del nuevo modelo y, al mismo tiempo, minimizar la interrupción para la mayoría de los usuarios.

Escalamiento de instancias de procesamiento

Puedes especificar una sola instancia (o nodo) para que entregue tus solicitudes de predicción en línea. En este instructivo, se usa un solo nodo, por lo que las variables MIN_NODES y MAX_NODES se configuran como 1.

Si deseas usar varios nodos para entregar tus solicitudes de predicción en línea, establece MAX_NODES en la cantidad máxima de nodos que deseas usar. Vertex AI escala de forma automática la cantidad de nodos que se usan para entregar tus predicciones, hasta la cantidad máxima que establezcas. Consulta la página de precios para comprender los costos del ajuste de escala automático con varios nodos.

Extremo

El método deploy espera hasta que se implemente el modelo y, luego, muestra un objeto Endpoint. Si es la primera vez que se implementa un modelo en el extremo, puede tardar unos minutos más en completar el aprovisionamiento de recursos.

DEPLOYED_NAME = "penguins_deployed_unique"

endpoint = model.deploy(deployed_model_display_name=DEPLOYED_NAME)

Realiza una solicitud de predicción en línea

Envía una solicitud de predicción en línea a su modelo implementado.

Prepara datos de prueba

Preparar datos de prueba convirtándolos en una lista de Python

df_holdout_y = df_holdout.pop(LABEL_COLUMN)
df_holdout_x = df_holdout

# Convert to list representation
holdout_x = np.array(df_holdout_x).tolist()
holdout_y = np.array(df_holdout_y).astype("float32").tolist()

Envía la solicitud de predicción

Ahora que tienes datos de prueba, puedes usarlos para enviar una solicitud de predicción. Usa la función predict del objeto Endpoint, que usa los siguientes parámetros:

• instances: Una lista de instancias de medición de pingüinos. Según tu modelo personalizado, cada instancia debe ser un array de números. Preparaste esta lista en el paso anterior.

La función predict muestra una lista, en la que cada elemento de la lista corresponde a la instancia en la solicitud. En el resultado de cada predicción, verás lo siguiente:

• El nivel de confianza de la predicción (predictions), entre 0 y 1, para cada una de las diez clases.

Luego, puedes ejecutar una evaluación rápida sobre los resultados de la predicción:

1. np.argmax: Convierte cada lista de niveles de confianza en una etiqueta
2. Imprime predicciones

predictions = endpoint.predict(instances=holdout_x)
y_predicted = np.argmax(predictions.predictions, axis=1)

y_predicted

Anular implementaciones de modelos

Para anular la implementación de todos los recursos Model del recurso Endpoint de entrega, usa el método undeploy_all del extremo.

endpoint.undeploy_all()

Realice una limpieza

Si deseas limpiar todos los recursos de Google Cloud que se usaron en este proyecto, puedes borrar el proyecto de Google Cloud que usaste para el instructivo.

De lo contrario, puedes borrar los recursos individuales que creaste en este instructivo:

• Trabajo de entrenamiento
• Modelo
• Extremo
• Bucket de Cloud Storage

import os

# Delete the training job
job.delete()

# Delete the model
model.delete()

# Delete the endpoint
endpoint.delete()

# Warning: Setting this to true deletes everything in your bucket
delete_bucket = False

if delete_bucket or os.getenv("IS_TESTING"):
    ! gsutil rm -r $BUCKET_URI