1. Descripción general
En este lab, usarás Vertex AI para compilar una canalización que entrene un modelo de Keras personalizado en TensorFlow. Luego, usaremos la nueva funcionalidad disponible en Vertex AI Experiments para realizar un seguimiento y comparar las ejecuciones de modelos para identificar qué combinación de hiperparámetros da como resultado el mejor rendimiento.
Qué aprenderá
Aprenderás a hacer lo siguiente:
- Entrena un modelo de Keras personalizado para predecir las calificaciones de los jugadores (p. ej., regresión).
- Aprenderás a usar el SDK de canalizaciones de Kubeflow para compilar canalizaciones de AA escalables.
- Crear y ejecutar una canalización de 5 pasos que transfiera datos de Cloud Storage, escale los datos, entrene el modelo, lo evalúe y guarde el modelo resultante de nuevo en Cloud Storage
- Aprovecha Vertex ML Metadata para guardar artefactos de modelos, como modelos y métricas de modelos
- Usa Vertex AI Experiments para comparar los resultados de las distintas ejecuciones de canalización.
El costo total de la ejecución de este lab en Google Cloud es de aproximadamente $1.
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.
Vertex AI incluye muchos productos distintos para respaldar flujos de trabajo de AA de extremo a extremo. Este lab se enfocará en los productos que se destacan a continuación: Experiments, Pipelines, ML Metadata y Workbench
3. Descripción general del caso de uso
Usaremos un conjunto de datos de fútbol popular obtenido de la serie de videojuegos FIFA de EA Sports. Incluye más de 25,000 partidos de fútbol y más de 10,000 jugadores para las temporadas 2008 a 2016. Los datos se procesaron previamente para que puedas comenzar a usarlos con mayor facilidad. Usarás este conjunto de datos durante el lab, que ahora se encuentra en un bucket público de Cloud Storage. Más adelante en el codelab, proporcionaremos más detalles sobre cómo acceder al conjunto de datos. Nuestro objetivo final es predecir la calificación general de un jugador en función de varias acciones en el juego, como intercepciones y penalizaciones.
¿Por qué Vertex AI Experiments es útil para la ciencia de datos?
La ciencia de datos es experimental por naturaleza; después de todo, se los llama científicos. Los buenos científicos de datos se basan en hipótesis y utilizan el método de prueba y error para probar varias hipótesis con la esperanza de que las iteraciones sucesivas generen un modelo con mejor rendimiento.
Si bien los equipos de ciencia de datos adoptaron la experimentación, a menudo tienen dificultades para hacer un seguimiento de su trabajo y de la “salsa secreta” que se descubrió a través de sus esfuerzos de experimentación. Esto puede deberse a varios motivos:
- El seguimiento de los trabajos de entrenamiento puede volverse engorroso, lo que facilita perder de vista lo que funciona frente a lo que no.
- Este problema se agrava cuando se observa un equipo de ciencia de datos, ya que es posible que no todos los miembros hagan un seguimiento de los experimentos ni compartan sus resultados con otros.
- La captura de datos lleva tiempo y la mayoría de los equipos aprovechan métodos manuales (p. ej., hojas o documentos) que generan información incoherente y incompleta para aprender.
Resumen: Vertex AI Experiments hace el trabajo por ti y te ayuda a realizar un seguimiento y comparar tus experimentos con mayor facilidad.
¿Por qué usar Vertex AI Experiments para videojuegos?
Históricamente, los videojuegos han sido un campo de pruebas para el aprendizaje automático y los experimentos de AA. Los juegos no solo producen miles de millones de eventos en tiempo real por día, sino que aprovechan todos esos datos aprovechando el AA y los experimentos de AA para mejorar las experiencias en el juego, retener a los jugadores y evaluar a los diferentes jugadores en su plataforma. Por lo tanto, pensamos que un conjunto de datos de videojuegos encajaría bien con nuestro ejercicio general de experimentos.
4. Configura el entorno
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: Habilita la API de Compute Engine
Ve a Compute Engine y selecciona Habilitar (si aún no está habilitada).
Paso 2: Habilita la API de Vertex AI
Navegue hasta la sección de Vertex AI en la consola de Cloud y haga clic en Habilitar API de Vertex AI.
Paso 3: Crea una instancia de Vertex AI Workbench
En la sección Vertex AI de Cloud Console, haz clic en Workbench:
Habilita la API de Notebooks si aún no está habilitada.
Una vez habilitada, haz clic en NOTEBOOKS ADMINISTRADOS (MANAGED NOTEBOOKS):
Luego, selecciona NUEVO NOTEBOOK (NEW NOTEBOOK).
Asígnale un nombre al notebook y, luego, haz clic en Configuración avanzada (Advanced settings).
En Configuración avanzada (Advanced settings), activa la opción Habilitar el cierre inactivo (Enable Idle Shutdown) y establece la cantidad de minutos en 60. Esto provocará que el notebook se cierre automáticamente cuando no esté en uso para que no se generen costos innecesarios.
Paso 4: Abre tu notebook
Una vez que se crea la instancia, selecciona Abrir JupyterLab (Open JupyterLab).
Paso 5: Realiza la autenticación (solo la primera vez)
La primera vez que uses una instancia nueva, se te solicitará que te autentiques. Sigue los pasos en la IU para hacerlo.
Paso 6: Selecciona el kernel adecuado
Los notebooks administrados proporcionan varios kernels en una sola IU. Selecciona el kernel de Tensorflow 2 (local).
5. Pasos iniciales para la configuración en tu notebook
Deberás seguir una serie de pasos adicionales para configurar tu entorno en el notebook antes de compilar tu canalización. Estos pasos incluyen instalar paquetes adicionales, configurar variables, crear tu bucket de almacenamiento en la nube, copiar el conjunto de datos de juegos de un bucket de almacenamiento público, importar bibliotecas y definir constantes adicionales.
Paso 1: Instala paquetes adicionales
Tendremos que instalar dependencias de paquetes adicionales que no están instaladas actualmente en tu entorno de notebook. Un ejemplo incluye el SDK de KFP.
!pip3 install --user --force-reinstall 'google-cloud-aiplatform>=1.15' -q --no-warn-conflicts
!pip3 install --user kfp -q --no-warn-conflicts
Luego, deberás reiniciar el kernel del notebook para poder usar los paquetes descargados en él.
# Automatically restart kernel after installs
import os
if not os.getenv("IS_TESTING"):
# Automatically restart kernel after installs
import IPython
app = IPython.Application.instance()
app.kernel.do_shutdown(True)
Paso 2: Establece las variables
Queremos definir nuestro PROJECT_ID. Si no conoces tu Project_ID, es posible que puedas obtener tu PROJECT_ID con gcloud.
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, configura tu PROJECT_ID aquí.
if PROJECT_ID == "" or PROJECT_ID is None:
PROJECT_ID = "[your-project-id]" # @param {type:"string"}
También querremos establecer la variable REGION, que se usa en el resto de este notebook. A continuación, se muestran las regiones compatibles con Vertex AI. Te recomendamos que elijas la región más cercana a ti.
- América: us-central1
- Europa: europe-west4
- Asia-Pacífico: asia-east1
No uses un bucket multirregional para el entrenamiento con Vertex AI. No todas las regiones brindan asistencia para todos los servicios de Vertex AI. Obtén más información sobre las regiones de Vertex AI.
#set your region
REGION = "us-central1" # @param {type: "string"}
Por último, estableceremos una variable TIMESTAMP. Esta variable se usa para evitar conflictos de nombres entre los usuarios en los recursos creados. Creas un TIMESTAMP para cada sesión de instancia y lo agregas al nombre de los recursos que creas en este instructivo.
#set timestamp to avoid collisions between multiple users
from datetime import datetime
TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")
Paso 3: Crea un bucket de Cloud Storage
Deberás especificar y aprovechar un bucket de etapa de Cloud Storage. En el bucket de etapa de pruebas, se retienen todos los datos asociados con tu conjunto de datos y los recursos de modelo en todas las sesiones.
A continuación, establece el nombre del bucket de Cloud Storage. Los nombres de los buckets deben ser únicos a nivel global en todos los proyectos de Google Cloud, incluso aquellos fuera de tu organización.
#set cloud storage bucket
BUCKET_NAME = "[insert bucket name here]" # @param {type:"string"}
BUCKET_URI = f"gs://{BUCKET_NAME}"
Si tu bucket aún NO existe, puedes ejecutar la siguiente celda para crearlo en Cloud Storage.
! gsutil mb -l $REGION -p $PROJECT_ID $BUCKET_URI
Luego, puedes verificar el acceso a tu bucket de Cloud Storage ejecutando la siguiente celda.
#verify access
! gsutil ls -al $BUCKET_URI
Paso 4: Copia nuestro conjunto de datos de juegos
Como se mencionó anteriormente, aprovecharás un conjunto de datos popular de los populares videojuegos de EA Sports, FIFA. Ya hicimos el trabajo de procesamiento previo por ti, por lo que solo tendrás que copiar el conjunto de datos del bucket de almacenamiento público y moverlo al que creaste.
# copy the data over to your cloud storage bucket
DATASET_URI = "gs://cloud-samples-data/vertex-ai/structured_data/player_data"
!gsutil cp -r $DATASET_URI $BUCKET_URI
Paso 5: Importa bibliotecas y define constantes adicionales
A continuación, querremos importar nuestras bibliotecas para Vertex AI, KFP, etcétera.
import logging
import os
import time
logger = logging.getLogger("logger")
logging.basicConfig(level=logging.INFO)
import kfp.v2.compiler as compiler
# Pipeline Experiments
import kfp.v2.dsl as dsl
# Vertex AI
from google.cloud import aiplatform as vertex_ai
from kfp.v2.dsl import Artifact, Input, Metrics, Model, Output, component
from typing import NamedTuple
También definiremos constantes adicionales a las que haremos referencia en el resto del notebook, como las rutas de acceso a nuestros datos de entrenamiento.
#import libraries and define constants
# Experiments
TASK = "regression"
MODEL_TYPE = "tensorflow"
EXPERIMENT_NAME = f"{PROJECT_ID}-{TASK}-{MODEL_TYPE}-{TIMESTAMP}"
# Pipeline
PIPELINE_URI = f"{BUCKET_URI}/pipelines"
TRAIN_URI = f"{BUCKET_URI}/player_data/data.csv"
LABEL_URI = f"{BUCKET_URI}/player_data/labels.csv"
MODEL_URI = f"{BUCKET_URI}/model"
DISPLAY_NAME = "experiments-demo-gaming-data"
BQ_DATASET = "player_data"
BQ_LOCATION = "US"
VIEW_NAME = 'dataset_test'
PIPELINE_JSON_PKG_PATH = "experiments_demo_gaming_data.json"
PIPELINE_ROOT = f"gs://{BUCKET_URI}/pipeline_root"
6. Construyamos nuestra canalización
Ahora podemos comenzar a usar Vertex AI para crear nuestra canalización de entrenamiento. Inicializaremos el SDK de Vertex AI, configuraremos nuestro trabajo de entrenamiento como un componente de canalización, compilaremos nuestra canalización, enviaremos nuestras ejecuciones de canalización y aprovecharemos el SDK de Vertex AI para ver los experimentos y supervisar su estado.
Paso 1: Inicializa el SDK de Vertex AI
Inicializa el SDK de Vertex AI y configura PROJECT_ID y BUCKET_URI.
#initialize vertex AI SDK
vertex_ai.init(project=PROJECT_ID, staging_bucket=BUCKET_URI)
Paso 2: Configurar el trabajo de entrenamiento como un componente de la canalización
Para comenzar a ejecutar nuestros experimentos, debemos especificar nuestro trabajo de entrenamiento definiéndolo como un componente de canalización. Nuestra canalización recibirá datos de entrenamiento y hiperparámetros (p.ej., DROPOUT_RATE, LEARNING_RATE, EPOCHS) como entradas y métricas de modelos de salida (p.ej., MAE y RMSE) y un artefacto del modelo.
@component(
packages_to_install=[
"numpy==1.21.0",
"pandas==1.3.5",
"scikit-learn==1.0.2",
"tensorflow==2.9.0",
]
)
def custom_trainer(
train_uri: str,
label_uri: str,
dropout_rate: float,
learning_rate: float,
epochs: int,
model_uri: str,
metrics: Output[Metrics],
model_metadata: Output[Model],
):
# import libraries
import logging
import uuid
from pathlib import Path as path
import pandas as pd
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense
from tensorflow.keras.layers import Dropout
from tensorflow.keras.metrics import Metric
from sklearn.metrics import accuracy_score
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.metrics import mean_absolute_error
import numpy as np
from math import sqrt
import os
import tempfile
# set variables and use gcsfuse to update prefixes
gs_prefix = "gs://"
gcsfuse_prefix = "/gcs/"
train_path = train_uri.replace(gs_prefix, gcsfuse_prefix)
label_path = label_uri.replace(gs_prefix, gcsfuse_prefix)
model_path = model_uri.replace(gs_prefix, gcsfuse_prefix)
def get_logger():
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
)
logger.addHandler(handler)
return logger
def get_data(
train_path: str,
label_path: str
) -> (pd.DataFrame):
#load data into pandas dataframe
data_0 = pd.read_csv(train_path)
labels_0 = pd.read_csv(label_path)
#drop unnecessary leading columns
data = data_0.drop('Unnamed: 0', axis=1)
labels = labels_0.drop('Unnamed: 0', axis=1)
#save as numpy array for reshaping of data
labels = labels.values
data = data.values
# Split the data
labels = labels.reshape((labels.size,))
train_data, test_data, train_labels, test_labels = train_test_split(data, labels, test_size=0.2, shuffle=True, random_state=7)
#Convert data back to pandas dataframe for scaling
train_data = pd.DataFrame(train_data)
test_data = pd.DataFrame(test_data)
train_labels = pd.DataFrame(train_labels)
test_labels = pd.DataFrame(test_labels)
#Scale and normalize the training dataset
scaler = StandardScaler()
scaler.fit(train_data)
train_data = pd.DataFrame(scaler.transform(train_data), index=train_data.index, columns=train_data.columns)
test_data = pd.DataFrame(scaler.transform(test_data), index=test_data.index, columns=test_data.columns)
return train_data,train_labels, test_data, test_labels
""" Train your Keras model passing in the training data and values for learning rate, dropout rate,and the number of epochs """
def train_model(
learning_rate: float,
dropout_rate: float,
epochs: float,
train_data: pd.DataFrame,
train_labels: pd.DataFrame):
# Train tensorflow model
param = {"learning_rate": learning_rate, "dropout_rate": dropout_rate, "epochs": epochs}
model = Sequential()
model.add(Dense(500, input_dim=train_data.shape[1], activation= "relu"))
model.add(Dropout(param['dropout_rate']))
model.add(Dense(100, activation= "relu"))
model.add(Dense(50, activation= "relu"))
model.add(Dense(1))
model.compile(
tf.keras.optimizers.Adam(learning_rate= param['learning_rate']),
loss='mse',
metrics=[tf.keras.metrics.RootMeanSquaredError(),tf.keras.metrics.MeanAbsoluteError()])
model.fit(train_data, train_labels, epochs= param['epochs'])
return model
# Get Predictions
def get_predictions(model, test_data):
dtest = pd.DataFrame(test_data)
pred = model.predict(dtest)
return pred
# Evaluate predictions with MAE
def evaluate_model_mae(pred, test_labels):
mae = mean_absolute_error(test_labels, pred)
return mae
# Evaluate predictions with RMSE
def evaluate_model_rmse(pred, test_labels):
rmse = np.sqrt(np.mean((test_labels - pred)**2))
return rmse
#Save your trained model in GCS
def save_model(model, model_path):
model_id = str(uuid.uuid1())
model_path = f"{model_path}/{model_id}"
path(model_path).parent.mkdir(parents=True, exist_ok=True)
model.save(model_path + '/model_tensorflow')
# Main ----------------------------------------------
train_data, train_labels, test_data, test_labels = get_data(train_path, label_path)
model = train_model(learning_rate, dropout_rate, epochs, train_data,train_labels )
pred = get_predictions(model, test_data)
mae = evaluate_model_mae(pred, test_labels)
rmse = evaluate_model_rmse(pred, test_labels)
save_model(model, model_path)
# Metadata ------------------------------------------
#convert numpy array to pandas series
mae = pd.Series(mae)
rmse = pd.Series(rmse)
#log metrics and model artifacts with ML Metadata. Save metrics as a list.
metrics.log_metric("mae", mae.to_list())
metrics.log_metric("rmse", rmse.to_list())
model_metadata.uri = model_uri
Paso 3: Crear nuestra canalización
Ahora configuraremos nuestro flujo de trabajo con Domain Specific Language (DSL) disponible en KFP y compilaremos nuestra canalización en un archivo JSON.
# define our workflow
@dsl.pipeline(name="gaming-custom-training-pipeline")
def pipeline(
train_uri: str,
label_uri: str,
dropout_rate: float,
learning_rate: float,
epochs: int,
model_uri: str,
):
custom_trainer(
train_uri,label_uri, dropout_rate,learning_rate,epochs, model_uri
)
#compile our pipeline
compiler.Compiler().compile(pipeline_func=pipeline, package_path="gaming_pipeline.json")
Paso 4: Envía nuestras ejecuciones de canalización
El trabajo duro consiste en configurar nuestro componente y definir nuestra canalización. Ya podemos enviar varias ejecuciones de la canalización que especificamos anteriormente. Para ello, deberemos definir los valores de nuestros diferentes hiperparámetros de la siguiente manera:
runs = [
{"dropout_rate": 0.001, "learning_rate": 0.001,"epochs": 20},
{"dropout_rate": 0.002, "learning_rate": 0.002,"epochs": 25},
{"dropout_rate": 0.003, "learning_rate": 0.003,"epochs": 30},
{"dropout_rate": 0.004, "learning_rate": 0.004,"epochs": 35},
{"dropout_rate": 0.005, "learning_rate": 0.005,"epochs": 40},
]
Con los hiperparámetros definidos, podemos aprovechar un for loop para ingresar correctamente las diferentes ejecuciones de la canalización:
for i, run in enumerate(runs):
job = vertex_ai.PipelineJob(
display_name=f"{EXPERIMENT_NAME}-pipeline-run-{i}",
template_path="gaming_pipeline.json",
pipeline_root=PIPELINE_URI,
parameter_values={
"train_uri": TRAIN_URI,
"label_uri": LABEL_URI,
"model_uri": MODEL_URI,
**run,
},
)
job.submit(experiment=EXPERIMENT_NAME)
Paso 5: Aprovecha el SDK de Vertex AI para ver los experimentos
El SDK de Vertex AI te permite supervisar el estado de las ejecuciones de canalización. También puedes usarlo para mostrar parámetros y métricas de las ejecuciones de canalización en el experimento de Vertex AI. Usa el siguiente código para ver los parámetros asociados con tus ejecuciones y su estado actual.
# see state/status of all the pipeline runs
vertex_ai.get_experiment_df(EXPERIMENT_NAME)
Puedes aprovechar el siguiente código para obtener actualizaciones sobre el estado de las ejecuciones de tu canalización.
#check on current status
while True:
pipeline_experiments_df = vertex_ai.get_experiment_df(EXPERIMENT_NAME)
if all(
pipeline_state != "COMPLETE" for pipeline_state in pipeline_experiments_df.state
):
print("Pipeline runs are still running...")
if any(
pipeline_state == "FAILED"
for pipeline_state in pipeline_experiments_df.state
):
print("At least one Pipeline run failed")
break
else:
print("Pipeline experiment runs have completed")
break
time.sleep(60)
También puedes llamar a trabajos de canalización específicos con run_name.
# Call the pipeline runs based on the experiment run name
pipeline_experiments_df = vertex_ai.get_experiment_df(EXPERIMENT_NAME)
job = vertex_ai.PipelineJob.get(pipeline_experiments_df.run_name[0])
print(job.resource_name)
print(job._dashboard_uri())
Por último, puedes actualizar el estado de tus ejecuciones en intervalos establecidos (como cada 60 segundos) para ver cómo los estados cambian de RUNNING a FAILED o COMPLETE.
# wait 60 seconds and view state again
import time
time.sleep(60)
vertex_ai.get_experiment_df(EXPERIMENT_NAME)
7. Identifica la ejecución con el mejor rendimiento
Excelente, ahora tenemos los resultados de las ejecuciones de nuestra canalización. Quizás te preguntes qué puedo aprender de los resultados. El resultado de tus experimentos debe contener cinco filas, una para cada ejecución de la canalización. Tendrá el siguiente aspecto:
Tanto el MAE como la RMSE son mediciones del error de predicción del modelo promedio, por lo que es conveniente obtener un valor más bajo para ambas métricas en la mayoría de los casos. Según el resultado de los experimentos de Vertex AI, podemos ver que nuestra ejecución más exitosa en ambas métricas fue la ejecución final con un dropout_rate de 0.001, un learning_rate de 0.001 y una cantidad total de epochs de 20. En función de este experimento, estos parámetros del modelo se usarían en producción, ya que proporcionan el mejor rendimiento del modelo.
Con eso, terminaste el lab.
🎉 ¡Felicitaciones! 🎉
Aprendiste a usar Vertex AI para hacer lo siguiente:
- Entrena un modelo de Keras personalizado para predecir las calificaciones de los jugadores (p.ej., regresión)
- Aprenderás a usar el SDK de canalizaciones de Kubeflow para compilar canalizaciones de AA escalables.
- Crea y ejecuta una canalización de 5 pasos que transfiera datos de GCS, los escale, entrene el modelo, lo evalúe y lo guarde en GCS
- Aprovecha Vertex ML Metadata para guardar artefactos de modelos, como modelos y métricas de modelos
- Usa Vertex AI Experiments para comparar los resultados de las distintas ejecuciones de canalización.
Para obtener más información sobre las distintas partes de Vertex, consulte la documentación.
8. Limpieza
Para que no se te cobre, te recomendamos que borres los recursos que creaste a lo largo de 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:
Paso 2: 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):
