Complete Vertex AI Tutorial: Unified ML Platform on Google Cloud

Vertex AI is Google Cloud's unified machine learning platform that brings together all Google Cloud's ML services. It provides tools for building, deploying, and scaling ML models with AutoML and custom training.

Why Vertex AI?

• Unified platform: All ML tools in one place
• AutoML: No-code model building
• Custom training: Full control with custom code
• MLOps: Built-in pipelines and monitoring
• Scalable: Enterprise-grade infrastructure

• Datasets
• Training (AutoML and Custom)
• Model Registry
• Endpoints
• Pipelines
• Feature Store
• Experiments

Prerequisites

pip install google-cloud-aiplatform

Authenticate
gcloud auth login
gcloud config set project your-project-id

Setup

1. Initialize Vertex AI

from google.cloud import aiplatform

aiplatform.init(
    project="your-project-id",
    location="us-central1",
    stagingbucket="gs://your-bucket"

)

gcloud services enable aiplatform.googleapis.com
gcloud services enable compute.googleapis.com
gcloud services enable storage.googleapis.com

from google.cloud import aiplatform

Create from BigQuery
dataset = aiplatform.TabularDataset.create(
    displayname="customer-churn-dataset",
    bqsource="bq://project.dataset.table"

)

Create from GCS
dataset = aiplatform.TabularDataset.create(
    displayname="customer-churn-dataset",
    gcssource="gs://bucket/data/train.csv"

)

print(f"Dataset created: {dataset.resourcename}")

2. Create Image Dataset

# Create image dataset
imagedataset = aiplatform.ImageDataset.create(

    displayname="product-images",
    gcssource="gs://bucket/images/",

    importschemauri=aiplatform.schema.dataset.ioformat.image.singlelabelclassification

)

# Create text dataset
textdataset = aiplatform.TextDataset.create(
    displayname="sentiment-dataset",

    gcssource="gs://bucket/text/data.jsonl",
    importschemauri=aiplatform.schema.dataset.ioformat.text.singlelabelclassification
)

AutoML Training

1. AutoML Tabular

# Create AutoML tabular training job
job = aiplatform.AutoMLTabularTrainingJob(
    displayname="churn-automl",

    optimizationpredictiontype="classification",

    optimizationobjective="maximize-au-roc"
)

Train model
model = job.run(
    dataset=dataset,
    targetcolumn="churn",

    trainingfractionsplit=0.8,

    validationfractionsplit=0.1,

    testfractionsplit=0.1,

    budgetmillinodehours=1000,
    modeldisplayname="churn-model"
)

print(f"Model trained: {model.resourcename}")

# Create AutoML image training job
job = aiplatform.AutoMLImageTrainingJob(
    displayname="image-classifier",
    predictiontype="classification",

    multilabel=False
)

Train model
model = job.run(
    dataset=imagedataset,

    trainingfractionsplit=0.8,

    validationfractionsplit=0.1,

    testfractionsplit=0.1,

    budgetmillinodehours=8000,
    modeldisplayname="product-classifier"
)

3. AutoML Text Classification

# Create AutoML text training job
job = aiplatform.AutoMLTextTrainingJob(
    displayname="sentiment-classifier",

    predictiontype="classification",
    multilabel=False

)

Train model
model = job.run(
    dataset=textdataset,
    trainingfractionsplit=0.8,
    validationfractionsplit=0.1,
    testfractionsplit=0.1,
    modeldisplayname="sentiment-model"
)

Custom Training

1. Custom Training Job

from google.cloud import aiplatform

Define custom training job
job = aiplatform.CustomTrainingJob(
    displayname="custom-sklearn-training",

    scriptpath="train.py",
    containeruri="us-docker.pkg.dev/vertex-ai/training/sklearn-cpu.1-0:latest",

    requirements=["pandas", "scikit-learn"],
    modelservingcontainerimageuri="us-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.1-0:latest"

)

Run training
model = job.run(
    dataset=dataset,
    modeldisplayname="sklearn-model",

    machinetype="n1-standard-4",
    replicacount=1,

    args=["--epochs", "100", "--batch-size", "32"]
)

# train.py
import argparse
import os
import pandas as pd
from sklearn.ensemble import RandomForestClassifier
from sklearn.modelselection import traintestsplit
from sklearn.metrics import accuracyscore

import joblib
from google.cloud import storage

def main():
    parser = argparse.ArgumentParser()
    parser.addargument("--epochs", type=int, default=100)
    parser.addargument("--batch-size", type=int, default=32)

    args = parser.parseargs()

    # Load data from environment variable
    trainingdatauri = os.environ.get("AIPTRAININGDATAURI")

    df = pd.readcsv(trainingdatauri)

    X = df.drop("target", axis=1)
    y = df["target"]

    Xtrain, Xtest, ytrain, ytest = traintestsplit(
        X, y, testsize=0.2, randomstate=42
    )

    # Train model
    model = RandomForestClassifier(nestimators=args.epochs)

    model.fit(Xtrain, ytrain)


    # Evaluate
    predictions = model.predict(Xtest)
    accuracy = accuracyscore(ytest, predictions)
    print(f"Accuracy: {accuracy}")

    # Save model
    modeldir = os.environ.get("AIPMODELDIR")

    joblib.dump(model, os.path.join(modeldir, "model.joblib"))

if name == "main":
    main()

3. Custom Container Training

# Define custom container training job
job = aiplatform.CustomContainerTrainingJob(
    displayname="pytorch-training",

    containeruri="gcr.io/your-project/pytorch-training:latest",
    modelservingcontainerimageuri="gcr.io/your-project/pytorch-serving:latest"
)

Run training
model = job.run(
    dataset=dataset,
    modeldisplayname="pytorch-model",
    machinetype="n1-standard-8",

    acceleratortype="NVIDIATESLAV100",
    acceleratorcount=1,

    replicacount=1
)

4. Dockerfile for Custom Container

FROM pytorch/pytorch:1.9.0-cuda10.2-cudnn7-runtime

WORKDIR /app

COPY requirements.txt .
RUN pip install -r requirements.txt

COPY train.py .

ENTRYPOINT ["python", "train.py"]

Hyperparameter Tuning

1. Define Hyperparameter Tuning Job

from google.cloud import aiplatform
from google.cloud.aiplatform import hyperparametertuning as hpt


Define hyperparameter spec
parameterspec = {
    "learningrate": hpt.DoubleParameterSpec(min=0.001, max=0.1, scale="log"),

    "numlayers": hpt.DiscreteParameterSpec(values=[2, 4, 6, 8], scale="linear"),
    "dropout": hpt.DoubleParameterSpec(min=0.0, max=0.5, scale="linear")
}

Define metric spec
metricspec = {"accuracy": "maximize"}


Create hyperparameter tuning job
job = aiplatform.HyperparameterTuningJob(
    displayname="hpt-training",
    customjob=aiplatform.CustomJob(

        displayname="hpt-custom-job",
        workerpoolspecs=[{
            "machinespec": {"machinetype": "n1-standard-4"},
            "replicacount": 1,

            "containerspec": {
                "imageuri": "gcr.io/your-project/training:latest",

                "args": []
            }
        }]
    ),
    metricspec=metricspec,

    parameterspec=parameterspec,

    maxtrialcount=20,

    paralleltrialcount=4,

    searchalgorithm="random"
)

Run job
job.run()

Model Registry

1. Upload Model

# Upload model to registry
model = aiplatform.Model.upload(
    displayname="sklearn-model",

    artifacturi="gs://bucket/models/sklearn/",
    servingcontainerimageuri="us-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.1-0:latest"

)

print(f"Model uploaded: {model.resourcename}")

2. List Models

# List all models
models = aiplatform.Model.list()
for m in models:
    print(f"{m.displayname}: {m.resourcename}")

Get specific model
model = aiplatform.Model("projects/123/locations/us-central1/models/456")
print(f"Model: {model.displayname}")

# Upload new version
newversion = aiplatform.Model.upload(
    displayname="sklearn-model",

    artifacturi="gs://bucket/models/sklearn-v2/",
    servingcontainerimageuri="us-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.1-0:latest",

    parentmodel=model.resourcename

)

# Create endpoint
endpoint = aiplatform.Endpoint.create(
    displayname="prediction-endpoint",
    description="Production endpoint for predictions"
)

print(f"Endpoint created: {endpoint.resourcename}")

# Deploy model to endpoint
endpoint.deploy(
    model=model,
    deployedmodeldisplayname="sklearn-deployed",
    machinetype="n1-standard-4",

    minreplicacount=1,

    maxreplicacount=5,

    trafficpercentage=100,
    sync=True
)

print("Model deployed")

3. Make Predictions

# Online prediction
instances = [
    {"feature1": 1.0, "feature2": 2.0, "feature3": 3.0},
    {"feature1": 4.0, "feature2": 5.0, "feature3": 6.0}
]

predictions = endpoint.predict(instances=instances)
print(f"Predictions: {predictions.predictions}")

4. Batch Prediction

# Create batch prediction job
batchjob = model.batchpredict(
    jobdisplayname="batch-prediction",
    gcssource="gs://bucket/batch-input/",

    gcsdestinationprefix="gs://bucket/batch-output/",

    machinetype="n1-standard-4",
    startingreplicacount=2,
    maxreplicacount=10
)

batchjob.wait()

print(f"Batch prediction complete: {batchjob.outputinfo}")

# Initialize experiment
aiplatform.init(experiment="my-experiment")

Start run
aiplatform.startrun("run-1")

Log parameters
aiplatform.logparams({

    "learningrate": 0.01,
    "epochs": 100,
    "batchsize": 32

})

Log metrics
aiplatform.logmetrics({
    "accuracy": 0.95,
    "f1score": 0.93

})

End run
aiplatform.endrun()

2. Compare Runs

# Get experiment
experiment = aiplatform.Experiment("my-experiment")

Get all runs
runsdf = experiment.getdataframe()

print(runsdf[["runname", "accuracy", "learningrate"]])

Best Practices

1. Resource Management

# Use context manager for cleanup
with aiplatform.init(project="project", location="us-central1"):
    # Your ML code here
    pass

Delete resources when done
endpoint.undeployall()

endpoint.delete()
model.delete()

# Use preemptible VMs for training
job = aiplatform.CustomTrainingJob(
    displayname="cost-optimized-training",
    scriptpath="train.py",

    containeruri="training-container"
)

model = job.run(
    machinetype="n1-standard-4",

    replicacount=1,
    bootdisktype="pd-ssd",
    bootdisksize_gb=100
)

Conclusion

Vertex AI provides:

Key takeaways:

• Use AutoML for quick prototyping
• Use custom training for control
• Register models for versioning
• Deploy to managed endpoints
• Track experiments systematically