Vertex AI PipelinesによるMLバッチ基盤の運用

サムネイル

はじめに
1. Vertex AI Pipelines採用の背景
2. 開発効率化
3. 運用と監視
4. コスト管理
おわりに

はじめに

こんにちは、DMMのML基盤チームの上田です。

DMMには多種多様なサービスがあり、検索の精度向上やレコメンドなど、ユーザー体験向上のために機械学習を活用しています。各施策ごとに専用のMLパイプラインを構築しているため、パイプラインの数は増え続け、要件も多岐にわたります。

サービスが増えるにつれ、各チームが独自にMLパイプラインを構築・運用するのは非効率になりがちです。そこで私たちは、ML基盤チーム（MLOpsエンジニア）が中心となり、Vertex AI Pipelinesをベースにした共通の運用基盤を整備してきました。検索チーム・レコメンドチーム（MLエンジニア）はこの基盤上でパイプラインを実装し、3つのチームが1つのモノレポで協働しています。

この体制では、MLエンジニアがロジックの実装に集中できる環境を提供しつつ、複数チームでの運用ならではの工夫が求められます。本記事では、継続的に運用してきた中で蓄積した知見をもとに、開発効率化、運用と監視、コスト管理の観点から、実践的なTipsを紹介します。同様の課題を抱えている方の参考になれば幸いです。

1. Vertex AI Pipelines採用の背景

私たちはGoogle CloudのVertex AI PipelinesをMLパイプラインの実行基盤として採用しています。Vertex AI Pipelinesは、Google Cloudが提供するマネージドなワークフローオーケストレーションサービスで、Kubeflow Pipelines（KFP）をベースにしています。

基盤選定にあたって重視したのは、差別化につながらない部分はマネージドサービスに任せるという考え方です。Kubernetesクラスタの運用やワークフローエンジンの保守をGoogle Cloudに委譲することで、ML基盤チームはプラットフォームとしての開発体験向上や、各チーム固有の課題解決に注力できます。

Vertex AI Pipelinesを採用した主な理由は以下の通りです。

フルマネージド

以前は自前のKubernetesクラスタやワークフローエンジンの運用にリソースを割いていましたが、Vertex AI Pipelinesに移行することでインフラ運用の負荷がほぼゼロになりました。クラスタのアップグレードやノードの管理が不要になり、開発体験の向上や、各チーム固有の課題解決に集中できるようになっています。

直感的なUI/UX

Vertex AI PipelinesのUIでは、各コンポーネントのログをUI上で直接確認でき、Cloud Loggingにアクセスせずにトラブルシューティングが完結します。また、生成されたモデルやデータセットがどのパイプライン実行で作成されたか追跡できるため、問題発生時の原因特定が格段に速くなりました。

メタデータ管理

Vertex ML Metadataに、パイプラインの実行履歴、使用したパラメータ、生成されたアーティファクトを自動的に記録します。特にデバッグ時に、例えば本番モデルの精度が低下した際、過去の実行履歴からどのデータセット・パラメータで学習したかを特定し、原因を調査できます。

KFP SDKの学習コストはありますが、後述する共通コンポーネントやテンプレート提供で、実装者の負担を下げる工夫をしています。

2. 開発効率化

多様なパイプラインを効率よく開発するために、コンポーネントの共通化と抽象化を進めています。ML基盤チームはプラットフォームエンジニアリングの考え方に基づいて、テンプレート、共通コンポーネント、ベストプラクティスなどのゴールデンパスを提供し、検索・レコメンドチームがビジネスロジックの実装に集中できる環境を整えています。

2.1 CI/CDとディレクトリ構成

まず、開発効率化の前提となるCI/CD環境とディレクトリ構成について説明します。

2.1.1 GitHub Actionsの採用

パイプラインのビルド・デプロイ・定期実行にはGitHub Actionsを採用しています。以前はCloud Scheduler + Cloud Run functionsでパイプラインをキックしていましたが、GitHub Actionsに移行することで、コードとCI/CDの実行環境が同じリポジトリに統合されました。これによりCloud SchedulerやCloud Run functionsといった外部サービスの管理が不要になりました。パイプラインの実装、テスト、本番実行まで、すべて同じスクリプトで統一できるため、ローカル実行と本番実行の差分がなく、デバッグが容易です。

デメリットとして、GitHub Actionsのcronスケジュールは、厳密に定刻通りに実行されない場合があります（数分程度の遅延あり）。ただし、私たちのパイプラインは日次バッチ中心で、秒単位の精度を求められるユースケースがないため、この制約は実運用上の問題になっていません。

2.1.2 ディレクトリ構成

各パイプラインはpipelines/<pipeline-name>/配下で管理され、以下のファイルで構成されています。

pipelines/
└── my-pipeline/
    ├── pipeline.py        # KFP SDKでパイプライン定義を記述
    ├── config.yaml        # 環境ごとのパラメータを一元管理
    └── components/        # コンポーネントのコード・Dockerfile
        ├── train/
        │   ├── main.py
        │   └── Dockerfile
        └── rerank/
            ├── main.py
            └── Dockerfile

2.1.3 デプロイフロー

MLエンジニアがパイプラインをmainブランチにマージすると、以下のフローが自動的に実行されます。

デプロイフロー

MLエンジニアがパイプラインを実装・更新
- pipeline.py: KFP SDKでパイプラインの処理フローを定義
- components/: 各コンポーネント（学習、推論）処理の実装
- config.yaml: 環境（dev/stg/prd）ごとのパラメータを設定
PRを作成してmainブランチにマージ
GitHub Actionsの処理
- コンポーネントのDockerイメージをビルドし、Google Artifact Registry（GAR）にpush
- pushしたイメージタグを参照してpipeline.pyをコンパイルし、KFPテンプレートを生成
- KFPテンプレートをGARにpush
- KFPテンプレートのタグをconfig.yamlから生成した環境ごとのworkflowファイル（.github/workflows/{env}-{pipeline_name}.yaml）に反映
定期実行
- 生成されたworkflowファイルのcron設定に従い、GitHub ActionsがGARからKFPテンプレートを取得してVertex AI Pipelinesを起動

MLエンジニアはconfig.yamlを更新するだけで、パイプラインごとに異なるスケジュールを設定でき、環境（dev/stg/prd）ごとのパラメータ管理も柔軟に行えます。

2.1.4 Config駆動のパイプライン管理

config.yamlには、環境ごとのパラメータ、スケジュール、通知先などを独自フォーマットで記述します。

labels: # 後述のコスト管理で使用
  maintainer: recommend-growth  # チーム識別
  service: dmmtv            # サービス識別
  site: general              # サイト識別

environments:
  dev:
    system_parameters:
      enable_caching: true  # キャッシュを有効化するか
      slack_mention_on_failed: U0XXXXXXXXX  # 通知先Slack ID
    pipeline_parameters:
      dt: ${yesterday:%Y-%m-%d}  # 実行時に使われる動的パラメータ
      project_id: my-project-dev
      ab_group_id: 0

  stg:
    system_parameters:
      cron: "30 2 * * *"  # 11:30 JST 実行
      enable_caching: false
      slack_mention_on_failed: U0XXXXXXXXX
    pipeline_parameters:
      # ... stg環境のパラメータ

  prd:
    system_parameters:
      cron: "30 2 * * *"
      slack_mention_on_failed: subteam_SXXXXXXXXXX  # チーム全体
    pipeline_parameters:
      # ... prd環境のパラメータ

このアプローチの利点は、全環境のパラメータが1ファイルに集約され、パイプラインのコードと同じディレクトリで管理できることです。パイプラインの設定がpipelines/<pipeline-name>/配下に集約されているため、MLエンジニアはCI/CDの仕組みを意識せずに、このディレクトリ内のファイルだけを編集すれば良いという明確な開発モデルを提供できています。GitHub Actionsのワークフローファイルを直接編集する必要がないため、パイプライン開発とCI/CD設定の責務が分離され、開発体験が向上しています。また、全環境の設定が縦に並んでいるため、「devでは設定したがprdでは設定し忘れた」といったミスを防ぐのにも役立っています。

動的パラメータの解決

config.yamlでは、OmegaConfのカスタムリゾルバを活用して、実行時に動的にパラメータを決定できます。例えば、dt: ${yesterday:%Y-%m-%d}のように記述すると、パイプライン実行時に前日の日付（日本時間基準）が自動的に計算され、パラメータとして渡されます。日次バッチでは昨日のデータを処理するユースケースがほとんどなので、多くのパイプラインで利用しています。

2.2 Goによる共通コンポーネント

DMMのMLパイプラインでは、BigQueryで集計・加工したデータを外部システムに転送する処理が頻繁に発生します。例えば、推論結果を検索インデックス更新のためのインデクシングAPIへ数百万件のデータを送信したり、生成したレコメンドをリアルタイム提供のためのDynamoDBに書き込んだりします。これらの処理は複数のパイプラインで共通的に必要とされており、各チームが独自に実装すると、同じような処理が重複してしまいます。

2.2.1 PythonからGoへの移行

当初はこれらのデータ転送処理もPythonで実装していました。しかし、数百万件規模のデータを扱うと、以下のような課題が出てきました。

処理時間の長さ: 逐次処理では外部APIへのリクエスト待ちが多く、全体の処理時間が数時間に及ぶケースがあった
メモリ使用量: 一度にすべてのデータをメモリに載せると、大量のメモリを消費してしまう
並行処理の複雑さ: Pythonのマルチスレッド・マルチプロセスは実装が複雑で、エラーハンドリングも難しい

そこで、これらのデータ転送処理をGo言語で実装し直し、共通コンポーネントとして提供することにしました。Goのgoroutineを活用した並行処理により、複数のAPIリクエストを同時に送信でき、I/O待ち時間を有効活用できます。また、BigQuery Storage APIからストリーミングでデータを読み込み、そのまま外部APIに送信することで、メモリ使用量も最小限に抑えられます。

2.2.2 複数パイプラインでの共通利用

Goで実装したデータ転送コンポーネントは、検索・レコメンド両チームの複数のパイプラインで利用されています。BigQueryから外部APIへのデータ転送という処理パターンは共通しているため、一度Goコンポーネントを実装すれば、さまざまなパイプラインで再利用できます。複雑なロジックもコンポーネント内に閉じ込めることで、各パイプラインでの実装負担を大幅に減らせています。

2.2.3 抽象化による認知負荷の軽減

検索・レコメンドチームのMLエンジニアは技術スタックとしてPythonを得意としており、Goの並行処理（goroutineやチャネル）に詳しくないメンバーがほとんどです。Team Topologiesでは、チームが抱える認知負荷を減らすことが生産性向上の鍵だと説いています。

そこで、Goコンポーネントの内部実装は完全に抽象化し、利用者はパラメータを渡すだけで使える設計にしています。複雑な並行処理やGoogle CloudからAWSの認証ロジックはすべてコンポーネント内に隠蔽されています。

例えば、BigQueryからDynamoDBにデータを送信する場合、MLエンジニアはGoのコードを書く必要がなく、以下のようにパラメータを指定するだけで利用できます。

args:
  - --query=SELECT * FROM `project.dataset.table` # データを取得するBigQueryクエリ
  - --target_dynamodb_table=my-table              # 書き込み先のDynamoDBテーブル

2.2.4 パフォーマンスの改善事例

Python実装からGo実装に移行した結果、あるパイプラインでは処理時間が約1/5に短縮されました。特にAPIリクエストのI/O待ちが多い処理では、並行処理の効果が顕著に現れます。また、メモリ使用量も大幅に削減され、より小さなマシンタイプで実行できるようになり、コスト削減にもつながっています。

2.3 ゴールデンパスとセルフサービス

プラットフォームエンジニアリングのもう一つの重要な要素がセルフサービスです。MLエンジニアが基盤チームに依頼せずとも、自分たちで必要な環境を構築できる仕組みを提供しています。

2.3.1 テンプレート提供

新しいパイプラインを作成する際、cookiecutterを使ってテンプレートから雛形を生成できるようにしています。必要なファイル構成があらかじめ整っているため、「何をどう書けば良いか」を考える必要がなく、すぐに開発を始められます。これがゴールデンパスの起点となります。

2.3.2 社内ライブラリの提供

AWSへの認証処理やBigQueryのラッパー、Logger（構造化ロギング）などの社内ライブラリをPythonパッケージとしてバージョン管理し、Google Artifact Registry（GAR）に公開しています。MLエンジニアはパイプラインで必要なライブラリを指定するだけで、共通機能を利用できます。

2.3.3 共通ベースイメージ

ML基盤チームは、主にGPUを利用するパイプライン向けに共通ベースイメージを事前にビルドして提供しています。PyTorchやFaissといった機械学習ライブラリは、CUDAやcuDNNなどGPU関連の依存関係が複雑で、バージョンの組み合わせによっては互換性の問題が発生します。これらの依存関係解決をベースイメージ側で吸収することで、MLエンジニアは複雑な環境構築を意識せずに開発できます。

また、前述の社内ライブラリも含めることで、コンテナビルド時間も短縮され、開発サイクルが高速化します。開発体験の向上において、こうした待ち時間の削減と複雑さの隠蔽は見過ごされがちですが、日々の開発効率に大きく影響する重要な要素です。

3. 運用と監視

MLパイプラインは日々定期実行され、ビジネスに直結するデータを生成しています。パイプラインが停止すると、サービスへの影響が大きいため、高い信頼性を確保する必要があります。

本セクションでは、以下の取り組みを紹介します。

リトライポリシー: 一時的障害への自動対応
失敗時のSlack通知: エラー検知と迅速な対応
実行時間ベースの監視: 24時間超過の自動停止、パイプライン間依存のSLO
パフォーマンス監視: 実行時間推移の可視化
データ品質監視: Dataplexによる自動チェック

3.1 リトライポリシー

外部APIやBigQueryへのアクセスでは、ネットワークの一時的な障害やレートリミットにより処理に失敗する場合があります。こうした一時的な障害に対応するため、コンポーネント単位でリトライポリシーを設定しています。

KFPでは、各コンポーネントに対して以下のようにリトライ回数とバックオフ係数を指定できます。

from kfp import dsl

@dsl.component
def my_component():
    # ... 処理 ...
    pass

# リトライ設定を適用
my_task = my_component()
my_task.set_retry(
    num_retries=3,           # 3回リトライ
    backoff_duration="60s",  # 初回待機時間60秒
)

この設定により、1回目の失敗後は60秒待機、2回目は120秒待機、3回目は240秒待機してリトライされます。外部APIへのアクセスなど、一時的な障害が起きやすい処理では、このようなリトライ設定が有効です。設定はガイドラインとして、コストの観点で実行時間が比較的長いもの(目安として20分以上)やGPUを使うものはリトライを1回、それ以外のものは3回として、設定の判断に迷わないようにしています。

3.2 失敗時のSlack通知

パイプラインが失敗した際、Cloud Logging → Datadog → Slackの経路で自動的に通知が飛ぶ仕組みを構築しています。単に失敗通知するだけでなく、エラーログの重要な部分だけを抽出してSlackに投稿しています。これにより、Google Cloudコンソールを開かなくても、ある程度原因の当たりをつけられます。

さらに、LLMを活用したログ解析により、エラーの原因特定をより高度化する取り組みも進めています。

developersblog.dmm.com

また、「Config駆動のパイプライン管理」で紹介したconfig.yamlのslack_mention_on_failed設定により、通知先を柔軟に制御できます。開発環境では個人にメンション、本番環境ではチーム全体にメンションするなど、環境や担当チームごとに適切な通知先を設定しています。

3.3 実行時間ベースの監視

パイプラインの実行時間を監視し、異常な長時間実行やパイプライン間の依存関係を管理しています。

24時間超過ジョブを停止するセーフティネット

私たちのパイプラインは基本的に1日1回実行され、数時間以内に完了することを想定しています。しかし、バグにより無限ループが発生したり、想定外に長時間実行されたりすると、課金が膨らみ続けます。

そこで、24時間を超えて実行されているパイプラインを自動的に強制停止するCloud Run functionsを定期実行しています。

Cloud Run functionsが1時間ごとに実行される
Vertex AI Pipelinesの実行中ジョブ一覧を取得
実行開始から24時間以上経過しているジョブを検出
該当ジョブを強制停止し、Slackに通知

通常のパイプラインは数時間以内に完了するため、24時間以上実行されているケースは明らかに異常です。これにより意図しないコスト増加を未然に防いでいます。

パイプライン間依存のSLO設定

パイプライン間の依存関係がある場合は、終了時刻ベースのSLOも設定しています。例えば、前段のパイプラインAが毎日午前10時までにBigQueryテーブルを生成し、後続のパイプラインBがそのテーブルを利用する場合、パイプラインAには終了時刻に関するSLOを設定します。

SLOを逸脱した場合はDatadogからアラートが飛び、後続パイプラインへの影響を事前に検知できます。これにより、パイプライン間の依存関係を明示的に管理し、連鎖的な遅延を防ぐことができます。

3.4 パフォーマンス監視

週次でDatadogダッシュボードを確認し、パイプラインの実行時間が徐々に伸びていないかチェックします。データ量の増加に伴う処理時間の延伸は自然なことですが、早期に検知することでパフォーマンス改善の必要性を判断できます。実行時間の推移を可視化することで、ボトルネックの特定や最適化の優先順位付けにも役立てています。

3.5 データ品質監視

パイプラインの実行状態だけでなく、生成されるデータの品質も監視しています。

Dataplex自動データ品質

パイプラインが生成する特徴量やレコメンド結果のBigQueryテーブルに対して、Dataplexの自動データ品質チェックを定期実行しています。Nullチェック、数値範囲チェック、ユニーク性チェックといった基本的な検証は、Dataplexの標準ルールで実施しています。さらに、Dataplexでは任意のSQLクエリを実行できるため、BigQuery MLの ML.VALIDATE_DATA_DRIFT関数を活用したデータドリフト検知も同じ仕組みで一元管理しています。

品質チェックの設定はTerraformで抽象化されており、MLエンジニアがYAMLファイルでルールを定義するだけで、新しいテーブルの品質監視を追加できます。

Dataplexデータプロファイリング

開発時には、Dataplexデータプロファイリング機能をラップした共通コンポーネントを提供しています。MLエンジニアはパイプライン定義にこのコンポーネントを追加するだけで、データの統計情報や分布を可視化でき、想定通りのデータが生成されているか確認できます。プロファイリング結果はVertex AI PipelinesのUI上でリンクとして表示され、Google CloudコンソールのDataplexページで詳細を確認できます。

運用と監視により信頼性を確保できても、コストが際限なく増加していては持続可能な運用とは言えません。次に、クラウドコストを適切に管理し、ビジネス価値とコストのバランスを取るための取り組みについて紹介します。

4. コスト管理

4.1 FinOpsとは何か

FinOps（Financial Operations）は、クラウドコストの管理を単なるコスト削減ではなく、ビジネス価値を最大化するための投資判断として捉える考え方です。エンジニア、ビジネス、財務の3者が協力し、コストの可視化・最適化・予測を継続的に行うことで、スピードとコストのバランスを取りながら成長を目指します。

FinOps Foundationでは、FinOpsの実践を以下の3つのフェーズで定義しています。

Inform（可視化）: コストを可視化し、誰が何にいくら使っているかを把握する
Optimize（最適化）: 無駄を削減し、コスト効率を改善する
Operate（運用）: 継続的な改善とガバナンスを確立する

本記事ではInformフェーズの取り組みについて紹介します。

4.2 なぜML基盤でFinOpsが必要なのか

DMMでは検索・レコメンドサービスを多数のサービスに提供しており、パイプラインの数も増え続けています。複数チームが1つの基盤を共有する環境では、コスト管理の難しさが顕在化します。

従来のコスト削減の発想では、パイプラインの実行回数を減らしたり、マシンスペックを下げたりすることで一時的にコストは下がりますが、それはビジネス価値の低下につながりかねません。私たちが目指すのは、単純なコスト削減ではなく、ビジネス価値を維持しながら無駄を削減することです。

そのためには、何に対してどれだけのコストがかかっているかを可視化し、意思決定の材料にすることが重要です。データ量が2倍になったときにコストも2倍になるのは健全ですが、データ量が変わらないのにコストが2倍になっているなら、それは非効率なクエリや不要な処理が原因かもしれません。このような気づきを得るために、FinOpsの仕組みが必要です。

4.3 ラベル付与戦略

コスト管理の基盤となるのが、共通フォーマットのラベルの付与です。「Config駆動のパイプライン管理」で紹介したconfig.yamlには、以下のラベルを定義しています。

service: どのサービスか（例: dmmtv, ebook）
maintainer: 担当チーム（例: search-growth, recommend-growth）
site: サイト識別
pipeline_name: パイプライン名（自動付与）

これらのラベルは、Vertex AI Pipelinesの実行時だけでなく、パイプラインからBigQueryジョブにも自動的に付与されます。社内ライブラリのBQラッパーを経由してクエリを実行することで、MLエンジニアが意識しなくても、すべてのBigQueryジョブに統一されたラベルが付与される仕組みです。

4.3.1 ラベル付与からコスト可視化までの流れ

ラベル付与からコスト可視化までの流れ

MLエンジニアはconfig.yamlでラベルを定義するだけで、Vertex AI PipelinesとBigQueryの両方に自動的にラベルが付与されます。Vertex AI PipelinesではCustom Training JobがSKUとしてラベルと一緒にBilling Exportに記録されます。BigQueryジョブは社内ライブラリのBQラッパーを経由して実行され、自動的にラベルが付与されます。

ラベル付与のロジックは完全に抽象化されているため、MLエンジニアはコスト管理のための特別な実装を意識することなく、通常通りパイプライン開発に集中できます。コストデータはBilling Exportから取得し、BigQueryのコストはINFORMATION_SCHEMA.JOBSとRESERVATIONSを組み合わせて算出し、Looker Studioで統合して可視化しています。

4.4 コストダッシュボード

付与したラベルをもとに、BigQueryのBilling Exportデータを集計し、Looker Studioでダッシュボード化しています。ダッシュボードでは、以下のような多角的な分析が可能です。

サービス別・パイプライン別の集計

「今月はこのサービスでいくらかかっているか」「このパイプラインのコストが先月と比べてどう変化しているか」といった情報を一目で確認できます。サービス軸、パイプライン軸、maintainer軸など、複数の切り口でコストを集計し、責任範囲を明確にしています。

コストダッシュボード

BigQueryジョブ詳細の可視化

特定のパイプラインを選択すると、そのパイプラインで実行された個々のBigQueryジョブの詳細まで掘り下げて確認できます。各ジョブについて：

ジョブのクエリ内容
スロット消費時間から概算したコスト
BigQuery コンソールのジョブ詳細ページへの直リンク

を表示しています。これにより、「どのクエリがコストを押し上げているのか」をピンポイントで特定でき、最適化の対象を明確にできます。

BQジョブ詳細

エンジニアのコスト意識向上

このダッシュボードを全エンジニアに公開することで、「自分の開発したパイプラインがいくらかかっているか」を誰でも確認できるようにしています。これにより開発時からコストを意識し、不要な処理を削減したり、クエリを最適化したりする文化が徐々に根付いてきています。

4.5 コスト意識の文化を醸成する

FinOpsの成功には、技術的な仕組みだけでなく文化醸成が重要です。基盤チームが中央集権でコスト監視をするだけではなく、MLエンジニア自身が自分の書いたコードがいくらかかっているかを意識する文化を作る必要があります。

FinOpsの原則の1つに「Everyone takes ownership for their technology usage（全員が自分の技術利用に責任を持つ）」があります。コストの責任をエンジニアに分散し、アーキテクチャ設計から日常運用まで、各チームが自分たちのクラウド利用を予算内で管理できるようにすることが求められています。開発の初期段階からコストを効率性の新しい指標として考慮する必要があります。

透明性がもたらす責任感

誰がどのパイプラインでいくら使っているかを全員が見られる状態にすることで、自然と責任感が生まれます。Looker StudioダッシュボードをMLエンジニアに公開し、自分のパイプラインのコストを誰でも確認できるようにした結果、以下のような変化が生まれています。