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. Nous utiliserons 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, 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 à partir de 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 (Expériences), Pipelines (Pipelines), ML Metadata (Métadonnées de ML) et Workbench.
3. Présentation du cas d'utilisation
Nous utiliserons un ensemble de données de football populaire provenant de la série de jeux vidéo FIFA d'EA Sports. 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 vous permettre de vous lancer plus facilement. Tout au long de l'atelier, vous utiliserez cet ensemble de données, qui se trouve désormais dans un bucket Cloud Storage public. Nous vous expliquerons plus en détail comment accéder au jeu de données plus tard dans l'atelier de programmation. Notre objectif final est de prédire la note globale d'un joueur en fonction de diverses actions dans le jeu, comme les interceptions et les pénalités.
Pourquoi les tests Vertex AI sont-ils utiles pour la data science ?
La data science est par nature expérimentale. Après tout, les data scientists sont des scientifiques. Les bons data scientists sont guidés par des hypothèses et utilisent la méthode d'essai et d'erreur pour tester différentes hypothèses dans l'espoir que les itérations successives aboutissent à un modèle plus performant.
Bien que les équipes de data science aient adopté l'expérimentation, elles ont souvent du mal à suivre leur travail et la "sauce secrète" révélée par leurs efforts d'expérimentation. Plusieurs raisons peuvent expliquer cette situation:
- Suivre les tâches d'entraînement peut devenir fastidieux. Il est donc facile de perdre de vue ce qui fonctionne et ce qui ne fonctionne pas.
- Ce problème s'aggrave lorsque vous examinez une équipe de data science, car tous les membres ne suivent pas nécessairement 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 donnent lieu à des informations incohérentes et incomplètes à partir desquelles tirer des leçons.
Résumé:Vertex AI Experiments vous fait gagner du temps en vous aidant à suivre et à comparer plus facilement vos expériences.
Pourquoi utiliser les tests Vertex AI pour les jeux vidéo ?
Les jeux vidéo ont toujours été un terrain d'expérimentation pour le machine learning. Les jeux génèrent des milliards d'événements en temps réel chaque jour, mais ils exploitent toutes ces données à l'aide d'expériences de ML et de ML pour améliorer les expériences en jeu, retenir les joueurs et évaluer les différents joueurs de leur plate-forme. Nous avons donc pensé qu'un ensemble de données sur les jeux vidéo convenait parfaitement à notre exercice d'expérimentation 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 noyaux dans une seule interface utilisateur. Sélectionnez le kernel pour Tensorflow 2 (local).
5. Étapes de configuration initiale dans votre notebook
Vous devrez suivre 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: Installer des packages supplémentaires
Nous allons devoir installer des dépendances de paquets supplémentaires qui ne sont pas actuellement installées dans votre environnement de notebook. Un exemple inclut 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éfinir des variables
Nous voulons définir notre PROJECT_ID. Si vous ne connaissez pas votre Project_ID, vous pouvez peut-être obtenir votre PROJECT_ID à l'aide de 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)
Sinon, définissez votre PROJECT_ID ici.
if PROJECT_ID == "" or PROJECT_ID is None:
PROJECT_ID = "[your-project-id]" # @param {type:"string"}
Nous allons également définir la variable REGION, qui sera 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
Veuillez ne pas utiliser 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 devrez spécifier et exploiter un bucket de préproduction Cloud Storage. C'est dans le bucket de préproduction que toutes les données associées à votre ensemble de données et aux ressources de votre modèle sont conservées d'une session à l'autre.
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 le créer.
! 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 sur les jeux
Comme indiqué précédemment, vous allez exploiter un ensemble de données populaire issu des jeux vidéo à succès d'EA Sports, FIFA. Nous avons effectué le prétraitement pour vous. Il vous suffit donc de copier l'ensemble de données à partir du bucket de stockage public et de 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: Importer des bibliothèques et définir des constantes supplémentaires
Nous allons ensuite 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 tout au long du notebook, comme 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
Nous pouvons maintenant 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 utiliser le SDK Vertex AI pour afficher les tests et surveiller leur état.
Étape 1: Initialiser le SDK Vertex AI
Initialisez le SDK Vertex AI en définissant vos PROJECT_ID et BUCKET_URI.
#initialize vertex AI SDK
vertex_ai.init(project=PROJECT_ID, staging_bucket=BUCKET_URI)
Étape 2: Configurer notre job d'entraînement en tant que composant de pipeline
Pour commencer à exécuter nos tests, nous devons spécifier notre tâche d'entraînement en la définissant comme un composant de pipeline. Notre pipeline recevra les données d'entraînement et les hyperparamètres (par exemple, DROPOUT_RATE, LEARNING_RATE, EPOCHS) en tant qu'entrées et métriques de sortie du 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éer notre pipeline
Nous allons maintenant configurer notre workflow à l'aide de 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 l'exécution de votre pipeline
Le travail le plus difficile est fait : 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ées 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 utiliser un for loop pour alimenter 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: Utiliser 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 utiliser le code ci-dessous pour obtenir des informations sur l'état d'exécution de votre 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 tâches 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 à des 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. Identifier la campagne la plus performante
Parfait, nous avons maintenant les résultats de nos exécutions de pipeline. Vous vous demandez peut-être ce que vous pouvez apprendre des résultats. La sortie de vos tests doit contenir cinq lignes, une pour chaque exécution du pipeline. L'URL doit ressembler à ceci:
L'EAM et la RMSE mesurent l'erreur de prédiction moyenne du modèle. Dans la plupart des cas, il est donc préférable d'obtenir une valeur plus faible pour les deux métriques. 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 de 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.
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éez et exécutez un pipeline en cinq étapes qui ingère des données à partir de 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 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 définitivement l'instance, sélectionnez Delete (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" :
