効率化の解除：Amazon SageMaker Pipelinesでの選択的な実行の活用

効率化の解除：SageMaker Pipelinesでの選択的な実行の活用

MLOps（Machine Learning Operations）は、機械学習（ML）モデルの本番化のプロセスを監視する重要な学問です。トレーニングとデプロイを行いたい単一のモデルに焦点を当てることは自然です。しかし現実には、数十、さらには数百のモデルを扱うことがあり、プロセスには複数の複雑なステップが関与する場合もあります。そのため、さまざまな複雑さを持つモデルをスケールで追跡、トレーニング、デプロイ、モニタリングするためのインフラストラクチャを整備することが重要です。ここでMLOpsツールが活躍します。MLOpsツールは、これらのプロセスを繰り返し実行可能で信頼性のあるワークフローに組み込んで、MLに適した形で簡素化するのに役立ちます。

Amazon SageMaker Pipelinesは、Amazon SageMakerの機能の一つであり、MLのエンドツーエンドのワークフローをスケールで自動化するための特化したワークフローオーケストレーションサービスです。データの準備、モデルのトレーニング、チューニング、バリデーションなどのタスクを統合的に管理することで、MLモデルの開発とメンテナンスを簡素化します。SageMaker Pipelinesは、ワークフロー管理を効率化し、実験を加速し、モデルの再トレーニングを容易にすることができます。

本投稿では、SageMaker Pipelinesの新機能であるSelective Executionに焦点を当てます。この新機能により、MLワークフローの特定の部分を選択して実行することができます。これにより、パイプラインのスコープ内のステップの実行に制限をかけることで、時間と計算リソースを大幅に節約することができます。さらに、Selective Executionの利点が明確になるさまざまなユースケースを探求し、その価値をさらに確固たるものにします。

ソリューションの概要

SageMaker Pipelinesは、Selective Executionのリリースにより、開発者のエクスペリエンスを革新し続けています。MLビルダーは、パイプライン内で特定のステップを選択して実行することができるようになりました。これにより、パイプライン全体を再実行する必要がなくなります。この機能により、選択したパイプラインのセクションを再実行しながら、選択したステップに関連するランタイムパラメータを変更することができます。

• 「AIプロジェクトはどのように異なるのか」
• 「Embroid」を紹介します：複数の小さなモデルから埋め込み情報を組み合わせるAIメソッドで、監視なしでLLMの予測を自動的に修正することができます
• 「ONNXフレームワークによるモデルの相互運用性と効率の向上」

選択したステップは、非選択ステップの結果に依存する場合があります。このような場合、これらの非選択ステップの出力は、現在のパイプラインバージョンの参照ランの結果から再利用されます。つまり、参照ランは既に完了している必要があります。デフォルトの参照ランは、現在のパイプラインバージョンの最新のランですが、異なるランも選択して参照として使用することができます。

参照ランの全体的な状態は成功、失敗、または停止である必要があります。Selective Executionが出力を使用しようとするとき、実行中であってはなりません。Selective Executionを使用する場合、パイプラインの一部として連続したステップを形成する限り、任意の数のステップを実行することができます。

次の図は、パイプラインの全体的な動作を示しています。

次の図は、Selective Executionを使用したパイプラインの動作を示しています。

以下のセクションでは、パイプラインの直接非循環グラフ（DAGs）における複雑なワークフローを含む、さまざまなシナリオでSelective Executionの使用方法を示します。

前提条件

Selective Executionを試すためには、SageMaker環境の以下のコンポーネントをセットアップする必要があります：

• SageMaker Python SDK – Python環境にインストールされているSageMaker Python SDKが最新であることを確認してください。以下のコマンドをノートブックまたはターミナルから実行して、SageMaker Python SDKのバージョンを2.162.0以上にアップグレードまたはインストールできます：python3 -m pip install sagemaker>=2.162.0またはpip3 install sagemaker>=2.162.0。
• SageMaker Studioへのアクセス（オプション） – Amazon SageMaker Studioは、パイプラインの実行を視覚化し、既存のパイプラインARNと対話するのに役立ちます。SageMaker Studioにアクセスできない場合や、オンデマンドのノートブックやその他のIDEを使用している場合でも、この投稿に従ってパイプラインARNとPython SDKを使用して対話することができます。

完全なエンドツーエンドのウォークスルーのサンプルコードは、GitHubのリポジトリで利用できます。

セットアップ

sagemaker>=1.162.0 Python SDKでは、sagemaker.workflow.selective_execution_configモジュールの一部として、SelectiveExecutionConfigクラスを導入しました。選択的実行機能は、以前に「成功」、「失敗」、または「停止」にマークされたパイプラインARNに依存しています。次のコードスニペットは、SelectiveExecutionConfigクラスをインポートし、参照パイプラインARNを取得し、パイプラインの実行を管理する関連するパイプラインステップとランタイムパラメータを取得する方法を示しています。

import boto3
from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.selective_execution_config import SelectiveExecutionConfig


sm_client = boto3.client('sagemaker')
# サンプルパイプラインの名前を参照する
pipeline_name = "AbalonePipeline"
# 以前の成功したパイプライン実行arnのフィルタリング
pipeline_executions = [_exec
    for _exec in Pipeline(name=pipeline_name).list_executions()['PipelineExecutionSummaries'] 
    if _exec['PipelineExecutionStatus'] == "Succeeded"
]
# 最後の成功した実行を取得する
latest_pipeline_arn = pipeline_executions[0]['PipelineExecutionArn']
print(latest_pipeline_arn)
>>> arn:aws:sagemaker:us-east-1:123123123123:pipeline/AbalonePipeline/execution/x62pbar3gs6h

# サンプルパイプラインのすべてのステップをリストする
execution_steps = sm_client.list_pipeline_execution_steps(
    PipelineExecutionArn=latest_pipeline_arn
)['PipelineExecutionSteps']
print(execution_steps)
>>> 
[{'StepName': 'Abalone-Preprocess',
  'StartTime': datetime.datetime(2023, 6, 27, 4, 41, 30, 519000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2023, 6, 27, 4, 41, 30, 986000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'AttemptCount': 0,
  'Metadata': {'ProcessingJob': {'Arn': 'arn:aws:sagemaker:us-east-1:123123123123:processing-job/pipelines-fvsmu7m7ki3q-Abalone-Preprocess-d68CecvHLU'}},
  'SelectiveExecutionResult': {'SourcePipelineExecutionArn': 'arn:aws:sagemaker:us-east-1:123123123123:pipeline/AbalonePipeline/execution/ksm2mjwut6oz'}},
 {'StepName': 'Abalone-Train',
  'StartTime': datetime.datetime(2023, 6, 27, 4, 41, 31, 320000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2023, 6, 27, 4, 43, 58, 224000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'AttemptCount': 0,
  'Metadata': {'TrainingJob': {'Arn': 'arn:aws:sagemaker:us-east-1:123123123123:training-job/pipelines-x62pbar3gs6h-Abalone-Train-PKhAc1Q6lx'}}},
 {'StepName': 'Abalone-Evaluate',
  'StartTime': datetime.datetime(2023, 6, 27, 4, 43, 59, 40000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2023, 6, 27, 4, 57, 43, 76000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'AttemptCount': 0,
  'Metadata': {'ProcessingJob': {'Arn': 'arn:aws:sagemaker:us-east-1:123123123123:processing-job/pipelines-x62pbar3gs6h-Abalone-Evaluate-vmkZDKDwhk'}}},
 {'StepName': 'Abalone-MSECheck',
  'StartTime': datetime.datetime(2023, 6, 27, 4, 57, 43, 821000, tzinfo=tzlocal()),
  'EndTime': datetime.datetime(2023, 6, 27, 4, 57, 44, 124000, tzinfo=tzlocal()),
  'StepStatus': 'Succeeded',
  'AttemptCount': 0,
  'Metadata': {'Condition': {'Outcome': 'True'}}}]

# パイプラインの設定可能なすべてのパラメータをリストする
# パラメータは選択的実行中に変更できます
parameters = sm_client.list_pipeline_parameters_for_execution(
    PipelineExecutionArn=latest_pipeline_arn
)['PipelineParameters']
print(parameters)
>>> 
[{'Name': 'XGBNumRounds', 'Value': '120'},
 {'Name': 'XGBSubSample', 'Value': '0.9'},
 {'Name': 'XGBGamma', 'Value': '2'},
 {'Name': 'TrainingInstanceCount', 'Value': '1'},
 {'Name': 'XGBMinChildWeight', 'Value': '4'},
 {'Name': 'XGBETA', 'Value': '0.25'},
 {'Name': 'ApprovalStatus', 'Value': 'PendingManualApproval'},
 {'Name': 'ProcessingInstanceCount', 'Value': '1'},
 {'Name': 'ProcessingInstanceType', 'Value': 'ml.t3.medium'},
 {'Name': 'MseThreshold', 'Value': '6'},
 {'Name': 'ModelPath',
  'Value': 's3://sagemaker-us-east-1-123123123123/Abalone/models/'},
 {'Name': 'XGBMaxDepth', 'Value': '12'},
 {'Name': 'TrainingInstanceType', 'Value': 'ml.c5.xlarge'},
 {'Name': 'InputData',
  'Value': 's3://sagemaker-us-east-1-123123123123/sample-dataset/abalone/abalone.csv'}]

ユースケース

このセクションでは、セレクティブ実行が時間とリソースを節約できる可能性のあるいくつかのシナリオを紹介します。セレクティブ実行の利点を示すために、データ抽出、トレーニング、評価、モデル登録、展開などのステップを含む典型的なパイプラインフローを参照として使用します。

SageMakerパイプラインでは、パイプラインパラメータを使用してパイプライン実行のランタイムパラメータを定義できます。新しい実行がトリガーされると、通常は開始から終了までのパイプライン全体が実行されます。ただし、ステップキャッシングが有効になっている場合、SageMakerパイプラインは現在のパイプラインステップの以前の実行で同じ属性値を持つものを検索しようとします。一致が見つかった場合、SageMakerパイプラインはステップを再計算する代わりに、以前の実行からの出力を使用します。ステップキャッシングが有効になっていても、SageMakerパイプラインはデフォルトでワークフロー全体を最後まで実行します。

セレクティブ実行機能のリリースにより、以前のパイプラインARNを使用してパイプラインワークフロー全体を再実行するか、ステップのサブセットを選択的に実行することができます。これは、ステップキャッシングが有効になっていなくても行うことができます。以下のユースケースでは、セレクティブ実行のさまざまな方法を示しています。

ユースケース1：単一のステップを実行する

データサイエンティストは、MLOpsパイプラインのトレーニング段階に焦点を当て、前処理や展開のステップについては心配したくありません。セレクティブ実行を使用すると、データサイエンティストはトレーニングステップに焦点を当て、モデルを改善するためにトレーニングパラメータやハイパーパラメータをリアルタイムで変更できます。これにより、ユーザーが選択したパイプラインステップの実行にのみ計算リソースが使用されるため、時間とコストを節約できます。以下のコードを参照してください：

# 参照パイプラインARNと実行するステップのサブセットを選択
selective_execution_config = SelectiveExecutionConfig(
    source_pipeline_execution_arn="arn:aws:sagemaker:us-east-1:123123123123:pipeline/AbalonePipeline/execution/9e3ljoql7s0n",
    selected_steps=["Abalone-Train"]
)

# パイプラインサブセットの実行を開始
select_execution = pipeline.start(
    selective_execution_config=selective_execution_config,
    parameters={
        "XGBNumRounds": 120,
        "XGBSubSample": 0.9,
        "XGBGamma": 2,
        "XGBMinChildWeight": 4,
        "XGBETA": 0.25,
        "XGBMaxDepth": 12
    }
)

以下の図は、1つのステップを含むパイプラインの進行状況と完了を示しています。

ユースケース2：複数の連続したパイプラインステップを実行する

前のユースケースを続けて、データサイエンティストは新しいモデルをトレーニングし、それをゴールデンテストデータセットと比較してパフォーマンスを評価したいと考えています。この評価は、ユーザー受け入れテスト（UAT）や本番展開のための厳格なガイドラインを満たしていることを確認するために重要です。ただし、データサイエンティストはパイプラインワークフロー全体を実行したり、モデルを展開したりしたくありません。セレクティブ実行を使用して、トレーニングと評価のステップに焦点を当て、必要な検証結果を得ながら時間とリソースを節約できます：

# 参照パイプラインARNと実行するステップのサブセットを選択
selective_execution_config = SelectiveExecutionConfig(
    source_pipeline_execution_arn="arn:aws:sagemaker:us-east-1:123123123123:pipeline/AbalonePipeline/execution/9e3ljoql7s0n",
    selected_steps=["Abalone-Train", "Abalone-Evaluate"]
)

# パイプラインサブセットの実行を開始
select_execution = pipeline.start(
    selective_execution_config=selective_execution_config,
    parameters={
        "ProcessingInstanceType": "ml.t3.medium",
        "XGBNumRounds": 120,
        "XGBSubSample": 0.9,
        "XGBGamma": 2,
        "XGBMinChildWeight": 4,
        "XGBETA": 0.25,
        "XGBMaxDepth": 12
    }
)

ユースケース3：失敗したパイプラインステップの更新と再実行

選択的実行を使用して、パイプライン内の失敗したステップを再実行するか、失敗したステップからパイプラインの実行を再開することができます。これはトラブルシューティングとデバッグに役立ちます。特定の問題に焦点を当てることができるため、効率的な問題解決とより速い反復時間を実現できます。以下の例は、パイプラインの失敗したステップだけを再実行する方法を示しています。

# 以前に失敗したパイプラインARNを選択します
selective_execution_config = SelectiveExecutionConfig(
    source_pipeline_execution_arn="arn:aws:sagemaker:us-east-1:123123123123:pipeline/AbalonePipeline/execution/fvsmu7m7ki3q",
    selected_steps=["Abalone-Evaluate"]
)

# 失敗したパイプラインのサブセットの実行を開始します
select_execution = pipeline.start(
    selective_execution_config=selective_execution_config
)

または、データサイエンティストはSelectiveExecutionConfigで失敗したステップとそれに続くすべてのステップを指定することで、失敗したステップからワークフローの末尾までパイプラインを再開することもできます。

ユースケース4：パイプラインのカバレッジ

一部のパイプラインでは、特定の条件が失敗した場合にのみ実行されるブランチが他のブランチよりも頻度が低い場合があります。これらのブランチを十分にテストして、失敗が発生した場合に期待どおりに動作することを確認することが重要です。頻度が低いブランチをテストすることで、パイプラインが堅牢であり、エラーハンドリングメカニズムが望ましいワークフローを維持し、信頼性のある結果を生成することができます。

selective_execution_config = SelectiveExecutionConfig(
    source_pipeline_execution_arn="arn:aws:sagemaker:us-east-1:123123123123:pipeline/AbalonePipeline/execution/9e3ljoql7s0n",
    selected_steps=["Abalone-Train", "Abalone-Evaluate", "Abalone-MSECheck", "Abalone-FailNotify"]
)

結論

この記事では、SageMaker Pipelinesの選択的実行機能について説明しました。この機能により、MLワークフローの特定のステップを選択的に実行することができます。これにより、時間と計算リソースの節約が可能となります。GitHubリポジトリには、選択的実行の使用方法を示すサンプルコードが用意されています。さまざまなシナリオで使用者に有利な場合があります。選択的実行についてさらに詳しく学びたい場合は、開発者ガイドとAPIリファレンスガイドを参照してください。

SageMaker Pipelinesワークフロー内の利用可能なステップを詳細に調べるには、Amazon SageMaker Model Building PipelineとSageMaker Workflowsを参照してください。さらに、AWS SageMaker Examples GitHubリポジトリでは、SageMaker Pipelinesを使用した異なるユースケースと実装アプローチの例を見つけることができます。これらのリソースは、SageMaker Pipelinesと選択的実行の全ポテンシャルを最大限に活用し、現在および将来のMLプロジェクトで役立つでしょう。

We will continue to update VoAGI; if you have any questions or suggestions, please contact us!