1. Panoramica
In questo lab utilizzerai Vertex AI per creare una pipeline che addestra un modello Keras personalizzato in TensorFlow. Useremo poi le nuove funzionalità disponibili in Vertex AI Experiments per monitorare e confrontare le esecuzioni del modello al fine di identificare quale combinazione di iperparametri genera le prestazioni migliori.
Cosa imparerai
Imparerai a:
- Addestrare un modello Keras personalizzato per prevedere le valutazioni dei giocatori (ad es. regressione)
- Utilizzare l'SDK Kubeflow Pipelines per creare pipeline ML scalabili
- Creare ed eseguire una pipeline in 5 passaggi che importa i dati da Cloud Storage, scala i dati, addestra il modello, li valuta e salva il modello risultante in Cloud Storage
- Sfrutta Vertex ML Metadata per salvare artefatti del modello come modelli e metriche del modello
- Utilizza Vertex AI Experiments per confrontare i risultati delle varie esecuzioni della pipeline
Il costo totale per l'esecuzione di questo lab su Google Cloud è di circa $1.
2. Introduzione a Vertex AI
Questo lab utilizza la più recente offerta di prodotti AI disponibile su Google Cloud. Vertex AI integra le offerte ML di Google Cloud in un'esperienza di sviluppo fluida. In precedenza, i modelli addestrati con AutoML e i modelli personalizzati erano accessibili tramite servizi separati. La nuova offerta combina entrambi in un'unica API, insieme ad altri nuovi prodotti. Puoi anche migrare progetti esistenti su Vertex AI.
Vertex AI include molti prodotti diversi per supportare i flussi di lavoro ML end-to-end. Questo lab si concentra sui prodotti evidenziati di seguito: Esperimenti, Pipeline, Metadati ML e Workbench
3. Panoramica del caso d'uso
Utilizzeremo un noto set di dati sul calcio proveniente da EA Sports serie di videogiochi FIFA. Comprende oltre 25.000 partite di calcio e più di 10.000 giocatori per le stagioni 2008-2016. I dati sono stati pre-elaborati in modo che tu possa iniziare più facilmente a lavorare in modo efficace. Durante il lab utilizzerai questo set di dati, che ora si trova in un bucket Cloud Storage pubblico. Forniremo maggiori dettagli su come accedere al set di dati più avanti nel codelab. Il nostro obiettivo finale è prevedere la valutazione complessiva di un giocatore in base a varie azioni durante il gioco come intercette e rigori.
Perché Vertex AI Experiments è utile per la data science?
La data science è di natura sperimentale: dopotutto, sono chiamati scienziati. I buoni data scientist si basano su ipotesi e utilizzano tentativi ed errori per testare varie ipotesi nella speranza che iterazioni successive si traducano in un modello più efficiente.
Sebbene i team di data science abbiano abbracciato la sperimentazione, spesso faticano a tenere traccia del proprio lavoro e del "segreto" che è stato scoperto attraverso i loro sforzi di sperimentazione. Ciò accade per diversi motivi:
- Monitorare i lavori della formazione può diventare complicato, così è facile perdere di vista ciò che funziona e ciò che non lo è
- Il problema si aggrava quando esamini un team di data science, in quanto non tutti i membri potrebbero monitorare gli esperimenti o persino condividere i loro risultati con altri.
- L'acquisizione dei dati richiede molto tempo e la maggior parte dei team utilizza metodi manuali (ad es.fogli o documenti) che generano informazioni incoerenti e incomplete da cui imparare.
Il tl;dr: Vertex AI Experiments fa il lavoro per te, aiutandoti a monitorare e confrontare più facilmente i tuoi esperimenti
Perché Vertex AI Experiments for Gaming?
Storicamente, i giochi sono stati un parco giochi per il machine learning e gli esperimenti di ML. I giochi non solo producono miliardi di eventi in tempo reale al giorno, ma sfruttano tutti questi dati sfruttando gli esperimenti di ML e ML per migliorare le esperienze in-game, fidelizzare i giocatori e valutare i diversi giocatori sulla loro piattaforma. Pertanto, abbiamo pensato che un set di dati per i giochi si adattasse bene all'esercizio complessivo degli esperimenti.
4. Configura l'ambiente
Per eseguire questo codelab, è necessario un progetto Google Cloud con fatturazione abilitata. Per creare un progetto, segui le istruzioni riportate qui.
Passaggio 1: abilita l'API Compute Engine
Passa a Compute Engine e seleziona Abilita se non è già abilitato.
Passaggio 2: abilita l'API Vertex AI
Vai alla sezione Vertex AI della tua console Cloud e fai clic su Abilita API Vertex AI.
Passaggio 3: crea un'istanza di Vertex AI Workbench
Dalla sezione Vertex AI della console Cloud, fai clic su Workbench:
Abilita l'API Notebooks, se non lo è già.
Dopo averla attivata, fai clic su NOTEBOOK GESTITI:
Quindi seleziona NUOVO Blocco note.
Assegna un nome al blocco note, quindi fai clic su Impostazioni avanzate.
In Impostazioni avanzate, abilita l'arresto per inattività e imposta il numero di minuti su 60. Ciò significa che il blocco note si arresterà automaticamente quando non è in uso, per evitare costi inutili.
Passaggio 4: apri il blocco note
Una volta creata l'istanza, seleziona Apri JupyterLab.
Passaggio 5: esegui l'autenticazione (solo la prima volta)
La prima volta che utilizzi una nuova istanza, ti verrà chiesto di autenticarti. Per farlo, segui i passaggi nell'interfaccia utente.
Passaggio 6: seleziona il kernel appropriato
I blocchi note gestiti forniscono più kernel in un'unica interfaccia utente. Seleziona il kernel per TensorFlow 2 (locale).
5. Passaggi di configurazione iniziali nel blocco note
Prima di creare la pipeline, dovrai eseguire una serie di passaggi aggiuntivi per configurare l'ambiente all'interno del blocco note. Questi passaggi includono l'installazione di eventuali pacchetti aggiuntivi, l'impostazione di variabili, la creazione del bucket Cloud Storage, la copia del set di dati per videogiochi da un bucket di archiviazione pubblico, l'importazione di librerie e la definizione di costanti aggiuntive.
Passaggio 1: installa pacchetti aggiuntivi
Dovremo installare ulteriori dipendenze del pacchetto non attualmente installate nell'ambiente del blocco note. Un esempio include l'SDK KFP.
!pip3 install --user --force-reinstall 'google-cloud-aiplatform>=1.15' -q --no-warn-conflicts
!pip3 install --user kfp -q --no-warn-conflicts
Quindi vorrai riavviare il kernel del blocco note in modo da poter utilizzare i pacchetti scaricati all'interno del blocco note.
# 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)
Passaggio 2. Imposta le variabili
Vogliamo definire i nostri PROJECT_ID. Se non conosci Project_ID, potresti riuscire a recuperare PROJECT_ID utilizzando 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)
Altrimenti, imposta qui il tuo PROJECT_ID.
if PROJECT_ID == "" or PROJECT_ID is None:
PROJECT_ID = "[your-project-id]" # @param {type:"string"}
Dovremmo anche impostare la variabile REGION, che verrà utilizzata nel resto del blocco note. Di seguito sono riportate le regioni supportate per Vertex AI. Ti consigliamo di scegliere la regione più vicina a te.
- Americhe: us-central1
- Europa: europe-west4
- Asia Pacifico: asia-east1
Non utilizzare un bucket multiregionale per l'addestramento con Vertex AI. Non tutte le regioni forniscono supporto per tutti i servizi Vertex AI. Scopri di più sulle regioni Vertex AI.
#set your region
REGION = "us-central1" # @param {type: "string"}
Infine, impostiamo una variabile TIMESTAMP. Questa variabile viene utilizzata per evitare conflitti tra i nomi tra gli utenti nelle risorse create. Devi creare un valore TIMESTAMP per ogni sessione dell'istanza e aggiungerlo al nome delle risorse create in questo tutorial.
#set timestamp to avoid collisions between multiple users
from datetime import datetime
TIMESTAMP = datetime.now().strftime("%Y%m%d%H%M%S")
Passaggio 3: crea un bucket Cloud Storage
Dovrai specificare e utilizzare un bucket temporaneo Cloud Storage. Il bucket temporaneo è il luogo in cui tutti i dati associati al set di dati e alle risorse del modello vengono conservati per più sessioni.
Imposta il nome del tuo bucket Cloud Storage di seguito. I nomi dei bucket devono essere univoci a livello globale in tutti i progetti Google Cloud, inclusi quelli esterni alla tua organizzazione.
#set cloud storage bucket
BUCKET_NAME = "[insert bucket name here]" # @param {type:"string"}
BUCKET_URI = f"gs://{BUCKET_NAME}"
Se il bucket NON esiste già, puoi eseguire la cella seguente per creare il tuo bucket Cloud Storage.
! gsutil mb -l $REGION -p $PROJECT_ID $BUCKET_URI
Puoi quindi verificare l'accesso al bucket Cloud Storage eseguendo la cella seguente.
#verify access
! gsutil ls -al $BUCKET_URI
Passaggio 4: copia il nostro set di dati di giochi
Come accennato in precedenza, sfrutterai un famoso set di dati di videogiochi, tra cui FIFA, videogiochi di successo di EA Sports. Abbiamo svolto il lavoro di pre-elaborazione per te, quindi dovrai solo copiare il set di dati dal bucket Cloud Storage pubblico e spostarlo nel bucket che hai creato.
# 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
Passaggio 5: importa le librerie e definisci altre costanti
Ora importeremo le nostre librerie per Vertex AI, KFP e così via.
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
Definiamo inoltre costanti aggiuntive a cui faremo riferimento nel resto del blocco note, come i percorsi dei file per i dati di addestramento.
#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. Costruiamo la nostra pipeline
Ora il divertimento può iniziare e possiamo iniziare a sfruttare Vertex AI per creare la nostra pipeline di addestramento. Inizializziamo l'SDK Vertex AI, configureremo il nostro job di addestramento come componente della pipeline, creeremo la pipeline, invieremo le esecuzioni della pipeline e sfrutteremo l'SDK Vertex AI per visualizzare gli esperimenti e monitorarne lo stato.
Passaggio 1: inizializza l'SDK Vertex AI
Inizializza l'SDK Vertex AI impostando PROJECT_ID e BUCKET_URI.
#initialize vertex AI SDK
vertex_ai.init(project=PROJECT_ID, staging_bucket=BUCKET_URI)
Passaggio 2: configura il nostro job di addestramento come componente della pipeline
Per iniziare a eseguire i nostri esperimenti, dobbiamo specificare il job di addestramento definendolo come componente della pipeline. La nostra pipeline prenderà i dati e gli iperparametri di addestramento (ad es. DROPOUT_RATE, LEARNING_RATE, EPOCHS) come metriche del modello di input e di output (ad es. MAE ed RMSE) e un artefatto del modello.
@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
Passaggio 3: crea la nostra pipeline
Ora configureremo il nostro flusso di lavoro utilizzando i Domain Specific Language (DSL) disponibili in KFP e compileremo la pipeline in un file 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")
Passaggio 4: invia le esecuzioni della pipeline
Il lavoro più impegnativo è la configurazione del componente e la definizione della pipeline. Siamo pronti a inviare varie esecuzioni della pipeline specificate sopra. Per fare ciò, dobbiamo definire i valori dei nostri diversi iperparametri come segue:
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 gli iperparametri definiti, possiamo quindi sfruttare un for loop per alimentare correttamente le diverse esecuzioni della 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)
Passaggio 5: utilizza l'SDK Vertex AI per visualizzare gli esperimenti
L'SDK Vertex AI consente di monitorare lo stato delle esecuzioni della pipeline. Puoi utilizzarlo anche per restituire parametri e metriche delle esecuzioni della pipeline nell'esperimento Vertex AI. Utilizza il seguente codice per visualizzare i parametri associati alle tue esecuzioni e il suo stato attuale.
# see state/status of all the pipeline runs
vertex_ai.get_experiment_df(EXPERIMENT_NAME)
Puoi utilizzare il codice riportato di seguito per ricevere aggiornamenti sullo stato delle esecuzioni della 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)
Puoi anche chiamare job specifici della pipeline utilizzando 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())
Infine, puoi aggiornare lo stato delle esecuzioni a intervalli prestabiliti (ad esempio ogni 60 secondi) per vedere il cambio di stato da 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 l'esecuzione a prestazioni migliori
Bene, ora abbiamo i risultati delle esecuzioni della pipeline. Forse ti starai chiedendo cosa posso imparare dai risultati. L'output degli esperimenti dovrebbe contenere cinque righe, una per ogni esecuzione della pipeline. L'URL avrà il seguente aspetto:
Sia MAE che RMSE sono misure dell'errore medio di previsione del modello, quindi nella maggior parte dei casi è auspicabile un valore più basso per entrambe le metriche. In base all'output di Vertex AI Experiments, possiamo vedere che l'esecuzione di maggior successo su entrambe le metriche è stata l'esecuzione finale con dropout_rate di 0,001, learning_rate se 0,001 e il numero totale di epochs pari a 20. Sulla base di questo esperimento, questi parametri del modello verrebbero utilizzati in ultima analisi per garantire le migliori prestazioni del modello.
Con questo, hai terminato il lab.
🎉 Complimenti! 🎉
Hai imparato come utilizzare Vertex AI per:
- Addestrare un modello Keras personalizzato per prevedere le valutazioni dei giocatori (ad es. regressione)
- Utilizzare l'SDK Kubeflow Pipelines per creare pipeline ML scalabili
- Creare ed eseguire una pipeline in 5 passaggi che importa i dati da GCS, li scala, addestra il modello, li valuta e li salva di nuovo in GCS
- Sfrutta Vertex ML Metadata per salvare artefatti del modello come modelli e metriche del modello
- Utilizza Vertex AI Experiments per confrontare i risultati delle varie esecuzioni della pipeline
Per scoprire di più sulle diverse parti di Vertex, consulta la documentazione.
8. Esegui la pulizia
Per evitare addebiti, ti consigliamo di eliminare le risorse create durante questo lab.
Passaggio 1: arresta o elimina l'istanza di Notebooks
Se vuoi continuare a utilizzare il blocco note creato in questo lab, ti consigliamo di disattivarlo quando non lo usi. Dall'interfaccia utente di Notebooks nella console Cloud, seleziona il blocco note, quindi seleziona Interrompi. Se vuoi eliminare completamente l'istanza, seleziona Elimina:
Passaggio 2: elimina il bucket Cloud Storage
Per eliminare il bucket di archiviazione, utilizzando il menu di navigazione nella console Cloud, vai a Storage, seleziona il bucket e fai clic su Elimina:
