Google Cloud Vertex AI: Complete Guide to ML Operations

Key Features

Feature	Description	Benefits
AutoML	Automated model training	Rapid development
Custom Training	Flexible model development	Full control
Feature Store	Feature management	Reusability
Model Monitoring	Production oversight	Reliability

Implementation Examples

AutoML Vision Model

from google.cloud import aiplatform

def create_automl_training_job(
    project: str,
    display_name: str,
    dataset_id: str,
    location: str = "us-central1",
    model_type: str = "CLOUD",
):
    aiplatform.init(project=project, location=location)
    
    # Get the training dataset
    dataset = aiplatform.ImageDataset(dataset_id)
    
    # Create and run AutoML training job
    job = aiplatform.AutoMLImageTrainingJob(
        display_name=display_name,
        prediction_type="classification",
    )
    
    model = job.run(
        dataset=dataset,
        model_display_name=display_name,
        training_fraction_split=0.8,
        validation_fraction_split=0.1,
        test_fraction_split=0.1,
        budget_milli_node_hours=8000,
    )
    
    return model

Custom Training Pipeline

from google.cloud import aiplatform
from google.cloud.aiplatform import pipeline_jobs

def create_training_pipeline(
    project: str,
    location: str,
    pipeline_name: str,
    training_container_uri: str,
    model_display_name: str,
):
    pipeline_job = pipeline_jobs.PipelineJob(
        display_name=pipeline_name,
        template_path="pipeline.json",
        pipeline_root=f"gs://{project}-pipeline-root",
        parameter_values={
            "project": project,
            "location": location,
            "training_container_uri": training_container_uri,
            "model_display_name": model_display_name,
        },
    )
    
    pipeline_job.run()
    return pipeline_job

Model Development

Feature Engineering

from google.cloud import aiplatform_v1
from google.cloud.aiplatform_v1 import FeaturestoreServiceClient
from google.cloud.aiplatform_v1 import Feature

def create_feature_store(
    project: str,
    location: str,
    featurestore_id: str,
    entity_type_id: str,
):
    client = FeaturestoreServiceClient()
    parent = f"projects/{project}/locations/{location}"
    
    # Create a new feature store
    featurestore = client.create_featurestore(
        parent=parent,
        featurestore_id=featurestore_id,
        featurestore={
            "name": f"{parent}/featurestores/{featurestore_id}",
            "labels": {"environment": "production"},
        },
    )
    
    # Create an entity type
    entity_type = client.create_entity_type(
        parent=featurestore.name,
        entity_type_id=entity_type_id,
        entity_type={
            "description": "Customer features for recommendation model",
        },
    )
    
    # Create features
    features = [
        Feature(
            name="purchase_count",
            value_type=Feature.ValueType.INT64,
            description="Number of purchases in last 30 days",
        ),
        Feature(
            name="total_spend",
            value_type=Feature.ValueType.DOUBLE,
            description="Total spend in last 30 days",
        ),
    ]
    
    for feature in features:
        client.create_feature(
            parent=entity_type.name,
            feature=feature,
            feature_id=feature.name,
        )
    
    return featurestore

Model Training Configuration

Parameter	Value	Description
Machine Type	n1-standard-8	Training compute
GPU Type	NVIDIA_TESLA_T4	Accelerator
Framework	TensorFlow 2.x	ML framework

Model Deployment

Online Prediction Endpoint

from google.cloud import aiplatform

def deploy_model(
    project: str,
    location: str,
    model_id: str,
    machine_type: str = "n1-standard-2",
):
    aiplatform.init(project=project, location=location)
    
    model = aiplatform.Model(model_id)
    
    endpoint = model.deploy(
        machine_type=machine_type,
        min_replica_count=1,
        max_replica_count=3,
        accelerator_type=None,
        accelerator_count=0,
    )
    
    return endpoint

Batch Prediction Job

def create_batch_prediction_job(
    project: str,
    location: str,
    model_id: str,
    gcs_source: str,
    gcs_destination: str,
):
    aiplatform.init(project=project, location=location)
    
    model = aiplatform.Model(model_id)
    
    batch_prediction_job = model.batch_predict(
        job_display_name=f"batch-predict-{model_id}",
        gcs_source=gcs_source,
        gcs_destination_prefix=gcs_destination,
        sync=False,
    )
    
    return batch_prediction_job

Monitoring and Maintenance

Model Monitoring

from google.cloud import aiplatform_v1
from google.cloud.aiplatform_v1 import ModelMonitoringJob

def create_monitoring_job(
    project: str,
    location: str,
    endpoint_id: str,
    display_name: str,
):
    client = aiplatform_v1.ModelMonitoringJobServiceClient()
    parent = f"projects/{project}/locations/{location}"
    
    monitoring_job = ModelMonitoringJob(
        display_name=display_name,
        endpoint=f"{parent}/endpoints/{endpoint_id}",
        monitor_config={
            "feature_monitoring_config": {
                "target_field": "prediction",
                "feature_stats_anomaly_detection": {
                    "enabled": True,
                },
            },
        },
        schedule_config={
            "monitor_interval": {"seconds": 86400},  # Daily monitoring
        },
    )
    
    operation = client.create_model_monitoring_job(
        parent=parent,
        model_monitoring_job=monitoring_job,
    )
    
    return operation.result()

Performance Metrics

Metric	Threshold	Action
Prediction Latency	< 100ms	Scale up
Model Accuracy	> 95%	Retrain
Feature Drift	< 0.1	Investigate

MLOps Best Practices

CI/CD Pipeline

# cloudbuild.yaml
steps:
- name: 'gcr.io/cloud-builders/docker'
  args: ['build', '-t', 'gcr.io/$PROJECT_ID/training', './training']
  
- name: 'gcr.io/cloud-builders/docker'
  args: ['push', 'gcr.io/$PROJECT_ID/training']
  
- name: 'gcr.io/google-cloud-sdk'
  entrypoint: 'gcloud'
  args:
  - 'ai'
  - 'custom-jobs'
  - 'create'
  - '--region=us-central1'
  - '--display-name=training-job'
  - '--config=job_config.yaml'
  
- name: 'gcr.io/google-cloud-sdk'
  entrypoint: 'gcloud'
  args:
  - 'ai'
  - 'models'
  - 'upload'
  - '--region=us-central1'
  - '--display-name=production-model'
  - '--container-image-uri=gcr.io/$PROJECT_ID/prediction'

Model Versioning

def version_model(
    project: str,
    location: str,
    model_id: str,
    version_description: str,
):
    aiplatform.init(project=project, location=location)
    
    model = aiplatform.Model(model_id)
    
    new_version = model.create_version(
        version_description=version_description,
        labels={
            "environment": "production",
            "version": "v1",
        },
    )
    
    return new_version

Troubleshooting Guide

Common Issues

Issue	Cause	Solution
Training Failures	Resource limits	Adjust quotas
Deployment Errors	Dependencies	Check versions
Performance Issues	Resource constraints	Scale resources

References

1. Vertex AI Documentation
2. AutoML Guide
3. Custom Training
4. Model Monitoring
5. MLOps Best Practices
6. Feature Store

Related Posts

• GCP Cloud Run Jobs - Serverless batch processing
• GCP Cost Optimization - Optimize ML costs
• GCP vs AWS - Cloud comparison
• GCP Serverless - Serverless architecture