1. Présentation
Dans cet atelier, vous allez utiliser Vertex AI pour créer un pipeline qui entraîne un modèle Keras personnalisé dans TensorFlow. Vous utiliserez ensuite la nouvelle fonctionnalité disponible dans Vertex AI Experiments pour suivre et comparer les exécutions de modèles afin d'identifier la combinaison d'hyperparamètres qui offre les meilleures performances.
Objectifs
Vous allez apprendre à effectuer les opérations suivantes :
- Entraîner un modèle Keras personnalisé pour prédire les notes des joueurs (par exemple, une régression)
- Utiliser le SDK Kubeflow Pipelines pour créer des pipelines de ML évolutifs
- Créer et exécuter un pipeline en cinq étapes qui ingère des données depuis Cloud Storage, les met à l'échelle, entraîne le modèle, l'évalue et enregistre le modèle obtenu dans Cloud Storage
- Exploiter Vertex ML Metadata pour enregistrer des artefacts de modèle tels que des modèles et des métriques de modèle
- Utiliser Vertex AI Experiments pour comparer les résultats des différentes exécutions de pipeline
Le coût total d'exécution de cet atelier sur Google Cloud est d'environ 1 $.
2. Présentation de Vertex AI
Cet atelier utilise la toute dernière offre de produits d'IA de Google Cloud. Vertex AI simplifie l'expérience de développement en intégrant toutes les offres de ML de Google Cloud. Auparavant, les modèles entraînés avec AutoML et les modèles personnalisés étaient accessibles depuis des services distincts. La nouvelle offre regroupe ces deux types de modèles mais aussi d'autres nouveaux produits en une seule API. Vous pouvez également migrer des projets existants vers Vertex AI.
Vertex AI comprend de nombreux produits différents qui permettent de gérer les workflows de ML de bout en bout. Cet atelier se concentre sur les produits mis en évidence ci-dessous : Experiments, Pipelines, ML Metadata et Workbench.
3. Présentation du cas d'utilisation
Nous allons utiliser un ensemble de données de football populaire provenant de la série de jeux vidéo FIFA d'EA Sports. Il comprend plus de 25 000 matchs de football et plus de 10 000 joueurs pour les saisons 2008-2016. Les données ont été prétraitées à l'avance pour que vous puissiez vous lancer plus facilement. Vous utiliserez cet ensemble de données tout au long de l'atelier. Il se trouve désormais dans un bucket Cloud Storage public. Nous vous fournirons plus de détails ultérieurement dans l'atelier de programmation sur la façon d'accéder à l'ensemble de données. Notre objectif final est de prédire la note globale d'un joueur en fonction de diverses actions en jeu telles que les interceptions et les pénalités.
Pourquoi Vertex AI Experiments est-il utile pour la science des données ?
La science des données est expérimentale par nature. Les professionnels de ce domaine sont appelés "scientifiques" après tout. Les bons data scientists sont axés sur les hypothèses. Ils utilisent une approche par essais et erreurs pour tester diverses hypothèses dans l'espoir que les itérations successives aboutiront à un modèle plus performant.
Bien que les équipes de science des données aient adopté l'expérimentation, elles ont souvent du mal à suivre leur travail et la "recette secrète" découverte grâce à leurs efforts d'expérimentation. Cela se produit pour plusieurs raisons :
- Le suivi des jobs d'entraînement peut devenir fastidieux, ce qui rend difficile de distinguer ce qui fonctionne de ce qui ne fonctionne pas.
- Ce problème s'aggrave lorsque vous examinez une équipe de science des données, car tous les membres ne suivent pas les tests ni ne partagent leurs résultats avec les autres.
- La capture de données prend du temps et la plupart des équipes utilisent des méthodes manuelles (par exemple, des feuilles ou des documents) qui génèrent des informations incohérentes et incomplètes à partir desquelles tirer des enseignements.
En résumé : Vertex AI Experiments fait le travail à votre place, ce qui vous permet de suivre et de comparer plus facilement vos tests.
Pourquoi utiliser Vertex AI Experiments pour les jeux ?
Les jeux ont toujours été un terrain de jeu pour le machine learning et les tests de ML. Non seulement les jeux génèrent des milliards d'événements en temps réel par jour, mais ils utilisent toutes ces données en exploitant le ML et les tests de ML pour améliorer les expériences en jeu, fidéliser les joueurs et évaluer les différents joueurs sur leur plate-forme. C'est pourquoi nous avons pensé qu'un ensemble de données de jeu correspondait bien à notre exercice de test global.
4. Configurer votre environnement
Pour suivre cet atelier de programmation, vous aurez besoin d'un projet Google Cloud Platform dans lequel la facturation est activée. Pour créer un projet, suivez ces instructions.
Étape 1 : Activez l'API Compute Engine
Accédez à Compute Engine et cliquez sur Activer si ce n'est pas déjà fait.
Étape 2 : Activez l'API Vertex AI
Accédez à la section Vertex AI de Cloud Console, puis cliquez sur Activer l'API Vertex AI.
Étape 3 : créez une instance Vertex AI Workbench
Dans la section Vertex AI de Cloud Console, cliquez sur Workbench :
Activez l'API Notebooks si ce n'est pas déjà fait.
Une fois l'API activée, cliquez sur NOTEBOOKS GÉRÉS :
Sélectionnez ensuite NOUVEAU NOTEBOOK.
Attribuez un nom à votre notebook, puis cliquez sur Paramètres avancés.
Sous "Paramètres avancés", activez l'arrêt en cas d'inactivité et définissez le nombre de minutes sur 60. Cela entraîne l'arrêt automatique du notebook lorsqu'il n'est pas utilisé. Vous ne payez donc pas de frais inutiles.
Étape 4 : Ouvrez votre notebook
Une fois l'instance créée, sélectionnez Ouvrir JupyterLab.
Étape 5 : Authentifiez-vous (première fois uniquement)
La première fois que vous utilisez une nouvelle instance, vous êtes invité à vous authentifier. Pour ce faire, suivez les étapes indiquées dans l'interface utilisateur.
Étape 6 : Sélectionnez le kernel approprié
Les notebooks gérés fournissent plusieurs kernels dans une seule interface utilisateur. Sélectionnez le kernel pour TensorFlow 2 (local).
5. Étapes de configuration initiale dans votre notebook
Vous devrez effectuer une série d'étapes supplémentaires pour configurer votre environnement dans votre notebook avant de créer votre pipeline. Ces étapes incluent l'installation de packages supplémentaires, la définition de variables, la création de votre bucket Cloud Storage, la copie de l'ensemble de données de jeu à partir d'un bucket de stockage public, l'importation de bibliothèques et la définition de constantes supplémentaires.
Étape 1 : Installez des packages supplémentaires
Nous devons installer des dépendances de packages supplémentaires qui ne sont pas actuellement installées dans votre environnement de notebook. Par exemple, le SDK KFP.
!pip3 install --user --force-reinstall 'google-cloud-aiplatform>=1.15' -q --no-warn-conflicts
!pip3 install --user kfp -q --no-warn-conflicts
Vous devez ensuite redémarrer le kernel du notebook pour pouvoir utiliser les packages téléchargés dans votre 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)
Étape 2 : Définissez des variables
Nous voulons définir notre PROJECT_ID. Si vous ne connaissez pas votre Project_ID, vous pourrez peut-être l'obtenir à l'aide de gcloud.PROJECT_ID
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)
Sinon, définissez votre PROJECT_ID ici.
if PROJECT_ID == "" or PROJECT_ID is None:
PROJECT_ID = "[your-project-id]" # @param {type:"string"}
Nous devons également définir la variable REGION, qui est utilisée dans le reste de ce notebook. Vous trouverez ci-dessous les régions compatibles avec Vertex AI. Nous vous recommandons de choisir la région la plus proche de vous.
- Amériques : us-central1
- Europe: europe-west4
- Asie-Pacifique : asia-east1
N'utilisez pas de bucket multirégional pour l'entraînement avec Vertex AI. Toutes les régions ne sont pas compatibles avec les services Vertex AI. Apprenez-en plus sur les régions Vertex AI.
#set your region
REGION = "us-central1" # @param {type: "string"}
Enfin, nous allons définir une variable TIMESTAMP. Cette variable permet d'éviter les conflits de noms entre utilisateurs sur les ressources créées. Vous créez un TIMESTAMP pour chaque session d'instance et l'ajoutez au nom des ressources créées dans ce tutoriel.
#set timestamp to avoid collisions between multiple users
from datetime import datetime
TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")
Étape 3 : Créez un bucket Cloud Storage
Vous devez spécifier et exploiter un bucket de préproduction Cloud Storage. Le bucket de préproduction est l'endroit où toutes les données associées à votre ensemble de données et à vos ressources de modèle sont conservées entre les sessions.
Définissez le nom de votre bucket Cloud Storage ci-dessous. Les noms de bucket doivent être uniques sur l'ensemble des projets Google Cloud, y compris les projets à l'extérieur de votre organisation.
#set cloud storage bucket
BUCKET_NAME = "[insert bucket name here]" # @param {type:"string"}
BUCKET_URI = f"gs://{BUCKET_NAME}"
Si votre bucket N'EXISTE PAS déjà, vous pouvez exécuter la cellule suivante pour créer votre bucket Cloud Storage.
! gsutil mb -l $REGION -p $PROJECT_ID $BUCKET_URI
Vous pouvez ensuite vérifier l'accès à votre bucket Cloud Storage en exécutant la cellule suivante.
#verify access
! gsutil ls -al $BUCKET_URI
Étape 4 : Copiez notre ensemble de données de jeu
Comme mentionné précédemment, vous allez utiliser un ensemble de données de jeu populaire provenant du jeu vidéo à succès FIFA d'EA Sports. Nous avons effectué le prétraitement pour vous. Vous n'avez donc qu'à copier l'ensemble de données à partir du bucket de stockage public et à le déplacer vers celui que vous avez créé.
# 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
Étape 5 : Importez des bibliothèques et définissez des constantes supplémentaires
Ensuite, nous allons importer nos bibliothèques pour Vertex AI, KFP, etc.
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
Nous allons également définir des constantes supplémentaires auxquelles nous ferons référence dans le reste du notebook, telles que le ou les chemins d'accès à nos données d'entraînement.
#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. Créons notre pipeline
Maintenant, le plaisir peut commencer et nous pouvons commencer à exploiter Vertex AI pour créer notre pipeline d'entraînement. Nous allons initialiser le SDK Vertex AI, configurer notre job d'entraînement en tant que composant de pipeline, créer notre pipeline, envoyer nos exécutions de pipeline et exploiter le SDK Vertex AI pour afficher les tests et surveiller leur état.
Étape 1 : Initialisez le SDK Vertex AI
Initialisez le SDK Vertex AI en définissant votre PROJECT_ID et votre BUCKET_URI.
#initialize vertex AI SDK
vertex_ai.init(project=PROJECT_ID, staging_bucket=BUCKET_URI)
Étape 2 : Configurez notre job d'entraînement en tant que composant de pipeline
Pour commencer à exécuter nos tests, nous devons spécifier notre job d'entraînement en le définissant comme composant de pipeline. Notre pipeline prendra en entrée des données d'entraînement et des hyperparamètres (par exemple, DROPOUT_RATE, LEARNING_RATE, EPOCHS) et générera des métriques de modèle (par exemple, MAE et RMSE) et un artefact de modèle.
@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
Étape 3 : Créez notre pipeline
Nous allons maintenant configurer notre workflow à l'aide du Domain Specific Language (DSL) disponible dans KFP et compiler notre pipeline dans un fichier 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")
Étape 4 : Envoyez nos exécutions de pipeline
Le travail difficile est terminé : nous avons configuré notre composant et défini notre pipeline. Nous sommes prêts à envoyer différentes exécutions du pipeline que nous avons spécifié ci-dessus. Pour ce faire, nous devons définir les valeurs de nos différents hyperparamètres comme suit :
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},
]
Une fois les hyperparamètres définis, nous pouvons exploiter une for loop pour alimenter correctement les différentes exécutions du pipeline :
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)
Étape 5 : Exploitez le SDK Vertex AI pour afficher les tests
Le SDK Vertex AI vous permet de surveiller l'état des exécutions de pipeline. Vous pouvez également l'utiliser pour renvoyer les paramètres et les métriques des exécutions de pipeline dans le test Vertex AI. Utilisez le code suivant pour afficher les paramètres associés à vos exécutions et leur état actuel.
# see state/status of all the pipeline runs
vertex_ai.get_experiment_df(EXPERIMENT_NAME)
Vous pouvez exploiter le code ci-dessous pour obtenir des mises à jour sur l'état de vos exécutions de pipeline.
#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)
Vous pouvez également appeler des jobs de pipeline spécifiques à l'aide de 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())
Enfin, vous pouvez actualiser l'état de vos exécutions à intervalles définis (par exemple, toutes les 60 secondes) pour voir les états passer de RUNNING à FAILED ou COMPLETE.
# wait 60 seconds and view state again
import time
time.sleep(60)
vertex_ai.get_experiment_df(EXPERIMENT_NAME)
7. Identifiez l'exécution la plus performante
Très bien, nous avons maintenant les résultats de nos exécutions de pipeline. Vous vous demandez peut-être ce que vous pouvez tirer de ces résultats. La sortie de vos tests doit contenir cinq lignes, une pour chaque exécution du pipeline. Elle se présentera comme suit :
La MAE et la RMSE sont des mesures de l'erreur de prédiction moyenne du modèle. Une valeur inférieure pour les deux métriques est donc souhaitable dans la plupart des cas. D'après la sortie de Vertex AI Experiments, l'exécution la plus réussie pour les deux métriques est l'exécution finale avec un dropout_rate de 0,001, un learning_rate de 0,001 et un nombre total d'epochs de 20. D'après ce test, ces paramètres de modèle seront finalement utilisés en production, car ils offrent les meilleures performances de modèle.
Voilà, vous avez terminé l'atelier !
🎉 Félicitations ! 🎉
Vous savez désormais utiliser Vertex AI pour :
- Entraîner un modèle Keras personnalisé pour prédire les notes des joueurs (par exemple, une régression)
- Utiliser le SDK Kubeflow Pipelines pour créer des pipelines de ML évolutifs
- Créer et exécuter un pipeline en cinq étapes qui ingère des données depuis GCS, les met à l'échelle, entraîne le modèle, l'évalue et enregistre le modèle obtenu dans GCS
- Exploiter Vertex ML Metadata pour enregistrer des artefacts de modèle tels que des modèles et des métriques de modèle
- Utiliser Vertex AI Experiments pour comparer les résultats des différentes exécutions de pipeline
Pour en savoir plus sur les différents composants de Vertex, consultez la documentation.
8. Nettoyage
Pour éviter que des frais ne vous soient facturés, nous vous recommandons de supprimer les ressources créées tout au long de cet atelier.
Étape 1 : Arrêtez ou supprimez votre instance de notebooks
Si vous souhaitez continuer à utiliser le notebook que vous avez créé dans cet atelier, nous vous recommandons de le désactiver quand vous ne vous en servez pas. À partir de l'interface utilisateur de Notebooks dans la console Cloud, sélectionnez le notebook et cliquez sur Arrêter. Si vous souhaitez supprimer l'instance définitivement, sélectionnez Supprimer :
Étape 2 : Supprimez votre bucket Cloud Storage
Pour supprimer le bucket de stockage, utilisez le menu de navigation de la console Cloud pour accéder à Stockage, sélectionnez votre bucket puis cliquez sur "Supprimer" :
