AWS ML/AI本番運用Vol3｜SageMaker MLOps完全ガイド

§1 なぜ AI/ML Vol3 か — 生成AI から MLOps への展開 + AI三部作ナビ

AWS の AI/ML 本番運用は「生成AI を動かす」だけでは完結しない。Bedrock で構築した RAG や Agents が組織に根付いたとき、次の問いが必ず立ち上がる。「自社固有データで学習した独自モデルを、安定的に本番運用するにはどうすればよいか」。

Vol1 では Bedrock Agents の Action Group と Knowledge Base を徹底解説した。Vol2 では Embedding / RAG / Knowledge Bases / Agents の生成AI 4本柱で AWS 標準パターンを確立した。そして本記事 Vol3 は、SageMaker を軸とした MLOps 本番運用 — Studio × Autopilot × Model Monitor × Pipelines × Feature Store — を扱う AI三部作の完結巻だ。

生成AI と MLOps の使い分け

生成AI (Bedrock) と カスタム機械学習 (SageMaker) は役割が異なる。

Bedrock で RAG を構築し定性的な質問応答を自動化する一方、SageMaker で需要予測や異常検知モデルを学習・デプロイし定量的な意思決定を支援する — この2軸を組み合わせることが、AWS 上の AI/ML 本番運用の全体像だ。

生成AI → MLOps への展開ロジック

Vol1/Vol2 で積み上げた生成AI 基盤を持つチームが SageMaker MLOps に踏み出すとき、接続点は3つある。

① 独自データ × カスタムモデル: Bedrock の Knowledge Base で構築したデータレイクを SageMaker Feature Store に取り込み、ドメイン特化モデルを学習する。RAG で補いきれない定量予測タスクに対応できる。

② 品質保証の高度化: Bedrock Agents の応答品質は人手評価に頼る部分が大きい。SageMaker Model Monitor を組み合わせることで、推論データの統計的ドリフト検知を自動化し、品質劣化を定量で捕捉できる。

③ MLOps パイプラインの自動化: 再学習トリガー (Drift 検知 → Pipeline 起動 → Autopilot 候補選択 → Model Registry 登録 → 承認ゲート) を SageMaker Pipelines で完全自動化し、モデルリリースの Lead time を短縮する。

本記事の読者想定

• MLエンジニア: SageMaker の各サービスを本番水準で統合したい
• データサイエンティスト: ローカル実験から Managed training / Autopilot への移行を進めたい
• AI/ML プラットフォームチーム: MLOps 基盤を AWS ネイティブに設計・構築したい

前提知識として AWS IAM / VPC の基礎と Python / Terraform の読み書きができれば十分だ。Bedrock / 生成AI の経験は不要だが、Vol1/Vol2 を読んでいると AI/ML 全体像の理解が深まる。

§2〜§8 で扱う5大テーマ

§7 では実務で頻出する詰まりポイント7選を判断ツリーで整理し、§8 でアンチパターン演習5問と全軸クロスリンクナビを提供する。

MLOps 成熟度モデルと本記事の位置付け

MLOps には段階的な成熟度がある。Google が定義した MLOps Maturity Level を AWS 環境に当てはめると次のようになる。

本記事が対象とするのは Level 1〜3 の実装だ。Level 0 からの脱却 (Studio 導入 / Autopilot 活用) から、Level 3 の完全自動 MLOps パイプライン (Model Monitor + Pipelines 連携) まで、段階的に解説する。

AWS SageMaker サービス全景

SageMaker は単一サービスではなく、ML ライフサイクルをカバーするサービス群だ。

SageMaker エコシステム (主要サービス)
├── Studio ← 統合IDE (本記事 §2)
│├── JupyterLab
│├── Code Editor (VS Code)
│└── Studio Notebook
├── Autopilot ← AutoML (本記事 §3)
│├── Tabular / Text / Time-series
│├── Candidate selection
│└── Explainability (SHAP)
├── Model Registry  ← モデルバージョン管理
├── Model Monitor← 本番監視 (本記事 §4)
│├── Data Quality
│├── Model Quality
│├── Bias Monitor
│└── Feature Attribution
├── Pipelines ← MLOps 自動化 (本記事 §5)
│├── Processing Step
│├── Training Step
│├── Conditional Step
│└── Lambda Step
└── Feature Store← 特徴量管理 (本記事 §6)
 ├── Online Store (低レイテンシ推論用)
 └── Offline Store (学習用 S3)

Model Registry は本記事で独立した節としては扱わないが、§5 の SageMaker Pipelines 解説の中でモデル登録・承認ゲートのパターンとして登場する。

なぜ今 SageMaker MLOps なのか — 2024年以降のトレンド

2024年以降、AWS は SageMaker の刷新を加速している。新 Studio UI の GA、Autopilot v2 (create_auto_ml_job_v2) の強化、Pipelines と EventBridge の統合改善など、MLOps 基盤としての完成度が急速に高まった。

特に以下の3点が本番採用のハードルを下げた:

① Infrastructure-as-Code の成熟: Terraform AWS Provider が SageMaker リソースの大部分をサポートし、Domain / User Profile / Pipeline / Feature Group を Terraform で管理できるようになった。本記事のコード例は全て Terraform + Python SDK の組み合わせで示す。

② コスト透明化: 新 Studio の Space アーキテクチャ、Spot instance 対応の Managed training、Serverless Inference の選択肢が増え、MLOps の TCO (Total Cost of Ownership) が予測しやすくなった。

③ セキュリティ統合の強化: VPC Only モード / Customer Managed KMS / IAM Identity Center 統合 / VPC Endpoint 経由の通信制限が標準化され、金融・医療などの規制業界でも採用しやすい環境が整った。

ML ライフサイクルと SageMaker サービスの対応

実際のプロジェクトで SageMaker がどのフェーズで機能するかを整理すると、全体設計が見えやすくなる。

この表のフローを左から右に自動化することが MLOps の本質だ。本記事では各フェーズの実装コードと Terraform を提供し、局所的な知識を統合パイプライン設計に繋げることを目指す。

本記事の進め方

初めて SageMaker を触る場合は §2 (Studio) から順番に読むことを推奨する。既に Studio/Autopilot を使用中で、本番品質の監視・自動化を強化したい場合は §4 (Model Monitor) と §5 (Pipelines) が優先度高い。Feature Store を未導入のチームは §6 が直接的な改善ポイントになる。

§7 の詰まりポイント7選は、本番稼働後に問題が発生した際の辞書的な参照として活用してほしい。§8 のアンチパターン演習は、設計段階で陥りがちな罠を事前に把握するための実践トレーニングだ。

§2 SageMaker Studio 本番運用 — 統合IDE × Domain設計 × JupyterLab × Studio Notebook

SageMaker Studio は、データ準備から学習・デプロイ・監視まで ML ライフサイクル全体を一つの IDE で扱える統合開発環境だ。2024年以降は新 SageMaker Studio が標準となり、旧 Studio Classic は段階的に廃止予定のため、本記事は新 Studio を前提として解説する。

2.1 新 SageMaker Studio と Studio Classic の違い

最も重要な差分は課金モデルだ。Studio Classic では JupyterServer App が常時起動しており、使っていなくても EFS / ml.t3.medium 相当のコストが発生し続ける。新 Studio では使用する Space だけが起動するため、コスト管理が大幅に改善された。

2.2 Domain 設計 — 組織単位の境界設計

SageMaker Studio の最上位概念は Domain だ。Domain は AWS リージョン内の論理的な境界であり、VPC / EFS / IAM Identity Center の統合設定を保持する。

Single-Domain vs Multi-Domain

本番環境ではMulti-Domain を推奨する。理由は2つある。①部門間でデータアクセス境界を EFS レベルで分離できる。②Execution Role を Domain 単位で管理することで最小権限原則を徹底できる。

User Profile と Execution Role

Domain 内の各ユーザーは User Profile として登録される。User Profile には独立した Execution Role (IAM Role) が紐づき、S3 / ECR / SageMaker API へのアクセス権限を制御する。

# Terraform: SageMaker Domain
resource "aws_sagemaker_domain" "main" {
  domain_name = "ml-platform-prod"
  auth_mode= "IAM"
  vpc_id= var.vpc_id
  subnet_ids  = var.private_subnet_ids
  app_network_access_type = "VpcOnly"

  default_user_settings {
 execution_role = aws_iam_role.sagemaker_execution.arn

 jupyter_lab_app_settings {
default_resource_spec {
  instance_type = "ml.t3.medium"
  sagemaker_image_arn = data.aws_sagemaker_prebuilt_ecr_image.jupyter.registry_path
}
 }

 kernel_gateway_app_settings {
default_resource_spec {
  instance_type = "ml.t3.medium"
}
 }
  }

  domain_settings {
 security_group_ids = [aws_security_group.studio.id]
  }

  kms_key_id = aws_kms_key.studio.arn

  tags = {
 Environment = "production"
 Team  = "ml-platform"
  }
}

app_network_access_type = "VpcOnly" を設定することで、Studio からのアウトバウンド通信が全て VPC 経由になる。インターネット直通を禁止して VPC Endpoint 経由のみに制限することが、本番環境のセキュリティ要件として必須だ。

# Terraform: User Profile
resource "aws_sagemaker_user_profile" "data_scientist" {
  domain_id= aws_sagemaker_domain.main.id
  user_profile_name = "data-scientist-alice"

  user_settings {
 execution_role = aws_iam_role.ds_execution.arn

 jupyter_lab_app_settings {
default_resource_spec {
  instance_type = "ml.t3.medium"
}
lifecycle_config_arns = [
  aws_sagemaker_studio_lifecycle_config.base.arn
]
 }
  }
}

lifecycle_config_arns に Lifecycle Configuration を指定することで、JupyterLab 起動時の自動セットアップ (pip install / 環境変数設定 / カスタムスクリプト) を一元管理できる。

2.3 Execution Role の IAM 設計

Execution Role には SageMaker サービスが必要とする最小権限を付与する。

# Terraform: Execution Role
resource "aws_iam_role" "sagemaker_execution" {
  name = "SageMakerExecutionRole-prod"

  assume_role_policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Principal = { Service = "sagemaker.amazonaws.com" }
Action = "sts:AssumeRole"
 }]
  })
}

resource "aws_iam_role_policy_attachment" "sagemaker_full" {
  role = aws_iam_role.sagemaker_execution.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonSageMakerFullAccess"
}

resource "aws_iam_role_policy" "s3_access" {
  name = "SageMakerS3Access"
  role = aws_iam_role.sagemaker_execution.id

  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [
{
  Effect = "Allow"
  Action = ["s3:GetObject", "s3:PutObject", "s3:ListBucket"]
  Resource = [
 "${aws_s3_bucket.ml_data.arn}",
 "${aws_s3_bucket.ml_data.arn}/*"
  ]
},
{
  Effect= "Allow"
  Action= ["kms:GenerateDataKey", "kms:Decrypt"]
  Resource = aws_kms_key.studio.arn
}
 ]
  })
}

AmazonSageMakerFullAccess は広範なため、本番では S3 バケット・KMS キー・ECR リポジトリを明示した最小権限ポリシーに置き換えることを推奨する。

2.4 VPC / EFS / KMS 統合設定

本番環境の Studio Domain は3つのインフラ統合が必須だ。

VPC 統合: VpcOnly モードで全通信を VPC 内に閉じ、SageMaker API / S3 / ECR への通信は VPC Endpoint (PrivateLink) を使う。必要な Endpoint は sagemaker.api / sagemaker.runtime / s3 / ecr.api / ecr.dkr / sts の最低6種類だ。

# VPC Endpoint for SageMaker API
resource "aws_vpc_endpoint" "sagemaker_api" {
  vpc_id  = var.vpc_id
  service_name  = "com.amazonaws.ap-northeast-1.sagemaker.api"
  vpc_endpoint_type= "Interface"
  subnet_ids = var.private_subnet_ids
  security_group_ids  = [aws_security_group.vpce.id]
  private_dns_enabled = true

  tags = { Name = "vpce-sagemaker-api" }
}

EFS 統合: Domain 作成時に EFS ファイルシステムが自動生成される。デフォルトでは全 User Profile が同一 EFS に接続するため、/home/user-profile-name/ のプレフィックスで分離される。KMS 暗号化は kms_key_id で指定する。

KMS 統合: aws_sagemaker_domain の kms_key_id に Customer Managed Key を指定することで、EFS / EBS / S3 出力データを統一的に暗号化できる。

2.5 JupyterLab と Studio Notebook — インスタンスタイプ選択

新 Studio では Space という概念でアプリを管理する。JupyterLab を起動するたびに Space が作られ、利用後に Space を削除することでコストをゼロにできる。

インスタンスタイプの選択指針

Local mode とは、JupyterLab のインスタンス上で直接 SageMaker Estimator を実行する開発モードだ。local を instance_type に指定するだけで切り替えられる。

from sagemaker.sklearn import SKLearn

# Local mode: JupyterLabインスタンス上で実行 (コスト最小)
estimator_local = SKLearn(
 entry_point="train.py",
 role=role,
 instance_type="local",  # ← local mode
 framework_version="1.2-1",
 hyperparameters={"n-estimators": 100, "max-depth": 5},
)
estimator_local.fit({"train": "file:///home/sagemaker-user/data/train/"})

# Managed training: 専用インスタンスで実行 (本番推奨)
estimator_managed = SKLearn(
 entry_point="train.py",
 role=role,
 instance_type="ml.m5.xlarge",  # ← managed instance
 instance_count=1,
 framework_version="1.2-1",
 hyperparameters={"n-estimators": 100, "max-depth": 5},
 output_path="s3://my-bucket/models/",
 use_spot_instances=True,  # Spot で最大 90% コスト削減
 max_wait=7200,
 max_run=3600,
)
estimator_managed.fit({"train": "s3://my-bucket/data/train/"})

Local mode → Managed training への移行ワークフローが MLOps 成熟度の第一歩だ。instance_type="local" で動作確認 → instance_type="ml.m5.xlarge" に切り替えて S3 入出力を設定 → Spot instance を有効化してコスト最適化、という3ステップで移行する。

2.6 Lifecycle Configuration による起動時自動化

Studio の JupyterLab Space 起動時に自動実行するスクリプトを Lifecycle Configuration として登録できる。チーム全体の環境統一に有効だ。

#!/bin/bash
# Lifecycle Configuration: JupyterLab起動時セットアップ
set -eux

# 必要パッケージのインストール
pip install --quiet \
  sagemaker \
  pandas \
  scikit-learn \
  boto3

# AWS設定
aws configure set region ap-northeast-1

# アイドル自動シャットダウン (60分で停止)
curl -o /tmp/auto-stop.py \
  https://raw.githubusercontent.com/aws-samples/sagemaker-studio-auto-shutdown-extension/main/auto_stop_idle/sagemaker-studio-autoshutdown.py
python /tmp/auto-stop.py --time-to-live-in-seconds 3600 --port 8888 &

echo "Lifecycle configuration complete."

# Terraform: Lifecycle Configuration 登録
resource "aws_sagemaker_studio_lifecycle_config" "base" {
  studio_lifecycle_config_name  = "base-setup"
  studio_lifecycle_config_app_type = "JupyterLab"
  studio_lifecycle_config_content  = base64encode(file("${path.module}/scripts/lifecycle.sh"))
}

次の §3 では、SageMaker Autopilot による AutoML 本番運用パターンを解説する。

§3 AutoML / Autopilot 本番運用 — 自動ML × Candidate selection × Explainability ★山場1

SageMaker Autopilot は、機械学習の専門知識がなくても構造化データ・テキスト・時系列・画像を対象に AutoML を実行できるマネージドサービスだ。Vol3 第一の山場として、本番で使える Autopilot 運用パターンを徹底解説する。

3.1 Autopilot 4 つの自動 ML モード

Autopilot v2 (create_auto_ml_job_v2) は 4 つのモードをサポートする。

モード選択の起点はデータ形式と予測対象だ。Tabular は汎用性が最も高く、新規プロジェクトでは最初に試す選択肢になる。Time Series は event_time の精度設計が予測品質を大きく左右するため、§6 で解説する Feature Store との組み合わせが特に有効だ。

3.2 AutoML ジョブの実行 — create_auto_ml_job_v2()

Autopilot v2 の推奨 API は create_auto_ml_job_v2()。旧 create_auto_ml_job() は機能制限があるため、新規実装では v2 一択だ。

import boto3
import time

sm = boto3.client("sagemaker", region_name="ap-northeast-1")

job_name = f"autopilot-tabular-{int(time.time())}"

response = sm.create_auto_ml_job_v2(
 AutoMLJobName=job_name,
 AutoMLJobInputDataConfig=[
  {
"ChannelType": "training",
"ContentType": "text/csv",
"DataSource": {
 "S3DataSource": {
  "S3DataType": "S3Prefix",
  "S3Uri": "s3://my-bucket/training/",
 }
},
  }
 ],
 OutputDataConfig={
  "S3OutputPath": "s3://my-bucket/autopilot-output/",
  "KmsKeyId": "arn:aws:kms:ap-northeast-1:123456789012:key/my-key",
 },
 AutoMLProblemTypeConfig={
  "TabularJobConfig": {
"TargetAttributeName": "churn",
"ProblemType": "BinaryClassification",
"CompletionCriteria": {
 "MaxCandidates": 20,
 "MaxRuntimePerTrainingJobInSeconds": 600,
 "MaxAutoMLJobRuntimeInSeconds": 7200,
},
  }
 },
 AutoMLJobObjective={"MetricName": "F1"},
 RoleArn="arn:aws:iam::123456789012:role/SageMakerAutopilotRole",
 ModelDeployConfig={
  "AutoGenerateEndpointName": False,
  "EndpointName": "",
 },
)

print(f"AutoML job started: {job_name}")

MaxCandidates と MaxAutoMLJobRuntimeInSeconds はコスト爆発防止の最重要パラメータだ。デフォルト値のまま動かすと数百候補を長時間試行し、コストが予想外に膨らむ。本番 PoC では MaxCandidates: 10〜20、MaxAutoMLJobRuntimeInSeconds: 3600〜7200 を起点として徐々に拡張する。

ジョブ完了待機と最良候補の取得は以下のように実装する。

import json

def wait_and_get_best_candidate(sm_client, job_name):
 while True:
  resp = sm_client.describe_auto_ml_job_v2(AutoMLJobName=job_name)
  status = resp["AutoMLJobStatus"]
  if status in ("Completed", "Failed", "Stopped"):
break
  print(f"Status: {status} — waiting 60s ...")
  time.sleep(60)

 if status != "Completed":
  raise RuntimeError(f"AutoML job ended with status: {status}")

 best = resp["BestCandidate"]
 return best

best_candidate = wait_and_get_best_candidate(sm, job_name)
print("Best candidate:", best_candidate["CandidateName"])
print(
 "Objective metric:",
 best_candidate["FinalAutoMLJobObjectiveMetric"]["MetricName"],
 best_candidate["FinalAutoMLJobObjectiveMetric"]["Value"],
)

3.3 最良候補の分析 — Explainability レポートと SHAP

Autopilot は CandidateArtifactLocations に Explainability レポートを自動生成する。このレポートで、モデルがどの特徴量をどれほど重視しているかを把握できる。

explain_location = (
 best_candidate
 .get("CandidateArtifactLocations", {})
 .get("Explainability", "")
)
print("Explainability report S3 path:", explain_location)
# → s3://my-bucket/autopilot-output/{job_name}/documentation/explainability/

# S3 からグローバル SHAP レポートを取得
s3 = boto3.client("s3")
bucket = "my-bucket"
key = (
 f"autopilot-output/{job_name}"
 "/documentation/explainability/index.json"
)
body = s3.get_object(Bucket=bucket, Key=key)["Body"].read()
report = json.loads(body)

shap_values = (
 report["explanations"]["kernel_shap"]["label0"]["global_shap_values"]
)
for entry in sorted(shap_values, key=lambda x: abs(x["attribution"]), reverse=True):
 print(f"  {entry['feature_name']}: SHAP = {entry['attribution']:.4f}")

SHAP グローバル値が大きい特徴量ほどモデルへの影響が大きい。本番運用では以下の 3 点を必ず確認する。

1. 想定外特徴量の高寄与: データリーク (target leakage) の兆候。訓練データに含まれてはいけない将来情報が混入している可能性がある
2. ビジネス説明不能な上位特徴量: 規制業種では説明責任を果たせないモデルをリリースできない。SHAP レポートは社内承認フローの証跡にもなる
3. 特徴量順位の時系列変化: 定期的に再実行して順位が大きく変動していた場合、データドリフトの初期兆候として扱い §4 の Model Monitor と照合する

3.4 Autopilot ↔ カスタムトレーニング 使い分け判断

Autopilot は強力だが、すべてのユースケースで最適解になるわけではない。以下の判断基準で使い分ける。

実務では「Autopilot でベースライン確立 → カスタムトレーニングで改善」の 2 段階アプローチが最もコスト効率が高い。Autopilot の結果をベースラインとして記録しておくことで、カスタムモデルの改善効果を定量的に評価できる。

3.5 Autopilot 生成コード活用 — DataTransformationStep / ModelTrainingStep の再利用

Autopilot は最良候補の前処理スクリプトとトレーニングコードを S3 に生成する。このコードを SageMaker Pipelines の Step として組み込むことで、Autopilot の探索結果をそのまま MLOps パイプラインに昇格できる。

from sagemaker.workflow.steps import ProcessingStep, TrainingStep
from sagemaker.workflow.pipeline import Pipeline
from sagemaker.processing import ScriptProcessor
from sagemaker.estimator import Estimator

ROLE = "arn:aws:iam::123456789012:role/SageMakerPipelinesRole"
REGION = "ap-northeast-1"

# Autopilot が生成した DataTransformation スクリプト (S3 パス)
data_transform_script = (
 f"s3://my-bucket/autopilot-output/{job_name}/"
 "sagemaker-automl-candidates/best-candidate/"
 "sagemaker_automl_data_processing_script.py"
)

processor = ScriptProcessor(
 image_uri=(
  f"763104351884.dkr.ecr.{REGION}.amazonaws.com/sklearn:1.2-1"
 ),
 command=["python3"],
 instance_type="ml.m5.xlarge",
 instance_count=1,
 role=ROLE,
)

data_transform_step = ProcessingStep(
 name="AutopilotDataTransform",
 processor=processor,
 inputs=[],# 生本番データを指定
 outputs=[],  # 変換済みデータの出力先を指定
 code=data_transform_script,
)

# Autopilot 最良候補のコンテナと HyperParameter を流用して TrainingStep を構築
estimator = Estimator(
 image_uri=best_candidate["InferenceContainers"][0]["Image"],
 instance_type="ml.m5.2xlarge",
 instance_count=1,
 role=ROLE,
 hyperparameters=best_candidate["CandidateSteps"][-1].get(
  "CandidateStepArguments", {}
 ),
)

train_step = TrainingStep(
 name="AutopilotModelTraining",
 estimator=estimator,
 inputs={
  "train": (
data_transform_step.properties
.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri
  )
 },
)

pipeline = Pipeline(
 name="AutopilotReuseDemo",
 steps=[data_transform_step, train_step],
)
pipeline.upsert(role_arn=ROLE)
print("Pipeline upserted successfully.")

Autopilot 生成コードは本番 Pipeline に移植できる品質で出力される。PoC 段階で Autopilot を使い、本番展開では生成コードを Pipeline に組み込む流れが、探索コストと運用コストの両方を最適化する現実的な戦略だ。

3.6 Terraform による Autopilot 実行ロール設定

Autopilot ジョブには S3・CloudWatch Logs・ECR へのアクセスが必要だ。以下の Terraform コードで最小権限の実行ロールを定義する。

resource "aws_iam_role" "autopilot" {
  name = "sagemaker-autopilot-role"

  assume_role_policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Effect = "Allow"
Principal = { Service = "sagemaker.amazonaws.com" }
Action = "sts:AssumeRole"
 }]
  })
}

resource "aws_iam_role_policy_attachment" "autopilot_full" {
  role = aws_iam_role.autopilot.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonSageMakerFullAccess"
}

resource "aws_iam_policy" "autopilot_s3" {
  name = "autopilot-s3-access"

  policy = jsonencode({
 Version = "2012-10-17"
 Statement = [
{
  Effect= "Allow"
  Action= ["s3:GetObject", "s3:PutObject", "s3:ListBucket"]
  Resource = [
 "arn:aws:s3:::my-bucket",
 "arn:aws:s3:::my-bucket/*",
  ]
},
{
  Effect= "Allow"
  Action= ["kms:Decrypt", "kms:GenerateDataKey"]
  Resource = "arn:aws:kms:ap-northeast-1:123456789012:key/my-key"
}
 ]
  })
}

resource "aws_iam_role_policy_attachment" "autopilot_s3" {
  role = aws_iam_role.autopilot.name
  policy_arn = aws_iam_policy.autopilot_s3.arn
}

KMS キーを使う場合は kms:Decrypt と kms:GenerateDataKey の両方を実行ロールに付与することを忘れずに。これが抜けると Autopilot ジョブが S3 output への書き込みで失敗するが、エラーメッセージは AccessDeniedException ではなく暗号化エラーとして出力されるため原因特定に時間がかかる。

Autopilot で素早く展開したモデルほど、本番環境での品質維持には継続的な監視が不可欠だ。§4 では Autopilot デプロイ後のモデルに対して精度劣化・データドリフトを検出し続ける Model Monitor を解説する。

§4 Model Monitor 本番運用 — Drift detection 4種 × Baseline × Schedule

Model Monitor はデプロイ済みモデルへの入力・出力を継続的にキャプチャし、学習時との統計的差異を自動検出する SageMaker のモデル品質監視サービス。エンドポイントへのデータキャプチャを有効化するだけで 4 種の Drift Monitor をスケジュール実行でき、CloudWatch アラームと組み合わせれば検出から自動再学習まで無人運用できる。

4-1 Drift detection 4種の全体像と優先順位

本番運用での優先順位:

1. Data Quality + Model Quality — 全モデルに必須。データパイプライン異常とモデル劣化の第一検知手段として両方を並走させる
2. Bias Drift — 金融・採用・医療などの規制対象ドメインに必須。GDPR や国内ガイドライン対応で継続稼働が求められる
3. Feature Attribution Drift — 説明可能性が求められる高リスクモデルに追加適用。SHAP 計算はコスト高なため週次実行を基本とする

4-2 Data Quality Monitor — KL divergence による統計分布監視

Data Quality Monitor は推論エンドポイントへの入力データ分布をベースラインと比較する。KL divergence（カルバック・ライブラー情報量）で分布乖離を定量化し、特徴量ごとの欠損率・外れ値比率・データ型不整合を継続的に検出する。

from sagemaker.model_monitor import DefaultModelMonitor
from sagemaker.model_monitor.dataset_format import DatasetFormat

my_monitor = DefaultModelMonitor(
 role=role,
 instance_count=1,
 instance_type="ml.m5.xlarge",
 volume_size_in_gb=20,
 max_runtime_in_seconds=3600,
)

# Step1: トレーニングデータからベースライン統計量を算出
baseline_job = my_monitor.suggest_baseline(
 baseline_dataset="s3://my-bucket/training/baseline.csv",
 dataset_format=DatasetFormat.csv(header=True),
 output_s3_uri="s3://my-bucket/monitor/baseline/",
 wait=True,
)

# Step2: 毎時スケジュールで Data Quality チェックを実行
my_monitor.create_monitoring_schedule(
 monitor_schedule_name="data-quality-hourly",
 endpoint_input=predictor.endpoint_name,
 output_s3_uri="s3://my-bucket/monitor/data-quality/results/",
 statistics=baseline_job.baseline_statistics(),
 constraints=baseline_job.suggested_constraints(),
 schedule_cron_expression="cron(0 * ? * * *)",
)

Terraform での Data Quality ジョブ定義:

resource "aws_sagemaker_data_quality_job_definition" "main" {
  name  = "data-quality-monitor-${var.environment}"
  role_arn = aws_iam_role.monitor_role.arn

  data_quality_app_specification {
 image_uri = "156813124566.dkr.ecr.ap-northeast-1.amazonaws.com/sagemaker-model-monitor-analyzer:latest"
  }

  data_quality_job_input {
 endpoint_input {
endpoint_name = var.endpoint_name
local_path = "/opt/ml/processing/endpointdata"
s3_input_mode = "File"
s3_data_distribution_type = "FullyReplicated"
 }
  }

  data_quality_job_output_config {
 monitoring_outputs {
s3_output {
  local_path  = "/opt/ml/processing/localpath/data-quality"
  s3_uri= "s3://${var.monitor_bucket}/data-quality/"
  s3_upload_mode = "EndOfJob"
}
 }
  }

  job_resources {
 cluster_config {
instance_count = 1
instance_type  = "ml.m5.xlarge"
volume_size_in_gb = 20
 }
  }

  stopping_condition {
 max_runtime_in_seconds = 3600
  }
}

4-3 Model Quality Monitor — 精度劣化の時系列追跡

Model Quality Monitor は予測結果と実際の正解ラベル（グラウンドトゥルース）を照合し、Accuracy・F1・AUC などの品質指標を時系列で追跡する。グラウンドトゥルースの収集タイムラグ（数時間〜数日）を考慮し、Data Quality より長いスケジュール間隔（日次）を設定するのが一般的。

from sagemaker.model_monitor import ModelQualityMonitor, CronExpressionGenerator

model_quality_monitor = ModelQualityMonitor(
 role=role,
 instance_count=1,
 instance_type="ml.m5.xlarge",
 volume_size_in_gb=20,
 max_runtime_in_seconds=1800,
)

# テストデータの予測ラベル + 正解ラベルからベースラインを算出
baseline_job = model_quality_monitor.suggest_baseline(
 baseline_dataset="s3://my-bucket/test/predictions_with_labels.jsonl",
 dataset_format=DatasetFormat.json(lines=True),
 output_s3_uri="s3://my-bucket/monitor/model-quality/baseline/",
 problem_type="BinaryClassification",
 inference_attribute="predicted_label",
 ground_truth_attribute="actual_label",
 wait=True,
)

# 日次モニタリング: グラウンドトゥルース収集後に実行
model_quality_monitor.create_monitoring_schedule(
 monitor_schedule_name="model-quality-daily",
 endpoint_input=predictor.endpoint_name,
 ground_truth_input="s3://my-bucket/ground-truth/",
 problem_type="BinaryClassification",
 output_s3_uri="s3://my-bucket/monitor/model-quality/results/",
 statistics=baseline_job.baseline_statistics(),
 constraints=baseline_job.suggested_constraints(),
 schedule_cron_expression=CronExpressionGenerator.daily(),
)

4-4 Bias Drift Monitor — 公平性バイアス検出（Clarify 連携）

Bias Monitor は SageMaker Clarify と連携し、保護属性（性別・年齢・地域など）に対する予測の公平性を継続監視する。DPL（Demographic Parity Loss）・DPPL などの指標で差別的予測の増大を早期検出し、規制対応や倫理的 AI の継続担保に活用する。

from sagemaker.model_monitor import ModelBiasMonitor
from sagemaker.clarify import BiasConfig, ModelConfig

bias_monitor = ModelBiasMonitor(
 role=role,
 instance_count=1,
 instance_type="ml.m5.xlarge",
 volume_size_in_gb=20,
)

model_config = ModelConfig(
 model_name=my_model_name,
 instance_count=1,
 instance_type="ml.m5.xlarge",
 accept_type="text/csv",
)

bias_config = BiasConfig(
 label_values_or_threshold=[1],
 facet_name="gender",# 保護属性カラム
 facet_values_or_threshold=[0],  # 保護グループ値
)

bias_monitor.create_monitoring_schedule(
 monitor_schedule_name="bias-drift-daily",
 endpoint_input=predictor.endpoint_name,
 ground_truth_input="s3://my-bucket/ground-truth/",
 analysis_config=BiasAnalysisConfig(
  bias_config=bias_config,
  headers=feature_headers,
 ),
 output_s3_uri="s3://my-bucket/monitor/bias/results/",
 schedule_cron_expression=CronExpressionGenerator.daily(),
)

4-5 Feature Attribution Drift — SHAP 重要度変化の監視

Feature Attribution Monitor は SHAP 値の分布変化を週次で追跡する。特定の特徴量の重要度が突然高くなる・低くなる変化は、データパイプライン上流での仕様変更やデータリーケージの兆候として早期発見に役立つ。SHAP 計算は ml.m5.xlarge でも 30 分超かかるため、スケジュールは週次を基本とする。

from sagemaker.model_monitor import ModelExplainabilityMonitor
from sagemaker.clarify import SHAPConfig, ExplainabilityAnalysisConfig

shap_config = SHAPConfig(
 baseline=[baseline_sample_values],  # 背景データのサンプル（SHAP 基準値）
 num_samples=15,
 agg_method="mean_abs",  # 絶対値平均で SHAP ドリフトを算出
 save_local_shap_values=True,
)

explainability_monitor = ModelExplainabilityMonitor(
 role=role,
 instance_count=1,
 instance_type="ml.m5.xlarge",
)

explainability_monitor.create_monitoring_schedule(
 monitor_schedule_name="feature-attr-weekly",
 endpoint_input=predictor.endpoint_name,
 analysis_config=ExplainabilityAnalysisConfig(
  explainability_config=shap_config,
  model_config=model_config,
  headers=feature_headers,
 ),
 output_s3_uri="s3://my-bucket/monitor/explainability/",
 schedule_cron_expression=CronExpressionGenerator.weekly(),
)

4-6 CloudWatch 統合 — 段階アラームと自動再学習トリガー

Model Monitor は監視結果を CloudWatch Metrics に自動エクスポートする。アラームを Warning / Critical の 2 段階に設定し、Critical 検出時は EventBridge 経由で SageMaker Pipelines の再学習を自動起動するフルオートメーション構成が本番標準。

# Warning: 違反件数が閾値超過で通知のみ
resource "aws_cloudwatch_metric_alarm" "data_quality_warning" {
  alarm_name = "model-monitor-data-quality-warning"
  comparison_operator = "GreaterThanThreshold"
  evaluation_periods  = 2
  metric_name= "feature_baseline_drift_check_violation_count"
  namespace  = "aws/sagemaker/Endpoints/data-metrics"
  period  = 3600
  statistic  = "Sum"
  threshold  = 5
  alarm_description= "Data quality: KL divergence exceeded warning threshold"
  alarm_actions = [aws_sns_topic.monitor_warning.arn]

  dimensions = {
 Endpoint  = var.endpoint_name
 MonitoringSchedule = "data-quality-hourly"
  }
}

# Critical: Accuracy が 3 日連続で 80% を下回ったら再学習トリガー
resource "aws_cloudwatch_metric_alarm" "model_quality_critical" {
  alarm_name = "model-monitor-accuracy-critical"
  comparison_operator = "LessThanThreshold"
  evaluation_periods  = 3
  metric_name= "binary_classification_accuracy"
  namespace  = "aws/sagemaker/Endpoints/model-metrics"
  period  = 86400
  statistic  = "Average"
  threshold  = 0.80
  alarm_description= "Model accuracy dropped below 80% for 3 consecutive days"
  alarm_actions = [aws_sns_topic.monitor_critical.arn]

  dimensions = {
 Endpoint = var.endpoint_name
  }
}

# EventBridge: Critical アラーム → SageMaker Pipelines 再学習を自動起動
resource "aws_cloudwatch_event_rule" "retrain_trigger" {
  name = "model-monitor-retrain-trigger"

  event_pattern = jsonencode({
 source= ["aws.cloudwatch"]
 detail-type = ["CloudWatch Alarm State Change"]
 detail = {
alarmName= ["model-monitor-accuracy-critical"]
state_value = ["ALARM"]
 }
  })
}

resource "aws_cloudwatch_event_target" "retrain_pipeline" {
  rule  = aws_cloudwatch_event_rule.retrain_trigger.name
  arn= "arn:aws:sagemaker:ap-northeast-1:${var.account_id}:pipeline/${var.retrain_pipeline_name}"
  role_arn = aws_iam_role.eventbridge_sagemaker_role.arn
}

§4 で監視設計が整ったら、次は Drift 検出を受けた「自動再学習パイプライン」の実装本体である SageMaker Pipelines に移る。

§5 SageMaker Pipelines 本番運用 — MLOps workflow × Step types × Conditional × Lambda steps ★山場2

SageMaker Pipelines は MLOps ワークフローを有向非巡回グラフ（DAG）として定義・実行するマネージドサービス。各 Step をノードとして依存関係を宣言し、データ前処理→学習→評価→条件分岐→デプロイまでのパイプライン全体を単一 Python 定義で管理できる。§4 で設定した Model Monitor の Drift 検出 → EventBridge → Pipelines 再学習の自動化はここで実装する。

5-1 SageMaker Pipelines の基本構造

SageMaker Pipelines の主要コンポーネント:

• Pipeline: Step の有向グラフ + PipelineParameter の集合体。pipeline.upsert() で定義を登録・更新する
• Step: 実行単位。各 Step は入力・出力プロパティを公開し、後続 Step からは step.properties.XXX で参照できる
• PipelineParameter: 実行時に差し替え可能なパラメータ（環境名・閾値・インスタンス種別等）
• ExecutionVariable: パイプライン実行 ID など実行時に自動生成される変数
• PropertyFile: Processing / Training Step の出力 JSON をパース → ConditionStep で評価するための橋渡し

5-2 Step types — 主要6種の役割と Python 実装

① ProcessingStep — データ前処理 / 特徴量エンジニアリング / 評価スクリプト実行

from sagemaker.workflow.steps import ProcessingStep
from sagemaker.processing import ScriptProcessor, ProcessingInput, ProcessingOutput

processor = ScriptProcessor(
 image_uri=processing_image_uri,
 instance_type="ml.m5.xlarge",
 instance_count=1,
 role=role,
)

step_process = ProcessingStep(
 name="DataPreprocessing",
 processor=processor,
 inputs=[
  ProcessingInput(
source="s3://my-bucket/raw/",
destination="/opt/ml/processing/input",
  )
 ],
 outputs=[
  ProcessingOutput(
output_name="train",
source="/opt/ml/processing/output/train",
destination="s3://my-bucket/processed/train/",
  ),
  ProcessingOutput(
output_name="validation",
source="/opt/ml/processing/output/validation",
destination="s3://my-bucket/processed/validation/",
  ),
 ],
 code="scripts/preprocess.py",
 cache_config=CacheConfig(enable_caching=True, expire_after="7d"),
)

② TrainingStep — モデル学習（EstimatorWrapper）

from sagemaker.workflow.steps import TrainingStep
from sagemaker.estimator import Estimator
from sagemaker.inputs import TrainingInput

estimator = Estimator(
 image_uri=training_image_uri,
 instance_type="ml.p3.2xlarge",
 instance_count=1,
 output_path="s3://my-bucket/model-artifacts/",
 role=role,
 hyperparameters={
  "max-depth": max_depth_param, # PipelineParameter
  "n-estimators": n_estimators_param,
 },
)

step_train = TrainingStep(
 name="ModelTraining",
 estimator=estimator,
 inputs={
  "train": TrainingInput(
s3_data=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
content_type="text/csv",
  ),
 },
 cache_config=CacheConfig(enable_caching=True, expire_after="7d"),
)

③ ModelStep — モデル登録（Model Registry + 手動承認ゲート）

from sagemaker.workflow.model_step import ModelStep
from sagemaker.model import Model

model = Model(
 image_uri=inference_image_uri,
 model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
 role=role,
)

model_register_args = model.register(
 content_types=["text/csv"],
 response_types=["text/csv"],
 inference_instances=["ml.m5.xlarge"],
 transform_instances=["ml.m5.xlarge"],
 model_package_group_name="my-model-group",
 approval_status="PendingManualApproval",  # 手動承認ゲート必須
 model_metrics=model_metrics,
)

step_register = ModelStep(name="RegisterModel", step_args=model_register_args)

④ ConditionStep — 精度閾値による条件分岐（AccuracyGate）

from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.functions import JsonGet

# 評価スクリプト (ProcessingStep) が出力した accuracy.json を参照
accuracy_value = JsonGet(
 step_name=step_evaluate.name,
 property_file=evaluation_report,
 json_path="classification_metrics.accuracy.value",
)

cond_gte = ConditionGreaterThanOrEqualTo(
 left=accuracy_value,
 right=accuracy_threshold,# PipelineParameter (float)
)

step_cond = ConditionStep(
 name="AccuracyGate",
 conditions=[cond_gte],
 if_steps=[step_register], # 条件 True → モデル登録
 else_steps=[step_fail],# 条件 False → パイプライン失敗
)

⑤ LambdaStep — 外部 API 呼び出し / 通知

from sagemaker.workflow.lambda_step import LambdaStep, LambdaOutput, LambdaOutputTypeEnum
from sagemaker.lambda_helper import Lambda

step_notify = LambdaStep(
 name="SlackNotification",
 lambda_func=Lambda(
  function_name="pipeline-notification",
  execution_role_arn=lambda_role_arn,
 ),
 inputs={
  "model_name": step_register.properties.ModelPackageName,
  "accuracy": accuracy_value,
  "environment": environment_param,
 },
 outputs=[
  LambdaOutput(
output_name="notification_status",
output_type=LambdaOutputTypeEnum.String,
  )
 ],
)

⑥ TuningStep — ハイパーパラメータ最適化（HPO）

from sagemaker.workflow.steps import TuningStep
from sagemaker.tuner import HyperparameterTuner, IntegerParameter, ContinuousParameter

tuner = HyperparameterTuner(
 estimator=estimator,
 objective_metric_name="validation:auc",
 objective_type="Maximize",
 hyperparameter_ranges={
  "max-depth": IntegerParameter(3, 10),
  "learning-rate": ContinuousParameter(0.001, 0.1),
 },
 max_jobs=20,
 max_parallel_jobs=4,
)

step_tuning = TuningStep(
 name="HPOTuning",
 tuner=tuner,
 inputs={
  "train": TrainingInput(
s3_data=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri
  ),
 },
)

5-3 Pipeline 全体定義とパラメータ管理

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.parameters import ParameterString, ParameterFloat, ParameterInteger

environment_param= ParameterString(name="Environment", default_value="staging")
accuracy_threshold  = ParameterFloat(name="AccuracyThreshold", default_value=0.85)
max_depth_param  = ParameterInteger(name="MaxDepth", default_value=6)
n_estimators_param  = ParameterInteger(name="NEstimators", default_value=100)

pipeline = Pipeline(
 name="mlops-training-pipeline",
 parameters=[
  environment_param,
  accuracy_threshold,
  max_depth_param,
  n_estimators_param,
 ],
 steps=[step_process, step_train, step_evaluate, step_cond, step_notify],
 sagemaker_session=sagemaker_session,
)

pipeline.upsert(role_arn=role)

5-4 Pipeline 実行制御 — EventBridge スケジュールと Model Monitor 連携トリガー

# EventBridge Scheduler: 毎週月曜 JST 02:00 に定期再学習
resource "aws_scheduler_schedule" "weekly_retrain" {
  name = "mlops-weekly-retrain"
  group_name = "mlops-schedules"

  flexible_time_window {
 mode = "OFF"
  }

  schedule_expression = "cron(0 17 ? * SUN *)"
  schedule_expression_timezone = "Asia/Tokyo"

  target {
 arn= "arn:aws:sagemaker:ap-northeast-1:${var.account_id}:pipeline/mlops-training-pipeline"
 role_arn = aws_iam_role.scheduler_role.arn

 sagemaker_pipeline_parameters {
pipeline_parameter {
  name  = "Environment"
  value = "production"
}
pipeline_parameter {
  name  = "AccuracyThreshold"
  value = "0.85"
}
 }
  }
}

# Pipeline 実行ロール
resource "aws_iam_role" "pipeline_execution_role" {
  name = "sagemaker-pipeline-execution-role"

  assume_role_policy = jsonencode({
 Version = "2012-10-17"
 Statement = [{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = { Service = "sagemaker.amazonaws.com" }
 }]
  })
}

resource "aws_iam_role_policy_attachment" "pipeline_sagemaker" {
  role = aws_iam_role.pipeline_execution_role.name
  policy_arn = "arn:aws:policies/AmazonSageMakerFullAccess"
}

SageMaker Pipelines 公式ドキュメント

§5 のパイプライン実行基盤が整ったら、次の §6 では学習・推論の両フェーズで使う特徴量の一元管理基盤 Feature Store を解説する。

§6 Feature Store 本番運用 — Online/Offline store × Feature group × Time travel

SageMaker Feature Store は、機械学習フィーチャーを一元管理するマネージドリポジトリです。Online Store（低レイテンシ推論用）と Offline Store（学習用S3）を同一 Feature Group で管理し、学習と推論で使うフィーチャーの乖離（Training-Serving Skew）を根絶できます。

6-1 Online Store vs Offline Store の役割

推論パイプラインでは get_record() で Online Store から最新フィーチャーを取得し、学習時は Athena で Offline Store を point-in-time クエリします。この二刀流が Training-Serving Skew 防止の核心です。

6-2 Feature Group 設計

Feature Group は Feature Store の基本管理単位です。設計時に決める3要素が後から変更不可のため、最初に正しく設計する必要があります。

必須設計項目:
– Record Identifier: フィーチャーセットを一意識別するキー（例: user_id, product_id）
– Event Time: フィーチャーが生成された時刻。Time travel の基準点
– Feature Definitions: フィーチャー名・型（Integral / Fractional / String）のスキーマ

import boto3
import time
import sagemaker
from sagemaker.feature_store.feature_group import FeatureGroup

sm_session = sagemaker.Session()
role = sagemaker.get_execution_role()

feature_group = FeatureGroup(
 name="user-purchase-features",
 sagemaker_session=sm_session
)

feature_group.create(
 s3_uri="s3://my-feature-store-bucket/user-purchase-features",
 record_identifier_name="user_id",
 event_time_feature_name="event_time",
 role_arn=role,
 enable_online_store=True,
 description="ユーザー購買行動フィーチャー (7日間集計)",
)

6-3 フィーチャーの書き込み・読み取り

featurestore_runtime = boto3.client("sagemaker-featurestore-runtime")

# フィーチャー書き込み (Online + Offline 同時)
featurestore_runtime.put_record(
 FeatureGroupName="user-purchase-features",
 Record=[
  {"FeatureName": "user_id","ValueAsString": "u_12345"},
  {"FeatureName": "event_time","ValueAsString": str(time.time())},
  {"FeatureName": "purchase_count_7d",  "ValueAsString": "8"},
  {"FeatureName": "avg_order_value", "ValueAsString": "4850.0"},
  {"FeatureName": "last_category","ValueAsString": "electronics"},
 ]
)

# Online Store からリアルタイム取得
response = featurestore_runtime.get_record(
 FeatureGroupName="user-purchase-features",
 RecordIdentifierValueAsString="u_12345",
 FeatureNames=["purchase_count_7d", "avg_order_value"]
)
record = {f["FeatureName"]: f["ValueAsString"] for f in response["Record"]}
# -> {'purchase_count_7d': '8', 'avg_order_value': '4850.0'}

6-4 Time Travel — Point-in-time Correctness

学習データ生成時に「そのラベルが付いた時点のフィーチャー値」を使わないとデータリークが発生します。Feature Store の Offline Store は全イベント履歴を S3 Parquet に蓄積するため、Athena で任意時点のスナップショットを取得できます。

-- Athena: 2026-04-01 時点のユーザーフィーチャーを point-in-time 取得
SELECT
 user_id,
 purchase_count_7d,
 avg_order_value,
 last_category
FROM "sagemaker_featurestore"."user_purchase_features_1234567890"
WHERE
 event_time <= 1743465600.0
 AND (is_deleted IS NULL OR is_deleted = 'false')
QUALIFY ROW_NUMBER() OVER (
 PARTITION BY user_id
 ORDER BY event_time DESC
) = 1;

このクエリにより、学習ラベルの収集日時と一致するフィーチャー値のみを使用でき、将来のデータが混入しない訓練セットが完成します。

6-5 SDK 詳細 — delete_record / describe / list / batch_get

put_record / get_record 以外にも本番運用で頻出する SDK API を整理します。

# レコード削除 (論理削除: is_deleted=true フラグ追加)
featurestore_runtime.delete_record(
 FeatureGroupName="user-purchase-features",
 RecordIdentifierValueAsString="u_12345",
 EventTime=str(time.time()),
)

# Feature Group の詳細取得 (スキーマ・S3 URI・Online Store 設定 確認)
sm_client = boto3.client("sagemaker")
desc = sm_client.describe_feature_group(FeatureGroupName="user-purchase-features")
print(desc["RecordIdentifierFeatureName"])  # -> "user_id"
print(desc["OfflineStoreConfig"]["S3StorageConfig"]["S3Uri"])
print(desc["FeatureGroupStatus"])  # -> "Created" / "Creating" / "Failed"

# Feature Group の一覧取得 (ページネーション対応)
paginator = sm_client.get_paginator("list_feature_groups")
for page in paginator.paginate():
 for fg in page["FeatureGroupSummaries"]:
  print(fg["FeatureGroupName"], fg["FeatureGroupStatus"])

# 複数レコード同時取得 (BatchGetRecord: 最大100件)
batch_response = featurestore_runtime.batch_get_record(
 Identifiers=[
  {
"FeatureGroupName": "user-purchase-features",
"RecordIdentifiersValueAsString": ["u_12345", "u_67890", "u_11111"],
"FeatureNames": ["purchase_count_7d", "avg_order_value"],
  }
 ]
)
for record in batch_response["Records"]:
 print(record["FeatureGroupName"], record["RecordIdentifierValueAsString"])

delete_record の重要な性質:
– Online Store: 論理削除（is_deleted=true フラグ。レコード自体は残る）
– Offline Store: S3 Parquet に「削除イベント」として追記。物理削除されない
– Athena クエリでは WHERE (is_deleted IS NULL OR is_deleted = 'false') で除外必須

6-6 統合パターン — Pipelines / Lambda / Endpoint 推論

Feature Store は単独で使うのではなく、SageMaker Pipelines / Lambda / Endpoint 推論と統合してこそ本番運用の真価を発揮します。

パターン1: Pipelines で学習データ生成

from sagemaker.workflow.feature_store_step import FeatureStoreOutput
from sagemaker.workflow.steps import ProcessingStep

# Pipelines の Processing Step で Athena クエリ → Feature Store Offline Store から
# point-in-time correctness の学習データセットを Parquet で出力
processing_step = ProcessingStep(
 name="GenerateTrainingDataset",
 processor=sklearn_processor,
 code="generate_training_data.py",
 inputs=[
  ProcessingInput(
source="s3://my-feature-store-bucket/user-purchase-features",
destination="/opt/ml/processing/input",
  )
 ],
 outputs=[
  ProcessingOutput(
source="/opt/ml/processing/output",
destination="s3://my-training-data/v1",
  )
 ],
)

パターン2: Lambda で推論前フィーチャー取得

# Lambda Function (Python 3.12 runtime)
import boto3
import json

featurestore_runtime = boto3.client("sagemaker-featurestore-runtime")
sagemaker_runtime = boto3.client("sagemaker-runtime")

def lambda_handler(event, context):
 user_id = event["user_id"]

 # 1. Feature Store Online Store から最新フィーチャー取得 (<10ms)
 fs_response = featurestore_runtime.get_record(
  FeatureGroupName="user-purchase-features",
  RecordIdentifierValueAsString=user_id,
  FeatureNames=["purchase_count_7d", "avg_order_value", "last_category"],
 )
 features = {f["FeatureName"]: f["ValueAsString"] for f in fs_response["Record"]}

 # 2. SageMaker Endpoint で推論
 payload = json.dumps([float(features["purchase_count_7d"]),
  float(features["avg_order_value"])])
 inference = sagemaker_runtime.invoke_endpoint(
  EndpointName="user-churn-predictor",
  ContentType="application/json",
  Body=payload,
 )
 return {"prediction": json.loads(inference["Body"].read())}

パターン3: Endpoint 推論時に Feature Store 直接参照

Endpoint に組み込んだ推論コンテナ（カスタム inference.py）から直接 Feature Store API を叩く設計。Lambda 経由のホップを省略しレイテンシを短縮できます。ただし Endpoint の IAM Role に sagemaker:GetRecord 権限の付与が必須です。

6-7 Online Store パフォーマンスチューニング

Online Store のレイテンシと throughput を最大化するための実装テクニックです。

並列 get_record による throughput 向上:

import concurrent.futures
import boto3

def fetch_features(user_id):
 fs = boto3.client("sagemaker-featurestore-runtime")
 response = fs.get_record(
  FeatureGroupName="user-purchase-features",
  RecordIdentifierValueAsString=user_id,
 )
 return user_id, response.get("Record", [])

user_ids = ["u_001", "u_002", "u_003", "u_004", "u_005"]
with concurrent.futures.ThreadPoolExecutor(max_workers=10) as executor:
 results = list(executor.map(fetch_features, user_ids))

BatchGetRecord の活用 (最大100件/リクエスト):

5件以上をまとめて取得する場合は batch_get_record の方がコスト・レイテンシともに有利です。HTTP オーバーヘッドを削減でき、100ms 程度で 100件のフィーチャーが揃います。

DynamoDB プロビジョン容量設計:

Online Store のバックエンドは DynamoDB です。デフォルトは On-Demand キャパシティですが、ピーク負荷が予測可能なら Provisioned に切り替えてコスト最適化が可能です。レコードあたり 0.5KB 想定で、推論 QPS から所要 RCU を逆算します。

§6では Feature Store の設計・実装パターン・統合パターン・パフォーマンスチューニングを解説しました。次の §7 では、SageMaker 本番運用で実際に踏まれやすい「詰まりポイント7選」を判断ツリーで体系化します。

§7 詰まりポイント7選 図解 (Mermaid01 判断ツリー)

SageMaker を本番運用する際に現場で繰り返し発生する詰まりポイントを7パターンに体系化します。以下の判断ツリーで自分の症状と原因を素早く特定してください。

flowchart TD
  A[SageMaker問題発生] --> B{Studio課金?}
  B -->|Yes| C[KernelGateway app停止確認]
  B -->|No| D{Autopilot失敗?}
  D -->|Yes| E{モード選択ミス?}
  E -->|Yes| F[Tabular/Text/Image確認]
  E -->|No| G[MaxCandidates/データ量確認]
  D -->|No| H{Monitor違反?}
  H -->|Yes| I[Baseline統計ファイル確認]
  H -->|No| J{Pipelines重複実行?}
  J -->|Yes| K[CacheConfig設定確認]
  J -->|No| L{Conditional分岐ズレ?}
  L -->|Yes| M[GreaterThan vs GreaterThanOrEqualTo確認]
  L -->|No| N[Feature Store同期ラグ確認]

詰まり1: Studio — KernelGateway app 停止し忘れで課金継続

症状: SageMaker Studio を閉じた後も ml.p3.2xlarge 等の高コストインスタンスの課金が止まらない。

原因: Studio の「ブラウザタブを閉じる」はアプリを停止しない。KernelGateway app は明示的に停止するまで稼働し続ける。

解決策:

import boto3

sm = boto3.client("sagemaker")

# 稼働中アプリを一覧表示
apps = sm.list_apps(DomainIdEquals="d-xxxxxxxxxx")
for app in apps["Apps"]:
 if app["Status"] == "InService" and app["AppType"] == "KernelGateway":
  sm.delete_app(
DomainId=app["DomainId"],
UserProfileName=app["UserProfileName"],
AppType=app["AppType"],
AppName=app["AppName"],
  )

詰まり2: Autopilot — Candidate 数上限（250）に達しても精度改善しない

症状: Autopilot ジョブが 250 Candidate まで試したが、最良 Candidate のスコアが要件を満たさない。

原因: デフォルトの Tabular モードで試行する変換・アルゴリズムの組み合わせが、データの性質と合っていない場合がある。

解決策: 特徴量エンジニアリングを強化した前処理済みデータを再投入し、FeatureSpecificationS3Uri で手動フィーチャー定義を追加する。または Autopilot をやめて Training Step + HPO Step を組み合わせた Pipelines に移行する。

詰まり3: Autopilot — Tabular / Text / Image モード誤選択で全件 FAIL

症状: Autopilot ジョブが開始直後に全 Candidate で Failed となり CloudWatch Logs にデータ形式エラーが出る。

原因: CSV 表形式データに Text モードを、画像データに Tabular モードを選択した等のモード誤選択。モードはデータ形式と予測ターゲットに合わせて選ぶ必要がある。

解決策:

詰まり4: Model Monitor — Baseline 統計ファイル未生成で Monitor 開始 → 常に Violation

症状: Model Monitor スケジュールを有効化したら全サンプルで Violation アラートが発生し続ける。

原因: suggest_baseline() を実行せずに create_monitoring_schedule() を呼び出した、または Baseline Processing Job が完了前にスケジュールを開始した。Baseline 統計ファイル（statistics.json / constraints.json）が S3 に存在しないと Monitor は常に違反判定する。

解決策:

from sagemaker.model_monitor import DefaultModelMonitor

monitor = DefaultModelMonitor(role=role, instance_type="ml.m5.xlarge")

# Baseline 統計ファイルを生成（必ず待機する）
monitor.suggest_baseline(
 baseline_dataset="s3://my-bucket/baseline/train_data.csv",
 dataset_format={"csv": {"header": True}},
 output_s3_uri="s3://my-bucket/baseline-results/",
 wait=True,# 完了を待つ
)

# Baseline 完了後にスケジュール設定
monitor.create_monitoring_schedule(...)

詰まり5: Pipelines — CacheConfig 無効で同一 ProcessingStep が毎回フル実行（コスト2〜5倍）

症状: Pipeline を再実行するたびに、入力データが変わっていないのにデータ前処理 Step が毎回数十分かかる。

原因: CacheConfig のデフォルトは enable_caching=False。同一入力・同一コードの Step でもキャッシュを使わず毎回フル実行する。

解決策:

from sagemaker.workflow.steps import CacheConfig

cache_config = CacheConfig(
 enable_caching=True,
 expire_after="P7D"  # 7日間キャッシュ有効
)

processing_step = ProcessingStep(
 name="DataPreprocessing",
 processor=processor,
 inputs=[...],
 outputs=[...],
 cache_config=cache_config,# キャッシュを有効化
)

詰まり6: Pipelines — ConditionStep の比較演算子誤りで分岐がずれる

症状: 精度 0.80 を閾値にモデルデプロイするはずが、0.80 ちょうどで閾値判定が「デプロイしない」側に分岐してしまう。

原因: GreaterThan は > の判定であり >= ではない。精度が閾値と等値の場合は False になる。

解決策:

from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.condition_step import ConditionStep

# 誤り: ConditionGreaterThan → 0.80 exactly だと False
# 正解: ConditionGreaterThanOrEqualTo → 0.80 以上が True
condition = ConditionGreaterThanOrEqualTo(
 left=JsonGet(
  step_name=evaluation_step.name,
  property_file=evaluation_report,
  json_path="regression_metrics.accuracy.value",
 ),
 right=0.80,
)

deploy_condition_step = ConditionStep(
 name="CheckAccuracy",
 conditions=[condition],
 if_steps=[deploy_step],
 else_steps=[register_as_rejected_step],
)

詰まり7: Feature Store — Online/Offline 同期ラグ（15分）を無視した推論精度劣化

症状: バッチ学習直後にモデルをデプロイしたが、直近15分以内のユーザー行動フィーチャーを使った推論の精度が学習時より劣化する。

原因: put_record() で書き込まれたフィーチャーは Offline Store（S3）への反映に最大15分かかる。学習データ生成時に直近フィーチャーが Offline Store に存在せず、推論時の Online Store とのズレが生じる。

解決策: 学習データ生成のタイムカット（point-in-time クエリ）を now - 30分 にズラし、同期ラグを確実に吸収する。または高頻度更新フィーチャーは Online Store のみで管理し、学習には DynamoDB Stream → S3 のバイパスパスを構築する。

§8 アンチパターン → 正解パターン変換演習 + AI三部作完成 + 44記事化告知

§1〜§7 で学んだ知識を実践で定着させるため、SageMaker 本番運用で頻出する5つのアンチパターンを正解パターンに変換する演習を行います。

アンチパターン演習 (5問)

Q1: Notebook 単独運用 → SageMaker Studio Domain 管理

状況: データサイエンティストが各自の EC2 や ローカル Jupyter でモデルを開発・学習しており、誰がどのモデルを学習したか管理できていない。

アンチパターン:

# 個人 EC2 で直接実行 — 誰が何を実行したか追跡不可
import sklearn
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)
joblib.dump(model, "./model.pkl")  # S3 管理なし

正解パターン:

from sagemaker.sklearn import SKLearn

# Studio + Training Job で実行履歴・成果物を一元管理
estimator = SKLearn(
 entry_point="train.py",
 role=role,
 instance_type="ml.m5.xlarge",
 framework_version="1.2-1",
 output_path="s3://my-bucket/model-artifacts/",
)
estimator.fit({"train": "s3://my-bucket/data/train/"})

Studio Domain 管理により、実行ユーザー・インスタンス・コード・成果物がすべて追跡可能になります。

Q2: 手動再学習 → Pipelines 自動化（EventBridge + ScheduledTrigger）

状況: モデルの月次再学習を担当者が手動で Notebook を開いて実行している。担当者不在時に再学習が漏れた。

アンチパターン: 担当者が毎月末に手動で Notebook を実行。

正解パターン:

from sagemaker.workflow.pipeline import Pipeline

# Pipeline を定義し EventBridge で月次自動実行
pipeline = Pipeline(
 name="MonthlyRetrainPipeline",
 steps=[preprocessing_step, training_step, evaluation_step, deploy_step],
)
pipeline.upsert(role_arn=role)

# EventBridge Scheduler で月次トリガー（コンソールまたは AWS CLI で設定）
# aws events put-rule --schedule-expression "cron(0 2 1 * ? *)" ...

Pipelines + EventBridge により、担当者不在でも月初に自動で再学習・再デプロイが完結します。

Q3: 統計監視なし → Model Monitor（Data Quality + Model Quality）

状況: 本番モデルの入力データ分布が変化してもアラートがなく、気づかずに精度劣化が進行していた。

アンチパターン: CloudWatch でエンドポイントの呼び出し回数のみ監視。データ品質の監視なし。

正解パターン:

from sagemaker.model_monitor import DefaultModelMonitor, CronExpressionGenerator

monitor = DefaultModelMonitor(role=role, instance_type="ml.m5.xlarge")
monitor.suggest_baseline(
 baseline_dataset="s3://my-bucket/baseline/train.csv",
 dataset_format={"csv": {"header": True}},
 output_s3_uri="s3://my-bucket/baseline-results/",
 wait=True,
)
monitor.create_monitoring_schedule(
 monitor_schedule_name="daily-data-quality-check",
 endpoint_input=predictor.endpoint_name,
 output_s3_uri="s3://my-bucket/monitor-reports/",
 statistics=monitor.baseline_statistics(),
 constraints=monitor.suggested_constraints(),
 schedule_cron_expression=CronExpressionGenerator.daily(),
)

Data Quality Monitor が入力データの統計的変化を自動検出し、CloudWatch Alarm でアラートを発報します。

Q4: Feature 重複計算 → Feature Store（一元管理 + Time travel）

状況: 「過去7日の購買件数」フィーチャーを学習パイプラインと推論 Lambda の両方で独立計算しており、計算ロジックの乖離で Training-Serving Skew が発生。

アンチパターン:

# 学習時: pandas で計算
train_df["purchase_7d"] = df.groupby("user_id")["purchase"].rolling("7D").sum()

# 推論時: Lambda で独自計算（ロジックが微妙にズレる）
purchase_7d = query_dynamo_for_recent_purchases(user_id, days=7)

正解パターン: Feature Store に purchase_count_7d を一元登録し、学習は Offline Store から、推論は Online Store から取得する（§6 参照）。フィーチャー計算ロジックが一箇所に集約されるため、Training-Serving Skew が構造的に発生しなくなります。

Q5: SageMaker Endpoint 直接運用 → Pipelines + Model Registry 連携デプロイ

状況: 新しいモデルを学習するたびに手動でエンドポイントを更新しており、性能劣化モデルが誤って本番に出てしまう事故が発生した。

アンチパターン: 学習後に手動で predictor.deploy() を実行。

正解パターン:

from sagemaker.workflow.model_step import ModelStep
from sagemaker.workflow.steps import ConditionStep

# 1. 精度評価 Step で閾値チェック
# 2. 合格した場合のみ Model Registry に登録
# 3. 手動承認後にエンドポイント更新（承認ゲート）
register_step = ModelStep(
 name="RegisterModel",
 step_args=model.register(
  content_types=["text/csv"],
  response_types=["text/csv"],
  model_package_group_name="MyModelGroup",
  approval_status="PendingManualApproval",  # 承認ゲート
 ),
)

Model Registry の承認ゲートにより、ML エンジニアまたは自動テストが合格した場合のみ本番デプロイが進みます。

graph LR
  V1[AI Vol1<br/>Bedrock Agents基礎] --> V2[AI Vol2<br/>生成AI 4本柱]
  V2 --> V3[AI Vol3<br/>SageMaker MLOps]
  V3 --> IAM[IAM Vol1-4]
  V3 --> SEC[Security Vol1-3]
  V3 --> OBS[Observability Vol1-2]
  V3 --> CON[Container Vol1-3]
  V3 --> ANL[Analytics Vol1]

全軸クロスリンクナビ

SageMaker MLOps は AWS の多数のサービスと連携するため、以下の関連シリーズと合わせて読むことで本番設計力が強化されます。

AI/ML Vol1 — Bedrock Agents本番運用基礎

AI/ML Vol2 — Bedrock生成AI 4本柱 (Embedding×RAG×KB×Agents)

IoT本番運用 Vol1 — IoT Core × Greengrass × SiteWise × Device Defender

SageMaker MLOps を本番運用する際は、IAM ロール設計（Studio Execution Role / Pipelines Role）・KMS 暗号化（Feature Store / S3 モデルアーティファクト）・VPC 接続（SageMaker VPC Endpoint）・CloudWatch 監視（Model Monitor アラート連携）・S3 ストレージ設計（Feature Store Offline / モデルアーティファクト）が不可欠です。各ドメインのシリーズ記事と組み合わせて設計を固めてください。