1. Descripción general
En este lab, usarás Vertex AI para crear una canalización que entrene un modelo de Keras personalizado en TensorFlow. Luego, usaremos la nueva funcionalidad disponible en Vertex AI Experiments para hacer un seguimiento y comparar las ejecuciones de modelos a fin de identificar qué combinación de hiperparámetros genera 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 vuelva a guardar el modelo resultante en Cloud Storage
- Aprovecha Vertex ML Metadata para guardar artefactos de modelo, como modelos y métricas del modelo.
- Utilizar Vertex AI Experiments para comparar los resultados de las diversas ejecuciones de canalizaciones
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 popular sobre fútbol extraído de EA Sports' FIFA 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 preprocesaron de antemano para que puedas comenzar con mayor facilidad. Utilizarás este conjunto de datos a lo largo del lab, que ahora puedes encontrar en un bucket público de Cloud Storage. Proporcionaremos más detalles más adelante en el codelab 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é es útil Vertex AI Experiments 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 están impulsados por hipótesis, utilizando el método de prueba y error para probar varias hipótesis con la esperanza de que las iteraciones sucesivas generen un modelo más eficaz.
Si bien los equipos de ciencia de datos adoptaron la experimentación, a menudo se esfuerzan por hacer un seguimiento de su trabajo y el "ingrediente secreto" descubierto gracias a sus esfuerzos de experimentación. Esto sucede por 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 miras a todo el equipo de ciencia de datos, ya que es posible que no todos los miembros estén haciendo un seguimiento de los experimentos o incluso compartiendo sus resultados con otros.
- La captura de datos lleva mucho tiempo y la mayoría de los equipos aprovechan los métodos manuales (p. ej., hojas de cálculo o documentos) que generan información incoherente y incompleta de la que aprender.
tl;dr: Vertex AI Experiments hace el trabajo por ti para ayudarte a hacer un seguimiento de tus experimentos y compararlos con más facilidad
¿Por qué usar Vertex AI Experiments para videojuegos?
Históricamente, los videojuegos han sido un espacio 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 también usan todos esos datos y aprovechan el AA y los experimentos de AA para mejorar las experiencias en el juego, retener a los jugadores y evaluarlos 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 para TensorFlow 2 (local).
5. Pasos de configuración iniciales en tu notebook
Deberás realizar una serie de pasos adicionales para configurar el entorno en tu notebook antes de crear la canalización. Entre estos pasos se incluyen la instalación de paquetes adicionales, la configuración de variables, la creación de tu bucket de Cloud Storage, la copia del conjunto de datos de videojuegos de un bucket de almacenamiento público, la importación de bibliotecas y la definición de constantes adicionales.
Paso 1: instalar 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 tu notebook.
# 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 queremos 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.
- 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 son compatibles con 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, configuraremos una variable TIMESTAMP. Esta variable se usa para evitar conflictos de nombres entre los usuarios en los recursos creados. Crea un TIMESTAMP para cada sesión de instancia y lo adjuntas 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 pruebas de Cloud Storage. El bucket de etapa de pruebas es donde se retienen todos los datos asociados con los recursos de tu conjunto de datos y 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, incluidos 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 NO existe, puedes ejecutar la siguiente celda para crear tu bucket de 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 videojuegos
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
Luego, importaremos las 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 los archivos de 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. Creemos nuestra canalización
Ahora la diversión puede comenzar y podemos comenzar a aprovechar Vertex AI para compilar nuestra canalización de entrenamiento. Inicializaremos el SDK de Vertex AI, configuraremos el trabajo de entrenamiento como un componente de la canalización, crearemos la canalización, enviaremos las ejecuciones de las canalizaciones 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 tu 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, necesitaremos especificar el trabajo de entrenamiento definiéndolo como un componente de la canalización. Nuestra canalización tomará hiperparámetros y datos de entrenamiento (p.ej., DROPOUT_RATE, LEARNING_RATE, EPOCHS) como entradas y métricas del modelo 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
Terminamos de configurar nuestro componente y de definir nuestra canalización. Ya está todo listo para enviar varias ejecuciones de la canalización que especificamos anteriormente. Para ello, necesitaremos 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 realizar feeds de forma correcta en 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 las canalizaciones. 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 tus canalizaciones.
#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 (por ejemplo, cada 60 segundos) para ver cómo cambian los estados 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 mejor rendimiento
Genial, 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á un aspecto similar al siguiente:
Tanto el MAE como la RMSE son mediciones del error de predicción del modelo promedio, por lo que un valor más bajo para ambas métricas es conveniente en la mayoría de los casos. Según los resultados de Vertex AI Experiments, según los resultados de Vertex AI Experiments, la ejecución más exitosa de ambas métricas fue la última con un dropout_rate de 0.001, un learning_rate si es 0.001 y la cantidad total de epochs fue 20. Según este experimento, estos parámetros del modelo se usarían en la producción, ya que da como resultado 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.
- Crear y ejecutar una canalización de 5 pasos que transfiera datos de GCS, escale los datos, entrene el modelo, lo evalúe y guarde el modelo resultante de nuevo en GCS
- Aprovecha Vertex ML Metadata para guardar artefactos de modelo, como modelos y métricas del modelo.
- Utilizar Vertex AI Experiments para comparar los resultados de las diversas ejecuciones de canalizaciones
Para obtener más información sobre las distintas partes de Vertex, consulte la documentación.
8. 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 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):
