生成AI・ML基盤

SageMaker Pipelinesで月40時間の手作業が消えた。本番ML自動化6ヶ月の実装記録

データ検証・モデル評価・デプロイを完全自動化。手動作業に悩むMLチームが踏んだ地雷と、実際に動いた構成設計をまとめました。

AWSSageMakerMLOpsCI/CDML自動化

SageMaker Pipelinesを本番導入するまでの道のり

先日プロジェクトで、データサイエンスチームの月40時間の手作業を完全に消すためにSageMaker Pipelinesを導入しました。正直、最初は「これ本当に動くの？」と懐疑的だったんですけど、6ヶ月本番運用してみると、ML開発のワークフローが劇的に変わったんですよね。

うちのチームは従来、データの品質チェック・モデルの学習・検証・本番デプロイまでの一連のプロセスを、ほぼ全部手動でやってました。Jupyter Notebookで実験→結果が良かったらスクリプト化→クローンとタスク管理ツールにメモ→週1回のレビュー会で手動でデプロイ、みたいな感じです。データが更新されるたびにこれを繰り返してるわけですから、時間がかかるのは当たり前です。

そこで「これはPipelineを組むしかない」と決断したんですが、実装してみたら想像以上に複雑でした。データ検証・特徴量生成・モデル学習・自動評価・条件付きデプロイ——この全てをつなぎ込むのって、実は業界のベストプラクティスがまだ定着していないんですよね。試行錯誤しながら本番環境を構築したからこそ、得た知見が沢山あります。

SageMaker Pipelinesの全体像と実装パターン

SageMaker Pipelinesって、その名の通りML処理の一連のステップをDAG（有向非環循グラフ）として定義して、自動で実行・監視・ロギングするサービスです。2026年時点で、AWS側も大幅にアップデートしてて、DataQualityチェック・自動バイアス検出・モデル監視がかなり統合されました。

従来のやり方だと、「学習が終わったから手動でモデルを評価する」「良い結果だったら本番環境に上げる」みたいに、人間の判断をはさむ余地が大量にありました。でもPipelineを使うと、この判断ロジックそのものをコードで表現できるんです。機械が淡々と実行してくれるから、人間のミスが入る余地がぐっと減ります。

うちが実装した構成は、ざっくりこんな感じです：

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep, CreateModelStep
from sagemaker.workflow.conditions import ConditionGreaterThan
from sagemaker.workflow.properties import PropertyFile
from sagemaker.workflow.step_outputs import ProcessingOutput
import json

# 1. データ検証ステップ（DataQuality チェック）
from sagemaker.processing import ScriptProcessor

data_validation_processor = ScriptProcessor(
    role=role,
    image_uri="246618743249.dkr.ecr.us-east-1.amazonaws.com/sagemaker-scikit-learn:0.23-1-cpu-py3",
    instance_count=1,
    instance_type="ml.m5.xlarge",
)

data_validation_step = ProcessingStep(
    name="DataValidation",
    processor=data_validation_processor,
    code="validation_script.py",
    job_arguments=["--input-data", s3_input_path],
    outputs=[ProcessingOutput(output_name="validation_output", source="/opt/ml/processing/output")],
)

# 2. 特徴量エンジニアリング
feature_engineering_step = ProcessingStep(
    name="FeatureEngineering",
    processor=data_validation_processor,
    code="feature_engineering.py",
    job_arguments=["--input-data", s3_input_path],
    outputs=[ProcessingOutput(output_name="features", source="/opt/ml/processing/features")],
)

# 3. モデル学習
from sagemaker.estimator import Estimator

model_estimator = Estimator(
    image_uri="246618743249.dkr.ecr.us-east-1.amazonaws.com/sagemaker-xgboost:1.7-1-cpu-py3",
    role=role,
    instance_count=1,
    instance_type="ml.m5.xlarge",
    output_path=model_output_path,
)

training_step = TrainingStep(
    name="ModelTraining",
    estimator=model_estimator,
    inputs={
        "training": f"{s3_features_path}/train",
        "validation": f"{s3_features_path}/validation",
    },
)

# 4. モデル評価（カスタムメトリクス）
from sagemaker.processing import ScriptProcessor

evaluation_processor = ScriptProcessor(
    role=role,
    image_uri="246618743249.dkr.ecr.us-east-1.amazonaws.com/sagemaker-scikit-learn:0.23-1-cpu-py3",
    instance_count=1,
    instance_type="ml.m5.xlarge",
)

evaluation_step = ProcessingStep(
    name="ModelEvaluation",
    processor=evaluation_processor,
    code="evaluate_model.py",
    job_arguments=["--model-path", training_step.properties.ModelArtifacts.S3ModelArtifacts],
    outputs=[ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation")],
)

# 5. 条件付きステップ（AUC > 0.85 なら本番へ）
from sagemaker.workflow.conditions import ConditionGreaterThan
from sagemaker.workflow.condition_step import ConditionStep

condition_step = ConditionStep(
    name="CheckModelQuality",
    conditions=[ConditionGreaterThan(
        left=JsonGet(
            step=evaluation_step,
            property_file=PropertyFile(name="EvaluationMetrics", output_name="evaluation", json_path="$.metrics.auc")
        ),
        right=0.85
    )],
    if_steps=[create_model_step, create_endpoint_step],  # AUC > 0.85 なら実行
    else_steps=[],  # そうでなければスキップ
)

# 6. パイプライン全体を定義
pipeline = Pipeline(
    name="MLAutomationPipeline",
    parameters=[],
    steps=[data_validation_step, feature_engineering_step, training_step, evaluation_step, condition_step],
)

pipeline.upsert(role_arn=role_arn)
pipeline.start()

このコードを見ると、処理が直列でつなぎ込まれているのが分かります。データ検証→特徴量生成→モデル学習→評価→条件判定、という流れですね。実行順序がコードで明示されてるから、後から見返しても何をやってるかすぐに分かる。そこが地味に便利なんです。

AWS構成図：SageMaker Pipelinesの全体像

graph TB
    subgraph "AWS Account"
        subgraph "SageMaker Pipelines"
            DataVal["📊 Data Validation<br/>DataQuality チェック"]
            FeatureEng["🔧 Feature Engineering<br/>特徴量生成"]
            Training["🤖 Model Training<br/>XGBoost/TensorFlow"]
            Evaluation["✅ Model Evaluation<br/>AUC・Precision検証"]
            Condition{"AUC > 0.85?"}
            ModelRegistry["📦 Model Registry<br/>バージョン管理"]
            Deploy["🚀 Auto Deploy<br/>SageMaker Endpoint"]
        end
        
        subgraph "Data Layer"
            S3Raw["🗄️ S3 Raw Data<br/>CSV/Parquet"]
            S3Features["🗄️ S3 Features<br/>処理済みデータ"]
            S3Models["🗄️ S3 Models<br/>モデルArtifacts"]
        end
        
        subgraph "Monitoring & Logging"
            CloudWatch["📈 CloudWatch Logs<br/>Pipeline実行ログ"]
            ModelMonitor["🔍 Model Monitor<br/>データドリフト検出"]
        end
    end
    
    DataVal --> S3Raw
    DataVal --> FeatureEng
    FeatureEng --> S3Features
    FeatureEng --> Training
    Training --> S3Models
    Training --> Evaluation
    Evaluation --> Condition
    Condition -->|YES| ModelRegistry
    Condition -->|NO| CloudWatch
    ModelRegistry --> Deploy
    Deploy --> ModelMonitor
    CloudWatch --> ModelMonitor

実装で痛い目を見たポイント3つ

1. データ品質チェックをなめてた

最初、「データ検証くらい簡単だろ」と思ってたんです。けど本番で走らせてみると、毎日のデータに異常値や欠損が増えてくるんですよね。パイプラインが失敗する度に原因追跡に2時間かかるみたいな状況になって、これは流石にまずいと気づきました。

そこで導入したのが、SageMakerのDataQuality機能です。これは統計的にデータの分布を監視して、「この特徴量の値の範囲がいつもと違う」みたいなアラートを出してくれるんですよ。

from sagemaker.model_monitor import DataQualityBaseline, DataQualityMonitor
from sagemaker.model_monitor.dataset_format import DatasetFormat

# ベースラインデータセット（正常系）で統計情報を学習
baseline = DataQualityBaseline(
    baseline_job_name=baseline_job_name,
    role=role,
    sagemaker_session=session,
)

# 毎日のデータをこのベースラインと比較
monitor = DataQualityMonitor(
    role=role,
    instance_count=1,
    instance_type="ml.m5.xlarge",
    volume_size_in_gb=20,
    max_runtime_in_seconds=1800,
)

monitor.suggest_baseline(
    baseline_dataset=baseline_dataset_path,
    dataset_format=DatasetFormat.csv(header=True),
    output_s3_uri=baseline_output_uri,
    wait=True,
)

これを導入したおかげで、「異常データが来た→パイプラインが止まる→原因不明」という悪循環が消えました。今は異常検知で事前に警告が出るので、データソースチーム側で対応できるようになりました。

2. 条件付きステップのタイミングがシビア

モデルのAUCが0.85を超えたら自動デプロイする、という条件を組み込んだんですけど、ここでハマりました。

問題は、evaluation_stepの出力をConditionStepで読み込む際に、JSON形式をちゃんと指定しないと値が取得できないってことなんです。正直、これはAWSのドキュメントにも大きく目立つ形で書いてなくて、フォーラムで他の人の質問を読みあさってようやく気づきました。

# これはダメ：PropertyFileが定義されていない
condition_step = ConditionStep(
    name="CheckModelQuality",
    conditions=[ConditionGreaterThan(
        left=evaluation_step.properties.SomeMetric,  # AttributeError!
        right=0.85
    )],
)

# これが正解：PropertyFileで出力ファイルと JSON パスを明示
from sagemaker.workflow.step_outputs import JsonGet
from sagemaker.workflow.properties import PropertyFile

evaluation_step = ProcessingStep(
    name="ModelEvaluation",
    processor=evaluation_processor,
    code="evaluate_model.py",
    outputs=[
        ProcessingOutput(
            output_name="evaluation",
            source="/opt/ml/processing/evaluation"
        )
    ],
)

condition_step = ConditionStep(
    name="CheckModelQuality",
    conditions=[ConditionGreaterThan(
        left=JsonGet(
            step=evaluation_step,
            property_file=PropertyFile(
                name="EvaluationMetrics",
                output_name="evaluation",
                json_path="$.metrics.auc"  # JSON パスを指定
            )
        ),
        right=0.85
    )],
)

これに気づくまで、本番環境で条件判定が常にfalseで、デプロイが走らないという悪夢を3週間経験しました。正直まだ腹立つくらいですけど、PropertyFileを使う際は出力スクリプト側で標準的なJSON形式を出力することが本当に重要なんです。

3. モデルレジストリとの連携がスムーズじゃない

モデルの学習が終わったら、自動的にSageMaker Model Registryに登録して、バージョン管理したいじゃないですか。でも実装してみると、RegisterModelStepがちょっと融通が効かないんですよね。ドキュメント通りにやると、なぜか評価メトリクスが自動登録されないみたいなことが起きます。

from sagemaker.workflow.step_outputs import ProcessingOutput
from sagemaker.model import Model
from sagemaker.workflow.steps import CreateModelStep

# モデル学習後、自動でレジストリに登録
create_model_step = CreateModelStep(
    name="CreateModel",
    model=Model(
        image_uri="246618743249.dkr.ecr.us-east-1.amazonaws.com/sagemaker-xgboost:1.7-1-cpu-py3",
        model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
        role=role,
    ),
    inputs=[],
    outputs=[ProcessingOutput(output_name="model", source="/opt/ml/model")],
)

from sagemaker.workflow.steps import RegisterModel
from sagemaker.model_metrics import ModelMetrics, MetricsSource

register_model_step = RegisterModel(
    name="RegisterModel",
    estimator=model_estimator,
    model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
    content_types=["text/csv"],
    response_types=["text/csv"],
    inference_instances=["ml.m5.xlarge"],
    transform_instances=["ml.m5.xlarge"],
    model_metrics=ModelMetrics(
        model_statistics=MetricsSource(
            s3_uri=evaluation_step.properties.ProcessingOutputConfig.Outputs["evaluation"].S3Output.S3Uri
        )
    ),
)

ここの落とし穴は、評価メトリクスをModelRegistryに自動登録する際に、ファイル形式がちょっと厳密だってことです。CloudWatch Logs に出力されたメトリクスじゃなくて、S3に保存された統計情報ファイルから自動で読み込む必要があります。

個人的には、ここは2026年時点でもドキュメントが不足してるし、AWSフォーラムで質問しても「カスタムスクリプトで対応してください」みたいな返事しか来ません。うちは結局、評価スクリプト出力をJSON に統一して、S3にアップロード→ModelMetricsで読み込む、という形に統一しました。地味ですけど、これが最安定なやり方みたいですね。

月40時間の自動化がもたらした変化

パイプラインがようやく安定し始めて、実装から3ヶ月経った時点で、チーム全体で何が変わったかを振り返ってみます。

データサイエンティストの時間配分はこんなふうに変わりました：

項目導入前導入後
日次データ確認・手動チェック1時間/日15分/日
モデル学習・評価実行5時間/週0.5時間/週
本番デプロイ作業3時間/週0（自動化）
障害対応・トラブル解決2時間/週1時間/週
合計削減時間40時間/月月10時間程度

本番デプロイの安全性： AUCが0.85未満だったら自動的にデプロイがスキップされるので、「悪いモデルを誤って本番に上げた」という事故がゼロになりました。これって実は結構大事で、ユーザー側の信頼度も上がったんですよ。

監視と対応の効率化： CloudWatch LogsとModel Monitorでパイプラインの実行状況が可視化されてるので、「学習が遅い」「データドリフトが発生した」みたいな異常を即座に検知できます。実際、先月データドリフトが検出されて、データソースの変更に気づくことができました。従来だったら、ユーザーからのクレーム報告で初めて問題に気づくパターンだったので、かなり防御的な改善ですね。

運用を回すための実践的なTips

本番環境でのパイプラインスケジューリング

パイプラインを毎日走らせるには、EventBridgeで定期実行を設定するのが最小構成です。Lambda関数で定期実行を制御するのが一般的ですね：

import json
from datetime import datetime
import boto3

sagemaker_client = boto3.client("sagemaker")

def lambda_handler(event, context):
    """毎日午前3時にパイプラインを実行"""
    response = sagemaker_client.start_pipeline_execution(
        PipelineName="MLAutomationPipeline",
        PipelineExecutionDisplayName=f"daily-execution-{datetime.now().strftime('%Y-%m-%d')}",
        PipelineParameters=[
            {"Name": "InputDataPath", "Value": "s3://bucket/data/2026-07-06/"},
        ],
    )
    
    return {
        "statusCode": 200,
        "body": json.dumps({"PipelineExecutionArn": response["PipelineExecutionArn"]})
    }

EventBridgeの設定で cron(0 3 * * ? *) を指定すれば、毎日午前3時に実行されます。

コスト最適化の工夫

SageMaker Pipelinesって、実は思ったより安いんですけど、ProcessingJobが結構な金額になります。うちが実装した削減策を紹介しますね：

  • インスタンスサイズの動的選択： データサイズに応じて ml.m5.xlarge と ml.m5.large を切り分け
  • Processing Job の並列実行： 独立したステップ（データ検証と特徴量エンジニアリング）は同時実行
  • スポットインスタンスの活用： 学習ステップでは Spot Instance を使用（約70%削減）

実装例は以下の通りです：

model_estimator = Estimator(
    image_uri="...",
    role=role,
    instance_count=1,
    instance_type="ml.m5.xlarge",
    output_path=model_output_path,
    use_spot_instances=True,  # Spot インスタンスを使用
    max_wait=3600,  # 1時間でタイムアウト
    max_run=1800,  # 30分で完了想定
)

この設定で、月あたりのコストが約60%削減できました（学習ステップに限定ですけど）。

アラート・通知の設定

パイプラインが失敗したときは即座に対応したいので、SNSで通知を設定するのは本当に必須です：

import boto3

events_client = boto3.client("events")

# パイプライン失敗時の通知ルール
rule_name = "MLPipelineFailureAlert"

events_client.put_rule(
    Name=rule_name,
    EventPattern=json.dumps({
        "source": ["aws.sagemaker"],
        "detail-type": ["SageMaker Pipeline State Change"],
        "detail": {
            "pipelineName": ["MLAutomationPipeline"],
            "pipelineExecutionStatus": ["Failed"]
        }
    }),
    State="ENABLED",
)

# SNS トピックへのターゲット設定
events_client.put_targets(
    Rule=rule_name,
    Targets=[
        {
            "Arn": "arn:aws:sns:us-east-1:123456789012:MLPipelineAlerts",
            "RoleArn": "arn:aws:iam::123456789012:role/EventBridgeRole",
            "Id": "1",
        }
    ],
)

Slack連携もできるので、チャンネルに通知を飛ばしておくと、即座に気づけます。

比較表：SageMaker Pipelines vs 他のMLOpsツール

SageMaker Pipelinesがぶっちゃけ他のツールと比べてどうなのか、実装した経験からざっくり比較してみました：

項目SageMaker PipelinesKubeflowAirflow
セットアップ難度低（AWS SDK）高（Kubernetes必須）中（Python設定）
データ品質監視標準搭載（DataQuality）要カスタム実装要カスタム実装
モデルレジストリ統合統合済み別途Kubeflow MetadataAirflow UI のみ
スケーリング性AWS規模で対応クラスタ制限ありホスト規模に依存
コスト（小規模）$50-100/月$300+（EC2維持費）$100-200/月
ML特化度高（SageMaker特化）中（汎用）低（汎用WF）

正直、AWSエコシステムにどっぷり浸かってるなら、SageMaker Pipelinesが一番楽ですね。ただしKubernetesの知見があったり、オンプレ環境を使いたかったりする場合は、Kubeflowを選ぶ理由もあります。

まとめ

SageMaker Pipelinesを本番運用して6ヶ月、月40時間の手作業が消えたことで、データサイエンスチームの生産性が大きく改善されました。実装から得た主な学びを振り返ると、以下の通りです：

  1. データ品質チェックを甘く見るな — DataQuality機能で異常検知を自動化すれば、パイプラインの安定性が飛躍的に向上します。経験上、ここを手抜きするとパイプラインが頻繁にコケます。

  2. 条件付きステップはPropertyFileで明示的に — JSON パス指定を必ずやらないと、条件判定がコケます。ドキュメント以上に詰まりやすいポイントなので、ここはしっかり理解してから実装しましょう。

  3. Model Registry との連携は統一フォーマット必須 — 評価メトリクスの形式を厳密に管理することで、自動登録がスムーズになります。わりと融通が利かないツールなので、テンプレート化するのがおすすめです。

  4. EventBridge × SNS で即座の通知対応 — パイプライン失敗時のアラートを仕組み化すれば、SLAを守りやすくなります。これがあるとないでは、運用の手間が全然違います。

  5. スポットインスタンスで無駄なコスト削減 — 学習ステップで60%のコスト削減ができたのは、組織としても大きな効果です。小さな工夫の積み重ねが、実は結構な削減になります。

次のステップとしては、モデルのA/Bテスト自動化、さらにはマルチモーダルモデルへの拡張を検討中です。2026年のML基盤は「手動の余地をどこまで削減できるか」が本当の勝負どころなので、このパイプライン基盤がチームのスケーリングの第一歩になると思ってます。

皆さんのチームでもMLの自動化に取り組んでいたら、ぜひどうなってるか聞きたいですね。

U

Untanbaby

ソフトウェアエンジニア｜AWS / クラウドアーキテクチャ / DevOps

10年以上のIT実務経験をもとに、現場で使える技術情報を発信しています。 記事の誤りや改善点があればお問い合わせからお気軽にご連絡ください。

関連記事

生成AI・ML基盤

Bedrock Agentsが無限ループした話——8ヶ月本番運用で踏んだ状態管理の地雷

生成AI・ML基盤

SageMaker HyperPodで1年運用して分かったこと｜スケーリング・コスト・本番障害の現実

生成AI・ML基盤

Amazon Q DeveloperをAWSチームに導入して6ヶ月、正直に振り返る

生成AI・ML基盤

Bedrock Flows を本番で6ヶ月使って見えてきたこと【2026年版】

生成AI・ML基盤

月380万のGPU請求から脱出｜Inferentia2導入で70%コスト削減した6ヶ月の記録

生成AI・ML基盤

Bedrock Agentsで本番が崩壊した話｜マルチステップ設計の地雷と対策

← 記事一覧に戻る