Google Cloud Vertex AI入門｜SESエンジニア向け機械学習プラットフォーム実践ガイド

「機械学習をプロダクションに導入したいが、インフラの構築が難しい」「MLモデルの開発からデプロイまでのパイプラインを効率化したい」「AI案件に参画するためにGCPのML基盤を学びたい」

Google Cloud Vertex AIは、機械学習モデルの開発・学習・デプロイ・運用を一元管理できる統合MLプラットフォームです。カスタムモデルの学習からAutoML、さらにGeminiなどの生成AIモデルまで、幅広いML/AIワークロードに対応しています。

この記事ではGoogle Cloud完全攻略シリーズEp.13として、Vertex AIの基礎から実践的な活用パターンまでを体系的に解説します。

Vertex AIとは

Vertex AIは、GoogleのML技術を結集した統合プラットフォームです。データサイエンティストからMLエンジニア、アプリケーション開発者まで、それぞれの役割に適したツールを提供します。

Vertex AIの主要コンポーネント

コンポーネント	機能	対象ユーザー
AutoML	ノーコードでモデル構築	ビジネスユーザー・初学者
Custom Training	カスタムモデルの学習	データサイエンティスト
Model Garden	事前学習済みモデルのカタログ	ML/AIエンジニア
Pipelines	MLOpsパイプライン	MLOpsエンジニア
Feature Store	特徴量の一元管理	データエンジニア
Prediction	モデルのデプロイ・推論	バックエンドエンジニア
Gemini API	生成AIモデルのAPI	アプリ開発者全般

開発ワークフロー

データ準備 → モデル開発 → 学習 → 評価 → デプロイ → モニタリング
    │           │         │      │        │           │
  BigQuery   Workbench  Custom/  Model     Endpoint   Model
  Cloud      Notebooks  AutoML   Registry             Monitoring
  Storage

AutoMLで機械学習モデルを構築

テーブルデータの分類モデル

コードを書かずにMLモデルを構築できるAutoMLの実践例：

# gcloudでAutoMLデータセットを作成
gcloud ai datasets create \
  --display-name="customer-churn-prediction" \
  --metadata-schema-uri="gs://google-cloud-aiplatform/schema/dataset/metadata/tabular_1.0.0.yaml" \
  --region=asia-northeast1

# CSVデータをインポート
gcloud ai datasets import \
  --dataset=DATASET_ID \
  --import-schema-uri="gs://google-cloud-aiplatform/schema/dataset/ioformat/tabular_single_line_1.0.0.yaml" \
  --gcs-source="gs://my-bucket/customer_data.csv" \
  --region=asia-northeast1

Python SDKでのAutoML学習

from google.cloud import aiplatform

# Vertex AI初期化
aiplatform.init(
    project="my-project",
    location="asia-northeast1",
    staging_bucket="gs://my-staging-bucket",
)

# データセットの作成
dataset = aiplatform.TabularDataset.create(
    display_name="customer-churn-dataset",
    gcs_source="gs://my-bucket/customer_data.csv",
)

# AutoMLモデルの学習
job = aiplatform.AutoMLTabularTrainingJob(
    display_name="churn-prediction-model",
    optimization_prediction_type="classification",
    column_transformations=[
        {"numeric": {"column_name": "age"}},
        {"numeric": {"column_name": "monthly_charges"}},
        {"numeric": {"column_name": "tenure_months"}},
        {"categorical": {"column_name": "contract_type"}},
        {"categorical": {"column_name": "payment_method"}},
    ],
)

model = job.run(
    dataset=dataset,
    target_column="churn",
    training_fraction_split=0.8,
    validation_fraction_split=0.1,
    test_fraction_split=0.1,
    budget_milli_node_hours=1000,  # 学習時間予算
    model_display_name="churn-prediction-v1",
)

print(f"Model resource name: {model.resource_name}")

カスタムモデルの学習

Vertex AI Workbenchでの開発

# Jupyter Notebook on Vertex AI Workbench

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import classification_report
import joblib

# データ読み込み（BigQueryから直接）
from google.cloud import bigquery
client = bigquery.Client()

query = """
SELECT
  age, monthly_charges, tenure_months,
  contract_type, payment_method,
  churn
FROM `my-project.analytics.customer_data`
WHERE date >= '2025-01-01'
"""
df = client.query(query).to_dataframe()

# 前処理
df_encoded = pd.get_dummies(df, columns=['contract_type', 'payment_method'])
X = df_encoded.drop('churn', axis=1)
y = df_encoded['churn']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# モデル学習
model = GradientBoostingClassifier(
    n_estimators=200,
    max_depth=5,
    learning_rate=0.1,
    random_state=42,
)
model.fit(X_train, y_train)

# 評価
y_pred = model.predict(X_test)
print(classification_report(y_test, y_pred))

# モデル保存
joblib.dump(model, 'model.joblib')

カスタムトレーニングジョブの実行

# カスタムトレーニングをVertex AIで実行
from google.cloud import aiplatform

# トレーニングスクリプト: trainer/train.py
# ↑ 上記のコードをスクリプト化したもの

job = aiplatform.CustomTrainingJob(
    display_name="churn-custom-training",
    script_path="trainer/train.py",
    container_uri="us-docker.pkg.dev/vertex-ai/training/scikit-learn-cpu.1-5:latest",
    requirements=["pandas", "scikit-learn", "google-cloud-bigquery"],
    model_serving_container_image_uri="us-docker.pkg.dev/vertex-ai/prediction/sklearn-cpu.1-5:latest",
)

model = job.run(
    replica_count=1,
    machine_type="n1-standard-4",
    model_display_name="churn-prediction-custom-v1",
)

GPU を使った深層学習

# PyTorchでのカスタムトレーニング
job = aiplatform.CustomTrainingJob(
    display_name="image-classification-training",
    script_path="trainer/train_pytorch.py",
    container_uri="us-docker.pkg.dev/vertex-ai/training/pytorch-gpu.2-2:latest",
    requirements=["torchvision", "timm"],
    model_serving_container_image_uri="us-docker.pkg.dev/vertex-ai/prediction/pytorch-gpu.2-2:latest",
)

model = job.run(
    replica_count=1,
    machine_type="n1-standard-8",
    accelerator_type="NVIDIA_TESLA_T4",
    accelerator_count=1,
    model_display_name="image-classifier-v1",
)

モデルのデプロイと推論

エンドポイントへのデプロイ

# エンドポイント作成
endpoint = aiplatform.Endpoint.create(
    display_name="churn-prediction-endpoint",
    project="my-project",
    location="asia-northeast1",
)

# モデルをデプロイ
model.deploy(
    endpoint=endpoint,
    deployed_model_display_name="churn-v1",
    machine_type="n1-standard-2",
    min_replica_count=1,
    max_replica_count=5,
    traffic_percentage=100,
    # Auto Scaling設定
    autoscaling_target_cpu_utilization=70,
)

# オンライン予測
prediction = endpoint.predict(
    instances=[
        {
            "age": 35,
            "monthly_charges": 75.5,
            "tenure_months": 24,
            "contract_type_Month-to-month": 0,
            "contract_type_One_year": 1,
            "contract_type_Two_year": 0,
            "payment_method_Credit_card": 1,
            "payment_method_Bank_transfer": 0,
        }
    ]
)

print(f"Churn probability: {prediction.predictions[0]}")

バッチ予測

# 大量データの一括予測
batch_job = model.batch_predict(
    job_display_name="churn-batch-prediction",
    gcs_source="gs://my-bucket/batch_input.jsonl",
    gcs_destination_prefix="gs://my-bucket/predictions/",
    machine_type="n1-standard-4",
    starting_replica_count=2,
    max_replica_count=10,
)

batch_job.wait()
print(f"Batch prediction output: {batch_job.output_info}")

Vertex AI Pipelines（MLOps）

パイプラインの構築

from kfp.v2 import dsl, compiler
from kfp.v2.dsl import component, Input, Output, Dataset, Model, Metrics
from google.cloud import aiplatform

@component(
    packages_to_install=["pandas", "google-cloud-bigquery"],
    base_image="python:3.11",
)
def extract_data(
    query: str,
    output_dataset: Output[Dataset],
):
    """BigQueryからデータを抽出"""
    from google.cloud import bigquery
    import pandas as pd

    client = bigquery.Client()
    df = client.query(query).to_dataframe()
    df.to_csv(output_dataset.path, index=False)

@component(
    packages_to_install=["pandas", "scikit-learn"],
    base_image="python:3.11",
)
def train_model(
    dataset: Input[Dataset],
    model_output: Output[Model],
    metrics: Output[Metrics],
):
    """モデルの学習と評価"""
    import pandas as pd
    from sklearn.ensemble import GradientBoostingClassifier
    from sklearn.model_selection import train_test_split
    from sklearn.metrics import accuracy_score, f1_score
    import joblib

    df = pd.read_csv(dataset.path)
    X = df.drop('churn', axis=1)
    y = df['churn']
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

    model = GradientBoostingClassifier(n_estimators=200, max_depth=5)
    model.fit(X_train, y_train)

    y_pred = model.predict(X_test)
    accuracy = accuracy_score(y_test, y_pred)
    f1 = f1_score(y_test, y_pred)

    metrics.log_metric("accuracy", accuracy)
    metrics.log_metric("f1_score", f1)

    joblib.dump(model, model_output.path)

@dsl.pipeline(
    name="churn-prediction-pipeline",
    description="顧客離脱予測モデルの学習パイプライン",
)
def training_pipeline(query: str):
    # Step 1: データ抽出
    extract_task = extract_data(query=query)

    # Step 2: モデル学習
    train_task = train_model(dataset=extract_task.outputs["output_dataset"])

# パイプラインのコンパイルと実行
compiler.Compiler().compile(
    pipeline_func=training_pipeline,
    package_path="pipeline.json",
)

job = aiplatform.PipelineJob(
    display_name="churn-pipeline-run",
    template_path="pipeline.json",
    pipeline_root="gs://my-bucket/pipeline-root",
    parameter_values={
        "query": "SELECT * FROM `my-project.analytics.customer_data`",
    },
)

job.submit()

Gemini APIとの統合

Vertex AI経由でのGemini利用

import vertexai
from vertexai.generative_models import GenerativeModel, Part

vertexai.init(project="my-project", location="asia-northeast1")

# Geminiモデルの利用
model = GenerativeModel("gemini-2.0-flash")

# テキスト生成
response = model.generate_content(
    "SESエンジニアのキャリアパスについて500字で解説してください"
)
print(response.text)

# マルチモーダル（画像理解）
image = Part.from_uri(
    "gs://my-bucket/architecture-diagram.png",
    mime_type="image/png",
)
response = model.generate_content([
    image,
    "このシステム構成図を分析して、改善点を提案してください",
])
print(response.text)

Function Calling

from vertexai.generative_models import FunctionDeclaration, Tool

# 関数定義
get_weather = FunctionDeclaration(
    name="get_weather",
    description="指定された場所の天気情報を取得する",
    parameters={
        "type": "object",
        "properties": {
            "location": {"type": "string", "description": "場所（例: 東京）"},
            "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]},
        },
        "required": ["location"],
    },
)

tool = Tool(function_declarations=[get_weather])
model = GenerativeModel("gemini-2.0-flash", tools=[tool])

response = model.generate_content("東京の天気を教えて")
# → Function Callが返される → 実際のAPIを呼び出して結果を返す

Terraformでの環境構築

# Vertex AIの基本リソース
resource "google_vertex_ai_dataset" "churn" {
  display_name          = "customer-churn-dataset"
  metadata_schema_uri   = "gs://google-cloud-aiplatform/schema/dataset/metadata/tabular_1.0.0.yaml"
  region                = var.region
}

resource "google_vertex_ai_endpoint" "prediction" {
  display_name = "churn-prediction-endpoint"
  region       = var.region

  network = google_compute_network.vpc.id

  encryption_spec {
    kms_key_name = google_kms_crypto_key.vertex.id
  }
}

# Model Monitoring
resource "google_vertex_ai_model_deployment_monitoring_job" "monitor" {
  display_name = "churn-model-monitoring"
  endpoint     = google_vertex_ai_endpoint.prediction.id
  region       = var.region

  model_deployment_monitoring_objective_configs {
    deployed_model_id = "churn-v1"
    objective_config {
      prediction_drift_detection_config {
        drift_thresholds {
          key   = "churn"
          value { value = 0.1 }
        }
      }
    }
  }

  model_monitoring_alert_config {
    email_alert_config {
      user_emails = ["[email protected]"]
    }
  }

  log_sample_rate              = 0.1
  monitoring_schedule_config {
    monitor_interval { value = 24, unit = "HOURS" }
  }
}

コスト管理

主な料金

サービス	料金
AutoML学習	$19.32 / node hour
カスタム学習（n1-standard-4）	$0.19 / hour
GPU学習（T4）	$0.35 / GPU hour
オンライン予測（n1-standard-2）	$0.095 / hour
バッチ予測	$0.035 / node hour
Gemini 2.0 Flash（入力）	$0.075 / 1M tokens

コスト最適化のポイント

1. AutoML予算設定: budget_milli_node_hours で学習時間を制限
2. Spot VM: カスタム学習でSpot VMを使用（最大60%コスト削減）
3. エンドポイントのスケーリング: min_replica_count=0でゼロスケール
4. モデルの軽量化: 量子化・蒸留で推論コストを削減

SES現場で求められるスキル

スキルレベルマップ

レベル1: ML基礎理解、AutoMLの操作
レベル2: カスタムモデルの学習・デプロイ
レベル3: MLOps パイプライン構築
レベル4: 大規模ML基盤の設計・運用

単価への影響

• AutoML操作・データ前処理 → 月額5〜10万円のスキルアップ
• カスタムモデル開発 + デプロイ → 月額70〜90万円の案件に対応
• MLOpsパイプライン構築 → 月額80〜110万円のMLエンジニア案件
• 大規模ML基盤設計 → 月額100万円超のアーキテクト案件

まとめ

Google Cloud Vertex AIは、MLモデルのライフサイクル全体を一元管理できる強力なプラットフォームです。

---

Google Cloud完全攻略シリーズの他の記事も合わせてご覧ください：

• Google Cloud入門ガイド
• BigQuery データ分析ガイド
• Cloud Functions サーバーレスガイド
• GKE Kubernetes入門
• Cloud IAM セキュリティガイド