3.3 自動オーケストレーションツールを使用して、CI/CD パイプラインを設定する

タスクステートメント

自動オーケストレーションツールを使用して、継続的インテグレーションおよび継続的デリバリー (CI/CD) パイプラインを設定する

対象知識

AWS CodePipeline、AWS CodeBuild、AWS CodeDeploy の機能とクォータ

AWS CodePipeline

• 機能:
  • ワークフロー全体のオーケストレーション
  • ステージとアクションの定義
  • 承認ステップ
• クォータ:
  • パイプラインあたり最大10ステージ
  • ステージあたり最大50アクション
  • カスタムアクション最大50個

AWS CodeBuild

• 機能:
  • コードのビルドとテスト
  • Docker イメージのビルド
  • カスタムビルド環境
• クォータ:
  • 同時ビルド数: デフォルト60
  • ビルドタイムアウト: 最大8時間
  • アーティファクトサイズ: 最大5GB

AWS CodeDeploy

• 機能:
  • アプリケーションのデプロイ自動化
  • ブルー/グリーン、カナリア、線形デプロイ
  • ロールバック
• 対応サービス:
  • EC2、ECS、Lambda、オンプレミス

データ取り込みのオートメーションと、オーケストレーションサービスとの統合

Amazon EventBridge

• 機能:
  • イベント駆動型アーキテクチャ
  • スケジュールベース実行
  • AWS サービス間の統合

AWS Step Functions

• 機能:
  • ステートマシンでワークフロー定義
  • エラーハンドリング、リトライ
  • 並列実行、条件分岐

Amazon MWAA (Managed Workflows for Apache Airflow)

• 機能:
  • 複雑なワークフローの定義
  • DAG (Directed Acyclic Graph)
  • 依存関係管理

バージョン管理システムと基本的な使用方法

Git の基本

• リポジトリ: コードの保存場所
• ブランチ: 並行開発
• コミット: 変更の記録
• マージ: ブランチの統合
• タグ: バージョンマーキング

AWS CodeCommit

• AWSマネージドGitリポジトリ
• IAM統合
• プルリクエスト、コードレビュー

GitHub / GitLab

• 広く使用されているGitホスティング
• CI/CD機能統合
• CodePipelineと統合可能

CI/CD の原則と ML ワークフローへの適合性

継続的インテグレーション (CI)

• 原則:
  • 頻繁なコード統合
  • 自動テスト
  • 早期問題検出
• ML での適用:
  • データバリデーション
  • モデルテスト
  • パフォーマンステスト

継続的デリバリー (CD)

• 原則:
  • 自動デプロイメント
  • 本番環境への迅速なリリース
  • ロールバック戦略
• ML での適用:
  • モデルの自動デプロイ
  • A/B テスト
  • モデルバージョニング

MLOps の特有の考慮事項

• データドリフト: データ分布の変化
• モデルドリフト: モデル性能の劣化
• 再トレーニング: 定期的なモデル更新
• モデルガバナンス: 承認プロセス

デプロイ戦略とロールバックアクション

ブルー/グリーン デプロイ

• 特徴:
  • 2つの同一環境（ブルー: 現行、グリーン: 新）
  • トラフィックを一度に切り替え
  • 即座にロールバック可能

Canary デプロイ

• 特徴:
  • トラフィックの一部のみ新バージョンへ
  • 段階的に増加（5% → 50% → 100%）
  • リスク軽減

線形デプロイ

• 特徴:
  • 一定割合ずつトラフィックをシフト
  • 例: 10%ずつ10分間隔

コードリポジトリとパイプラインの連携方法

• Webhook: コミット時にパイプライン起動
• ポーリング: 定期的にリポジトリをチェック
• イベントブリッジ: AWSサービス間の統合

対象スキル

1. CodeBuild、CodeDeploy、CodePipeline を設定およびトラブルシューティングする

CodePipeline の設定

# buildspec.yml (CodeBuild)
version: 0.2

phases:
  pre_build:
    commands:
      - echo Logging in to Amazon ECR...
      - aws ecr get-login-password --region $AWS_DEFAULT_REGION | docker login --username AWS --password-stdin $ECR_REPO
  
  build:
    commands:
      - echo Build started on `date`
      - docker build -t $IMAGE_REPO_NAME:$IMAGE_TAG .
      - docker tag $IMAGE_REPO_NAME:$IMAGE_TAG $ECR_REPO/$IMAGE_REPO_NAME:$IMAGE_TAG
  
  post_build:
    commands:
      - echo Build completed on `date`
      - docker push $ECR_REPO/$IMAGE_REPO_NAME:$IMAGE_TAG
      - echo Writing image definitions file...
      - printf '[{"name":"ml-container","imageUri":"%s"}]' $ECR_REPO/$IMAGE_REPO_NAME:$IMAGE_TAG > imagedefinitions.json

artifacts:
  files:
    - imagedefinitions.json
    - appspec.yaml

CodePipeline 定義 (CloudFormation)

Resources:
  MLPipeline:
    Type: AWS::CodePipeline::Pipeline
    Properties:
      Name: ml-model-pipeline
      RoleArn: !GetAtt CodePipelineRole.Arn
      Stages:
        # ソースステージ
        - Name: Source
          Actions:
            - Name: SourceAction
              ActionTypeId:
                Category: Source
                Owner: AWS
                Provider: CodeCommit
                Version: 1
              Configuration:
                RepositoryName: ml-model-repo
                BranchName: main
              OutputArtifacts:
                - Name: SourceOutput
        
        # ビルドステージ
        - Name: Build
          Actions:
            - Name: BuildAction
              ActionTypeId:
                Category: Build
                Owner: AWS
                Provider: CodeBuild
                Version: 1
              Configuration:
                ProjectName: !Ref CodeBuildProject
              InputArtifacts:
                - Name: SourceOutput
              OutputArtifacts:
                - Name: BuildOutput
        
        # デプロイステージ
        - Name: Deploy
          Actions:
            - Name: DeployAction
              ActionTypeId:
                Category: Deploy
                Owner: AWS
                Provider: CloudFormation
                Version: 1
              Configuration:
                ActionMode: CREATE_UPDATE
                StackName: ml-endpoint-stack
                TemplatePath: BuildOutput::template.yaml
              InputArtifacts:
                - Name: BuildOutput

トラブルシューティング

# CodeBuild ログの確認
aws codebuild batch-get-builds --ids $BUILD_ID

# CodePipeline 実行履歴
aws codepipeline get-pipeline-execution \
  --pipeline-name ml-model-pipeline \
  --pipeline-execution-id $EXECUTION_ID

# CodeDeploy デプロイ状況
aws deploy get-deployment --deployment-id $DEPLOYMENT_ID

2. 継続的デプロイフロー構造を適用してパイプラインを呼び出す

Gitflow

main (本番)
  ├─ develop (開発)
  │   ├─ feature/new-model (機能開発)
  │   └─ feature/data-pipeline (機能開発)
  └─ hotfix/bug-fix (緊急修正)

GitHub Flow (シンプル)

main
  ├─ feature/model-v2
  └─ feature/new-feature

パイプライン起動の設定

# GitHub Actions との統合例
name: ML Pipeline
on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  trigger-pipeline:
    runs-on: ubuntu-latest
    steps:
      - name: Trigger CodePipeline
        run: |
          aws codepipeline start-pipeline-execution \
            --name ml-model-pipeline

3. AWS のサービスを使用してオーケストレーションを自動化する

SageMaker Pipelines

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TrainingStep, CreateModelStep
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.functions import JsonGet

# データ処理ステップ
processing_step = ProcessingStep(
    name="PreprocessData",
    processor=sklearn_processor,
    inputs=[
        ProcessingInput(
            source=input_data,
            destination="/opt/ml/processing/input"
        )
    ],
    outputs=[
        ProcessingOutput(
            output_name="train",
            source="/opt/ml/processing/train"
        ),
        ProcessingOutput(
            output_name="validation",
            source="/opt/ml/processing/validation"
        )
    ],
    code="preprocessing.py"
)

# トレーニングステップ
training_step = TrainingStep(
    name="TrainModel",
    estimator=xgb_estimator,
    inputs={
        "train": TrainingInput(
            s3_data=processing_step.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri
        ),
        "validation": TrainingInput(
            s3_data=processing_step.properties.ProcessingOutputConfig.Outputs["validation"].S3Output.S3Uri
        )
    }
)

# モデル評価ステップ
evaluation_step = ProcessingStep(
    name="EvaluateModel",
    processor=sklearn_processor,
    inputs=[
        ProcessingInput(
            source=training_step.properties.ModelArtifacts.S3ModelArtifacts,
            destination="/opt/ml/processing/model"
        )
    ],
    outputs=[
        ProcessingOutput(
            output_name="evaluation",
            source="/opt/ml/processing/evaluation"
        )
    ],
    code="evaluation.py"
)

# 条件付きデプロイ
condition_step = ConditionStep(
    name="CheckAccuracy",
    conditions=[
        ConditionGreaterThanOrEqualTo(
            left=JsonGet(
                step_name=evaluation_step.name,
                property_file="evaluation",
                json_path="metrics.accuracy"
            ),
            right=0.8
        )
    ],
    if_steps=[create_model_step, deploy_step],
    else_steps=[]
)

# パイプラインの作成
pipeline = Pipeline(
    name="ml-training-pipeline",
    steps=[
        processing_step,
        training_step,
        evaluation_step,
        condition_step
    ]
)

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

Step Functions での ML ワークフロー

{
  "Comment": "ML Training Workflow",
  "StartAt": "PrepareData",
  "States": {
    "PrepareData": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sagemaker:createProcessingJob.sync",
      "Parameters": {
        "ProcessingJobName.$": "$.processingJobName",
        "ProcessingInputs": [...],
        "ProcessingOutputConfig": {...}
      },
      "Next": "TrainModel"
    },
    "TrainModel": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sagemaker:createTrainingJob.sync",
      "Parameters": {
        "TrainingJobName.$": "$.trainingJobName",
        "AlgorithmSpecification": {...},
        "InputDataConfig": [...]
      },
      "Next": "EvaluateModel"
    },
    "EvaluateModel": {
      "Type": "Task",
      "Resource": "arn:aws:lambda:us-east-1:123456789012:function:EvaluateModel",
      "Next": "CheckAccuracy"
    },
    "CheckAccuracy": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.accuracy",
          "NumericGreaterThanEquals": 0.8,
          "Next": "DeployModel"
        }
      ],
      "Default": "TrainingFailed"
    },
    "DeployModel": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sagemaker:createEndpoint",
      "End": true
    },
    "TrainingFailed": {
      "Type": "Fail",
      "Error": "AccuracyTooLow",
      "Cause": "Model accuracy below threshold"
    }
  }
}

4. トレーニングジョブと推論ジョブを設定する

EventBridge でスケジュール実行

import boto3

events = boto3.client('events')

# スケジュールルールの作成（毎日午前2時）
events.put_rule(
    Name='daily-model-training',
    ScheduleExpression='cron(0 2 * * ? *)',
    State='ENABLED'
)

# ターゲットの設定（SageMaker トレーニングジョブ）
events.put_targets(
    Rule='daily-model-training',
    Targets=[
        {
            'Id': '1',
            'Arn': 'arn:aws:sagemaker:us-east-1:123456789012:pipeline/ml-training-pipeline',
            'RoleArn': role_arn,
            'SageMakerPipelineParameters': {
                'PipelineParameterList': [
                    {
                        'Name': 'InputData',
                        'Value': f's3://{bucket}/data/latest/'
                    }
                ]
            }
        }
    ]
)

Lambda でパイプライン起動

import boto3
import json

sagemaker = boto3.client('sagemaker')

def lambda_handler(event, context):
    # S3イベントからトリガー
    bucket = event['Records'][0]['s3']['bucket']['name']
    key = event['Records'][0]['s3']['object']['key']
    
    # パイプライン起動
    response = sagemaker.start_pipeline_execution(
        PipelineName='ml-training-pipeline',
        PipelineParameters=[
            {
                'Name': 'InputData',
                'Value': f's3://{bucket}/{key}'
            }
        ]
    )
    
    return {
        'statusCode': 200,
        'body': json.dumps(f'Pipeline started: {response["PipelineExecutionArn"]}')
    }

5. CI/CD パイプラインで自動テストを作成する

ユニットテスト

# test_preprocessing.py
import unittest
from preprocessing import clean_data, encode_features

class TestPreprocessing(unittest.TestCase):
    def test_clean_data(self):
        raw_data = {'col1': [1, 2, None, 4]}
        cleaned = clean_data(raw_data)
        self.assertIsNotNone(cleaned['col1'][2])
    
    def test_encode_features(self):
        data = {'category': ['A', 'B', 'A']}
        encoded = encode_features(data)
        self.assertEqual(len(encoded.columns), 2)

if __name__ == '__main__':
    unittest.main()

統合テスト

# test_integration.py
def test_model_endpoint():
    import boto3
    
    runtime = boto3.client('sagemaker-runtime')
    
    # テストデータで推論
    response = runtime.invoke_endpoint(
        EndpointName='test-endpoint',
        ContentType='application/json',
        Body='{"features": [1, 2, 3, 4]}'
    )
    
    result = json.loads(response['Body'].read())
    assert 'predictions' in result
    assert len(result['predictions']) > 0

エンドツーエンドテスト

# test_e2e.py
def test_full_pipeline():
    import sagemaker
    
    # パイプライン実行
    pipeline = sagemaker.workflow.pipeline.Pipeline(name='ml-pipeline')
    execution = pipeline.start()
    
    # 完了を待機
    execution.wait()
    
    # 結果を検証
    assert execution.describe()['PipelineExecutionStatus'] == 'Succeeded'
    
    # モデルメトリクスを確認
    steps = execution.list_steps()
    eval_step = [s for s in steps if s['StepName'] == 'EvaluateModel'][0]
    
    # メトリクスが閾値を超えているか確認
    # ...

buildspec.yml でのテスト実行

version: 0.2

phases:
  install:
    commands:
      - pip install -r requirements.txt
      - pip install pytest pytest-cov
  
  pre_build:
    commands:
      - echo Running unit tests...
      - pytest tests/unit/ -v --cov=src
  
  build:
    commands:
      - echo Running integration tests...
      - pytest tests/integration/ -v
  
  post_build:
    commands:
      - echo Build completed
      - echo Running e2e tests...
      - pytest tests/e2e/ -v

reports:
  test-results:
    files:
      - 'test-results.xml'
    file-format: 'JUNITXML'
  
  coverage-report:
    files:
      - 'coverage.xml'
    file-format: 'COBERTURAXML'

6. モデルを再トレーニングするためのメカニズムを構築および統合する

モデルドリフト検出と自動再トレーニング

import boto3
from datetime import datetime

cloudwatch = boto3.client('cloudwatch')
sagemaker = boto3.client('sagemaker')

def check_model_drift():
    # CloudWatch からモデル精度メトリクスを取得
    response = cloudwatch.get_metric_statistics(
        Namespace='AWS/SageMaker',
        MetricName='ModelAccuracy',
        Dimensions=[
            {'Name': 'EndpointName', 'Value': 'production-endpoint'}
        ],
        StartTime=datetime.now() - timedelta(days=7),
        EndTime=datetime.now(),
        Period=3600,
        Statistics=['Average']
    )
    
    avg_accuracy = sum([d['Average'] for d in response['Datapoints']]) / len(response['Datapoints'])
    
    # 精度が閾値を下回ったら再トレーニング
    if avg_accuracy < 0.75:
        trigger_retraining()

def trigger_retraining():
    # パイプライン起動
    response = sagemaker.start_pipeline_execution(
        PipelineName='ml-retraining-pipeline',
        PipelineParameters=[
            {
                'Name': 'ReasonForRetraining',
                'Value': 'ModelDrift'
            }
        ]
    )
    
    print(f"Retraining triggered: {response['PipelineExecutionArn']}")

EventBridge Rule で定期再トレーニング

{
  "source": ["aws.cloudwatch"],
  "detail-type": ["CloudWatch Alarm State Change"],
  "detail": {
    "alarmName": ["ModelAccuracyAlarm"],
    "state": {
      "value": ["ALARM"]
    }
  }
}

MLOps パイプラインの例

┌─────────────┐     ┌──────────────┐     ┌────────────┐
│   Source    │────>│    Build     │────>│    Test    │
│ (CodeCommit)│     │ (CodeBuild)  │     │  (Pytest)  │
└─────────────┘     └──────────────┘     └────────────┘
                                                │
                                                v
┌─────────────┐     ┌──────────────┐     ┌────────────┐
│   Deploy    │<────│   Approve    │<────│  Register  │
│ (SageMaker) │     │  (Manual)    │     │  (Registry)│
└─────────────┘     └──────────────┘     └────────────┘

重要なポイント

CI/CD のベストプラクティス

1. 小さく頻繁なコミット: 問題の早期発見
2. 自動テスト: ユニット、統合、E2E
3. 環境の一貫性: dev = staging = prod
4. ロールバック戦略: 迅速な復旧
5. モニタリング: パイプライン、モデル性能
6. ドキュメント: README、ランブック

ML パイプラインの特有の考慮事項

• データバージョニング: DVC、S3バージョニング
• モデルバージョニング: SageMaker Model Registry
• 実験追跡: SageMaker Experiments
• 承認プロセス: 本番デプロイ前の人間による承認
• A/B テスト: 新旧モデルの並行運用

関連リンク

• [[3.1_デプロイインフラストラクチャの選択]]
• [[3.2_インフラストラクチャの作成とスクリプト化]]
• [[4.1_モデル推論のモニタリング]]
• [[2.2_モデルのトレーニングと改良]]
• [[99_AWSサービス一覧]]
• [[00_試験概要]]