パイプラインのステップ

SageMaker パイプラインはステップで構成されています。これらのステップでは、プロパティを使用して、パイプラインが実行するアクションとステップ間の関係を定義します。

ステップタイプ

次に、各ステップタイプの要件について説明し、ステップの実装例を紹介します。これらは、必要なリソースと入力を提供しないため、動作可能な実装ではありません。これらのステップの実装に関するチュートリアルについては、「 SageMaker パイプラインを作成して管理する」を参照してください。

Amazon SageMaker Model Building Pipelines では、次のステップタイプをサポートしています。

• Processing

• トレーニング

• チューニング

• AutoML

• Model

• CreateModel

• RegisterModel

• 変換

• 条件

• Callback

• Lambda

• ClarifyCheck

• QualityCheck

• EMR

• 失敗

処理ステップ

処理ステップを使用すると、データ処理のための処理ジョブを作成できます。ジョブの処理の詳細については、「データを処理して、モデルを評価する」を参照してください。

処理ステップには、プロセッサ、処理コードを定義する Python スクリプト、処理の出力、ジョブ引数が必要です。以下の例は、ProcessingStep 定義の作成方法を示しています。

from sagemaker.sklearn.processing import SKLearnProcessor

sklearn_processor = SKLearnProcessor(framework_version='1.0-1',
                                     role=<role>,
                                     instance_type='ml.m5.xlarge',
                                     instance_count=1)

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

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"),
    ProcessingOutput(output_name="test", source="/opt/ml/processing/test")
]

step_process = ProcessingStep(
    name="AbaloneProcess",
    step_args = sklearn_processor.run(inputs=inputs, outputs=outputs,
        code="abalone/preprocessing.py")
)

ランタイムパラメータを渡す

次の例は、 PySpark ランタイムパラメータをプロセッサからに渡す方法を示していますProcessingStep。

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.spark.processing import PySparkProcessor
from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep

pipeline_session = PipelineSession()

pyspark_processor = PySparkProcessor(
    framework_version='2.4',
    role=<role>,
    instance_type='ml.m5.xlarge',
    instance_count=1,
    sagemaker_session=pipeline_session,
)

step_args = pyspark_processor.run(
    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"),
        ProcessingOutput(output_name="test", source="/opt/ml/processing/test")
    ],
    code="preprocess.py",
    arguments=None,
)


step_process = ProcessingStep(
    name="AbaloneProcess",
    step_args=step_args,
)

処理ステップの要件の詳細については、「sagemaker.workflow.steps」を参照してください。 ProcessingStepドキュメンテーション。詳細な例については、Amazon SageMaker Pipelines サンプルノートブックの「ジョブをオーケストレーションしてモデルをトレーニングおよび評価する」の「フィーチャーエンジニアリングの処理ステップの定義」を参照してください。

トレーニングステップ

トレーニングステップを使用すると、モデルをトレーニングするためのトレーニングジョブを作成できます。トレーニングジョブの詳細については、「Amazonでモデルをトレーニングする」を参照してください SageMaker。

トレーニングステップには、推定器と、トレーニングおよび検証データの入力が必要です。以下の例は、TrainingStep 定義の作成方法を示しています。トレーニングステップの要件の詳細については、「sagemaker.workflow.steps」を参照してください。 TrainingStepドキュメンテーション。

from sagemaker.workflow.pipeline_context import PipelineSession

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

from sagemaker.xgboost.estimator import XGBoost

pipeline_session = PipelineSession()

xgb_estimator = XGBoost(..., sagemaker_session=pipeline_session)

step_args = xgb_estimator.fit(
    inputs={
        "train": TrainingInput(
            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                "train"
            ].S3Output.S3Uri,
            content_type="text/csv"
        ),
        "validation": TrainingInput(
            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                "validation"
            ].S3Output.S3Uri,
            content_type="text/csv"
        )
    }
)

step_train = TrainingStep(
    name="TrainAbaloneModel",
    step_args=step_args,
)

チューニングステップ

チューニングステップを使用すると、ハイパーパラメータチューニングジョブを作成できます。このジョブは、ハイパーパラメータの最適化 (HPO) とも呼ばれます。ハイパーパラメータチューニングジョブは、それぞれがモデルバージョンを生成する、複数のトレーニングジョブを実行します。ハイパーパラメータのチューニングについては、「で自動モデルチューニングを実行する SageMaker」を参照してください。

チューニングジョブは、 SageMaker パイプラインの実験に関連付けられ、トレーニングジョブがトライアルとして作成されます。詳細については、「Experiments の統合」を参照してください。

チューニングステップには、HyperparameterTunerトレーニングの入力とトレーニングの入力が必要です。HyperparameterTuner の warm_start_config パラメータを指定すると、以前のチューニングジョブを再トレーニングできます。ハイパーパラメータのチューニングとウォームスタートの詳細については、「ウォームスタートのハイパーパラメータ調整ジョブを実行する」を参照してください。

sagemaker.workflow.steps の get_top_model_s3_uri メソッドを使用します。 TuningStepクラスを使用して、最もパフォーマンスの高いモデルバージョンの 1 つからモデルアーティファクトを取得します。 SageMaker パイプラインでのチューニングステップの使用方法を解説したノートブックについては、「sagemaker-pipelines-tuning-step.ipynb」を参照してください。

重要

チューニングステップは、Amazon SageMaker Python SDK v3.8.0 と Amazon SageMaker Studio v3.8.0 で導入されました。チューニングステップを使用する前には Studio を更新する必要があります。更新しないと、パイプライン DAG が表示されません。Studio の更新方法については、「 SageMaker Studio をシャットダウンしてアップデート」を参照してください。

以下の例は、TuningStep 定義の作成方法を示しています。

from sagemaker.workflow.pipeline_context import PipelineSession

from sagemaker.tuner import HyperparameterTuner
from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TuningStep

tuner = HyperparameterTuner(..., sagemaker_session=PipelineSession())
    
step_tuning = TuningStep(
    name = "HPTuning",
    step_args = tuner.fit(inputs=TrainingInput(s3_data="s3://my-bucket/my-data"))
)

最適なモデルバージョンを取得する

以下の例は、get_top_model_s3_uri メソッドを使用して、チューニングジョブから最適なモデルバージョンを取得する方法を示しています。最大でパフォーマンス上位 50 件のバージョンが利用でき、ランク付けされていますHyperParameterTuningJobObjective。top_k 引数はバージョンのインデックスです。top_k=0 がパフォーマンスが最も高いバージョン、top_k=49 がパフォーマンスが最も低いバージョンになります。

best_model = Model(
    image_uri=image_uri,
    model_data=step_tuning.get_top_model_s3_uri(
        top_k=0,
        s3_bucket=sagemaker_session.default_bucket()
    ),
    ...
)

チューニングステップの要件の詳細については、「sagemaker.workflow.steps」を参照してください。 TuningStepドキュメンテーション。

AutoML ステップ

AutoML API を使用して AutoML ジョブを作成し、モデルを自動的にトレーニングします。AutoML ジョブの詳細については、「Amazon SageMaker Autopilot によるモデル開発の自動化」を参照してください。

注記

現在、AutoML ステップはアンサンブルトレーニングモードのみをサポートしています。

次のコード例は、を使用して定義を作成する方法を示していますAutoMLStep。

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.workflow.steps import AutoMLStep

pipeline_session = PipelineSession()

auto_ml = AutoML(...,
    role="<role>",
    target_attribute_name="my_target_attribute_name",
    mode="ENSEMBLING",
    sagemaker_session=pipeline_session) 

input_training = AutoMLInput(
    inputs="s3://my-bucket/my-training-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="training",
)
input_validation = AutoMLInput(
    inputs="s3://my-bucket/my-validation-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="validation",
)

step_args = auto_ml.fit(
    inputs=[input_training, input_validation]
)

step_automl = AutoMLStep(
    name="AutoMLStep",
    step_args=step_args,
)     

最適なモデルバージョンを取得する

AutoML ステップは、複数のモデル候補を自動的にトレーニングします。次のように、get_best_auto_ml_modelメソッドと IAMrole を使用してモデルアーティファクトにアクセスすることで、AutoML ジョブから最適な客観的メトリックを含むモデルを取得できます。

best_model = step_automl.get_best_auto_ml_model(role=<role>)

詳細については、 SageMaker Python SDK の AutoML ステップを参照してください。

モデルステップ

SageMaker を使用してモデルを作成または登録します。ModelStepModelStep要件の詳細については、sagemaker.workflow.model_step を参照してください。 ModelStepドキュメンテーション。

モデルの作成

ModelStep SageMaker を使用してモデルを作成できます。AModelStep には、モデルアーティファクトと、 SageMaker モデルを作成するために使用する必要があるインスタンスタイプに関する情報が必要です。モデルの詳細については、「Amazon SageMaker でモデルをトレーニングする」を参照してください SageMaker。

以下の例は、ModelStep 定義の作成方法を示しています。

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

step_train = TrainingStep(...)
model = Model(
    image_uri=pytorch_estimator.training_image_uri(),
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=PipelineSession(),
    role=role,
)

step_model_create = ModelStep(
   name="MyModelCreationStep",
   step_args=model.create(instance_type="ml.m5.xlarge"),
)

モデルの登録

ModelStepsagemaker.model.Modelsagemaker.pipeline.PipelineModel SageMaker aを使用してaまたはaをAmazonモデルレジストリに登録できます。PipelineModel は推論パイプラインを表します。これは、推論要求を処理する連続したコンテナで構成されるモデルです。モデルの登録方法の詳細については、「モデルレジストリを使用したモデルの登録とデプロイ」を参照してください。

次の例では、を登録する aModelStep を作成する方法を示していますPipelineModel。

import time

from sagemaker.workflow.pipeline_context import PipelineSession
from sagemaker.sklearn import SKLearnModel
from sagemaker.xgboost import XGBoostModel

pipeline_session = PipelineSession()

code_location = 's3://{0}/{1}/code'.format(bucket_name, prefix)

sklearn_model = SKLearnModel(
   model_data=processing_step.properties.ProcessingOutputConfig.Outputs['model'].S3Output.S3Uri,
   entry_point='inference.py',
   source_dir='sklearn_source_dir/',
   code_location=code_location,
   framework_version='1.0-1',
   role=role,
   sagemaker_session=pipeline_session,
   py_version='py3'
)

xgboost_model = XGBoostModel(
   model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
   entry_point='inference.py',
   source_dir='xgboost_source_dir/',
   code_location=code_location,
   framework_version='0.90-2',
   py_version='py3',
   sagemaker_session=pipeline_session,
   role=role
)

from sagemaker.workflow.model_step import ModelStep
from sagemaker import PipelineModel

pipeline_model = PipelineModel(
   models=[sklearn_model, xgboost_model],
   role=role,sagemaker_session=pipeline_session,
)

register_model_step_args = pipeline_model.register(
    content_types=["application/json"],
   response_types=["application/json"],
   inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
   transform_instances=["ml.m5.xlarge"],
   model_package_group_name='sipgroup',
)

step_model_registration = ModelStep(
   name="AbaloneRegisterModel",
   step_args=register_model_step_args,
)

CreateModel ステップ

重要

SageMaker Python SDK の v2.90.0 以降では、モデルステップを使用してモデルを作成することをお勧めします。 CreateModelStep以前のバージョンの SageMaker Python SDK でも引き続き動作しますが、現在積極的にサポートされていません。

CreateModel SageMaker ステップを使用してモデルを作成します。モデルの詳細については、「Amazon SageMaker でモデルをトレーニングする」を参照してください SageMaker。

モデル作成ステップには、モデルアーティファクトと、 SageMaker モデルの作成に使用する必要のあるインスタンスタイプに関する情報が必要です。次の例は、CreateModel ステップの定義の作成方法を示しています。CreateModelステップ要件の詳細については、sagemaker.workflow.steps を参照してください。 CreateModelStepドキュメンテーション。

from sagemaker.workflow.steps import CreateModelStep

step_create_model = CreateModelStep(
    name="AbaloneCreateModel",
    model=best_model,
    inputs=inputs
)

RegisterModel ステップ

重要

SageMaker Python SDK の v2.90.0 以降では、モデルステップを使用してモデルを登録することをお勧めします。 RegisterModel以前のバージョンの SageMaker Python SDK でも引き続き動作しますが、現在積極的にサポートされていません。

RegisterModelステップを使用して SageMaker.Model.Model または sagemaker.pipeline を登録します。 PipelineModel SageMaker アマゾンのモデルレジストリで。PipelineModel は推論パイプラインを表します。これは、推論要求を処理する連続したコンテナで構成されるモデルです。

モデルの登録方法の詳細については、「モデルレジストリを使用したモデルの登録とデプロイ」を参照してください。RegisterModelステップ要件の詳細については、sagemaker.workflow.step_collection を参照してください。 RegisterModelドキュメンテーション。

次の例は、PipelineModel を登録する RegisterModel ステップの作成方法を示しています。

import time
from sagemaker.sklearn import SKLearnModel
from sagemaker.xgboost import XGBoostModel

code_location = 's3://{0}/{1}/code'.format(bucket_name, prefix)

sklearn_model = SKLearnModel(model_data=processing_step.properties.ProcessingOutputConfig.Outputs['model'].S3Output.S3Uri,
 entry_point='inference.py',
 source_dir='sklearn_source_dir/',
 code_location=code_location,
 framework_version='1.0-1',
 role=role,
 sagemaker_session=sagemaker_session,
 py_version='py3')

xgboost_model = XGBoostModel(model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts,
 entry_point='inference.py',
 source_dir='xgboost_source_dir/',
 code_location=code_location,
 framework_version='0.90-2',
 py_version='py3',
 sagemaker_session=sagemaker_session,
 role=role)

from sagemaker.workflow.step_collections import RegisterModel
from sagemaker import PipelineModel
pipeline_model = PipelineModel(models=[sklearn_model,xgboost_model],role=role,sagemaker_session=sagemaker_session)

step_register = RegisterModel(
 name="AbaloneRegisterModel",
 model=pipeline_model,
 content_types=["application/json"],
 response_types=["application/json"],
 inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
 transform_instances=["ml.m5.xlarge"],
 model_package_group_name='sipgroup',
)

model が指定されていない場合、モデル登録ステップには、以下の例に示すような推定器が必要です。

from sagemaker.workflow.step_collections import RegisterModel

step_register = RegisterModel(
    name="AbaloneRegisterModel",
    estimator=xgb_train,
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    content_types=["text/csv"],
    response_types=["text/csv"],
    inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
    transform_instances=["ml.m5.xlarge"],
    model_package_group_name=model_package_group_name,
    approval_status=model_approval_status,
    model_metrics=model_metrics
)

変換ステップ

変換ステップは、バッチ変換でデータセット全体に対して推論を実行するために使用します。バッチ変換の詳細については、「推論パイプラインでバッチ変換を実行する」を参照してください。

変換ステップには、変換器と、バッチ変換を実行する対象のデータが必要です。次の例は、Transform ステップの定義の作成方法を示しています。Transformステップ要件の詳細については、sagemaker.workflow.steps を参照してください。 TransformStepドキュメンテーション。

from sagemaker.workflow.pipeline_context import PipelineSession

from sagemaker.transformer import Transformer
from sagemaker.inputs import TransformInput
from sagemaker.workflow.steps import TransformStep

transformer = Transformer(..., sagemaker_session=PipelineSession())

step_transform = TransformStep(
    name="AbaloneTransform",
    step_args=transformer.transform(data="s3://my-bucket/my-data"),
)

条件ステップ

条件ステップを使用すると、ステップのプロパティの条件を評価し、パイプラインで次に実行するアクションを評価できます。

条件ステップには、条件のリスト、条件が true と評価された場合に実行するステップのリスト、条件が false と評価された場合に実行するステップのリストが必要です。以下の例は、ConditionStep 定義の作成方法を示しています。

機能制限

• SageMaker Pipelines は、ネストされた条件ステップの使用をサポートしていません。条件ステップを別の条件ステップの入力として渡すことはできません。

• 条件ステップの両方のブランチで同じステップを使用することはできません。両方のブランチに同じステップ機能が必要な場合は、ステップを複製して別の名前を付けます。

from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo
from sagemaker.workflow.condition_step import (
    ConditionStep,
    JsonGet
)

cond_lte = ConditionLessThanOrEqualTo(
    left=JsonGet(
        step_name=step_eval.name,
        property_file=evaluation_report,
        json_path="regression_metrics.mse.value"
    ),
    right=6.0
)

step_cond = ConditionStep(
    name="AbaloneMSECond",
    conditions=[cond_lte],
    if_steps=[step_register, step_create_model, step_transform],
    else_steps=[]
)

ConditionStep要件の詳細については、sagemaker.workflow.condition_step を参照してください。 ConditionStepAPI リファレンス。サポートされている条件の詳細については、 SageMaker Python SDK ドキュメントの「Amazon SageMaker モデル構築パイプライン-条件」を参照してください。

Callback ステップ

Callbackステップは、Amazon SageMaker Model Building Pipelines から直接提供されない追加のプロセスやAWSのサービスをワークフローに組み込むために使用します。Callback ステップを実行すると、次の手順が実行されます。

• SageMaker Pipelines が、ユーザーが指定した Amazon Simple Queue Service (Amazon SQS) キューにメッセージを送信します。メッセージには、 SageMaker Pipelines によって生成されたトークンと、ユーザーが用意した入力パラメータのリストが含まれます。メッセージを送信したら、 SageMaker Pipelines はユーザーからの応答を待ちます。

• ユーザーは Amazon SQS キューからメッセージを取得し、カスタムプロセスを開始します。

• プロセスが終了すると、ユーザーは次のいずれかの API を呼び出し、 SageMaker Pipelines によって生成された以下のトークンを送信します。

  • SendPipelineExecutionStepSuccess、および出力パラメータのリスト

  • SendPipelineExecutionStepFailure、および障害の理由

• API 呼び出しにより、 SageMaker Pipelines はパイプラインプロセスを続行するか、プロセスを失敗させます。

Callbackステップ要件の詳細については、sagemaker.workflow.callback_step を参照してください。 CallbackStepドキュメンテーション。詳細なソリューションについては、「 SageMaker パイプラインを拡張して、コールバックステップを使用してカスタムステップを含める」を参照してください。

重要

Callbackステップは、Amazon SageMaker Python SDK v2.45.0 と Amazon SageMaker Studio v3.6.2 で導入されました。Callback ステップを使用する前に Studio を更新する必要があります。更新しないと、パイプライン DAG が表示されません。Studio の更新方法については、「 SageMaker Studio をシャットダウンしてアップデート」を参照してください。

以下の例は、前の手順の実装方法を示しています。

from sagemaker.workflow.callback_step import CallbackStep

step_callback = CallbackStep(
    name="MyCallbackStep",
    sqs_queue_url="https://sqs.us-east-2.amazonaws.com/012345678901/MyCallbackQueue",
    inputs={...},
    outputs=[...]
)

callback_handler_code = '
    import boto3
    import json

    def handler(event, context):
        sagemaker_client=boto3.client("sagemaker")

        for record in event["Records"]:
            payload=json.loads(record["body"])
            token=payload["token"]

            # Custom processing

            # Call SageMaker to complete the step
            sagemaker_client.send_pipeline_execution_step_success(
                CallbackToken=token,
                OutputParameters={...}
            )
'

注記

CallbackStepの出力パラメータはネストしないでください。たとえば、ネストされた辞書を出力パラメータとして使用すると、辞書は単一の文字列として扱われます (例: {"output1": "{\"nested_output1\":\"my-output\"}"})。 ネストされた値を指定した場合、特定の出力パラメータを参照しようとすると、再試行できないクライアントエラーが発生します。

動作を停止する

Callback ステップの実行中は、パイプラインプロセスが停止しません。

StopPipelineExecutionCallbackステップの実行中のパイプラインプロセスを呼び出すと、 SageMaker Pipelines から指定した SQS キューに対して、追加の Amazon SQS メッセージが送信されます。SQS メッセージの本文には、Stopping に設定された [Status] (ステータス) フィールドが含まれます。SQS メッセージの本文の例は次のとおりです。

{
  "token": "26vcYbeWsZ",
  "pipelineExecutionArn": "arn:aws:sagemaker:us-east-2:012345678901:pipeline/callback-pipeline/execution/7pinimwddh3a",
  "arguments": {
    "number": 5,
    "stringArg": "some-arg",
    "inputData": "s3://sagemaker-us-west-2-012345678901/abalone/abalone-dataset.csv"
  },
  "status": "Stopping"
}

メッセージの受信時に SendPipelineExecutionStepSuccess または SendPipelineExecutionStepFailure を呼び出して必要なアクション (リソースのクリーンアップなど) を実行するためのロジックを、Amazon SQS メッセージコンシューマーに追加してください。

SageMaker Pipelines がこれらの呼び出しのいずれかを受信した場合にのみ、Pipelines はパイプラインプロセスを停止します。

Lambda ステップ

Lambda ステップを使用すると、AWS Lambda 関数を実行できます。既存の Lambda SageMaker 関数を実行できるほか、新しい Lambda 関数を作成して実行することもできます。 SageMaker パイプラインでの Lambda ステップの使用方法を解説したノートブックについては、「sagemaker-pipelines-lambda-step.ipynb」を参照してください。

重要

Lambda ステップは、Amazon SageMaker Python SDK v3.9.1 で導入されました。 SageMaker Lambda ステップを使用する前には Studio を更新する必要があります。更新しないと、パイプライン DAG が表示されません。Studio の更新方法については、「 SageMaker Studio をシャットダウンしてアップデート」を参照してください。

SageMaker には、Lambda 関数を作成、更新、呼び出し、削除するための sagemaker.lambda クラスが用意されています。 Lambdaには次の署名があります。

Lambda(
    function_arn,       # Only required argument to invoke an existing Lambda function

    # The following arguments are required to create a Lambda function:
    function_name,
    execution_role_arn,
    zipped_code_dir,    # Specify either zipped_code_dir and s3_bucket, OR script
    s3_bucket,          # S3 bucket where zipped_code_dir is uploaded
    script,             # Path of Lambda function script
    handler,            # Lambda handler specified as "lambda_script.lambda_handler"
    timeout,            # Maximum time the Lambda function can run before the lambda step fails
    ...
)

sagemaker.workflow.lambda_step。 LambdaSteplambda_funcクラスには型の引数がありますLambda。既存の Lambda 関数は、関数の Amazon リソースネーム (ARN) を function_arn に指定するだけで呼び出すことができます。function_arn の値を指定しない場合は、handler と以下のいずれかを指定する必要があります。

• zipped_code_dir - 圧縮された Lambda 関数のパス

  s3_bucket - zipped_code_dir がアップロードされる Amazon S3 バケット

• script - Lambda 関数スクリプトファイルのパス

以下の例は、既存の Lambda 関数を呼び出す Lambda ステップ定義の作成方法を示しています。

from sagemaker.workflow.lambda_step import LambdaStep
from sagemaker.lambda_helper import Lambda

step_lambda = LambdaStep(
    name="ProcessingLambda",
    lambda_func=Lambda(
        function_arn="arn:aws:lambda:us-west-2:012345678910:function:split-dataset-lambda"
    ),
    inputs={
        s3_bucket = s3_bucket,
        data_file = data_file
    },
    outputs=[
        "train_file", "test_file"
    ]
)

以下の例は、Lambda 関数スクリプトを使用して Lambda 関数を作成して呼び出す Lambda ステップ定義の作成方法を示しています。

from sagemaker.workflow.lambda_step import LambdaStep
from sagemaker.lambda_helper import Lambda

step_lambda = LambdaStep(
    name="ProcessingLambda",
    lambda_func=Lambda(
      function_name="split-dataset-lambda",
      execution_role_arn=execution_role_arn,
      script="lambda_script.py",
      handler="lambda_script.lambda_handler",
      ...
    ),
    inputs={
        s3_bucket = s3_bucket,
        data_file = data_file
    },
    outputs=[
        "train_file", "test_file"
    ]
)

入力と出力

Lambda 関数に入力または出力がある場合、これらも Lambda ステップで定義する必要があります。

注記

入力パラメータと出力パラメータはネストしないでください。たとえば、ネストされた辞書を出力パラメータとして使用すると、辞書は単一の文字列として扱われます (例: {"output1": "{\"nested_output1\":\"my-output\"}"})。 ネストされた値を指定して後で参照しようとすると、再試行できないクライアントエラーが発生します。

Lambda ステップを定義する場合、inputs はキーと値のペアの辞書でなければなりません。inputs 辞書の各値は、プリミティブ型 (文字列、整数、または浮動小数点) であることが必要です。ネストされたオブジェクトはサポートされません。定義しない場合、inputs 値はデフォルトで None になります。

outputs 値は、キーのリストでなければなりません。これらのキーは、Lambda 関数の出力に定義されている辞書を参照します。inputs と同様に、これらのキーはプリミティブ型でなければならず、ネストされたオブジェクトはサポートされません。

タイムアウトと停止動作

Lambda クラスには、Lambda 関数の実行が可能な最大時間を指定する timeout 引数があります。デフォルト値は 120 秒で、最大値は 10 分です。Lambda 関数の実行中にタイムアウトに達すると、Lambda ステップは失敗しますが、Lambda 関数は引き続き実行されます。

Lambda ステップによって呼び出された Lambda 関数を停止できないため、Lambda ステップの実行中はパイプラインプロセスを停止できません。Lambda 関数の実行中にプロセスの停止を試みると、パイプラインは Lambda 関数が終了するか、タイムアウトに達するまで待機し、いずれかが発生すると停止します。Lambda 関数が終了した場合、パイプラインプロセスのステータスは Stopped になります。タイムアウトに達した場合、パイプラインプロセスのステータスは Failed になります。

ClarifyCheck ステップ

ClarifyCheck ステップを使用すると、バイアス分析とモデルの説明可能性に関するベースラインドリフトチェックを前のベースラインに対して実行できます。その後、ベースラインを生成してメソッドに登録し、model.register()モデルステップそのメソッドの出力をを使用してに渡すことができますstep_args。これらのドリフトチェック用のベースラインは、Amazon SageMaker Model Monitor によってモデルエンドポイントで使用できるため、ベースラインを別途提案する必要がなくなります。ClarifyCheck ステップでは、ドリフトチェック用のベースラインをモデルレジストリからプルすることもできます。ClarifyCheckこのステップでは、特定のベースラインに対する制約の提案や制約の検証など、さまざまなモデルモニタリング機能を提供する Amazon SageMaker Clarify の構築済みコンテナを利用します。詳細については、「 SageMaker Clarify コンテナの使用を開始する」を参照してください。

ClarifyCheck ステップの設定

パイプラインで使用するたびに以下のチェックタイプの 1 つのみを実行するように ClarifyCheck ステップを設定できます。

• データバイアスのチェック

• モデルバイアスのチェック

• モデルの説明可能性のチェック

実行するには、clarify_check_config パラメータに次のいずれかのチェックタイプ値を指定します。

• DataBiasCheckConfig

• ModelBiasCheckConfig

• ModelExplainabilityCheckConfig

ClarifyCheckこのステップでは、 SageMaker Clarify の構築済みコンテナを実行し、チェックと処理ジョブのための専用の設定を必要とする処理ジョブが起動します。 ClarifyCheckConfigCheckJobConfigこれらの構成のヘルパー関数は、 SageMaker Clarify 処理ジョブがモデルバイアス、またはモデルの説明可能性をチェックするためのClarify処理ジョブの計算方法に合わせて調整されます。詳細については、「バイアス分析と説明可能性のための SageMaker Clarify 処理ジョブを実行する」を参照してください。

ドリフトチェックのステップ動作を制御する

ClarifyCheck ステップには、その動作を制御するために、以下の 2 つのブールフラグが必要です。

• skip_check: このパラメータでは、前のベースラインに対するドリフトチェックをスキップするかどうかを指定します。False に設定されている場合、設定されたチェックタイプの前のベースラインが使用可能である必要があります。

• register_new_baseline: このパラメータでは、新しく計算されたベースラインにステッププロパティ BaselineUsedForDriftCheckConstraints を通じてアクセスできるかどうかを指定します。False に設定されている場合、設定されたチェックタイプの前のベースラインも使用可能である必要があります。このベースラインには BaselineUsedForDriftCheckConstraints プロパティを通じてアクセスできます。

詳細については、「Amazon SageMaker Model Building Pipelines を使用したベースライン計算 ClarifyCheck 、ドリフト検出、ライフサイクル、 QualityCheck 手順」を参照してください。

ベースラインを操作する

オプションで model_package_group_name を指定すると、既存のベースラインを検索でき、ClarifyCheck ステップでは、モデルパッケージグループ内の最新の承認済みモデルパッケージの DriftCheckBaselines がプルされます。supplied_baseline_constraints パラメータを通じて前のベースラインを指定することもできます。model_package_group_name と supplied_baseline_constraints の両方を指定すると、ClarifyCheck ステップでは supplied_baseline_constraints パラメータで指定されたベースラインが使用されます。

ClarifyCheckステップ要件の使用方法の詳細については、sagemaker.workflow.steps を参照してください。 ClarifyCheckStepPython 用アマゾン SageMaker SageMaker SDK にあります。ClarifyCheck SageMaker パイプラインでステップの使用方法を解説した Amazon SageMaker Studio ノートブックについては、「sagemaker-pipeline-model-monitor-clarify-steps.ipynb」を参照してください。

例 データバイアスチェック用の ClarifyCheck ステップを作成する

from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.clarify_check_step import DataBiasCheckConfig, ClarifyCheckStep
from sagemaker.workflow.execution_variables import ExecutionVariables

check_job_config = CheckJobConfig(
    role=role,
    instance_count=1,
    instance_type="ml.c5.xlarge",
    volume_size_in_gb=120,
    sagemaker_session=sagemaker_session,
)

data_bias_data_config = DataConfig(
    s3_data_input_path=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    s3_output_path=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'databiascheckstep']),
    label=0,
    dataset_type="text/csv",
    s3_analysis_config_output_path=data_bias_analysis_cfg_output_path,
)

data_bias_config = BiasConfig(
    label_values_or_threshold=[15.0], facet_name=[8], facet_values_or_threshold=[[0.5]]  
)

data_bias_check_config = DataBiasCheckConfig(
    data_config=data_bias_data_config,
    data_bias_config=data_bias_config,
)h

data_bias_check_step = ClarifyCheckStep(
    name="DataBiasCheckStep",
    clarify_check_config=data_bias_check_config,
    check_job_config=check_job_config,
    skip_check=False,
    register_new_baseline=False
   supplied_baseline_constraints="s3://sagemaker-us-west-2-111122223333/baseline/analysis.json",
    model_package_group_name="MyModelPackageGroup"
)

QualityCheck ステップ

QualityCheck ステップを使用すると、パイプラインのデータの品質とモデルの品質についてのベースラインの提案とベースラインドリフトチェックを前のベースラインに対して実行できます。その後、ベースラインを生成してメソッドに登録し、model.register()モデルステップそのメソッドの出力をを使用してに渡すことができますstep_args。Model Monitor では、これらのドリフトチェック用のベースラインをモデルエンドポイントで使用できるため、ベースラインを別途提案する必要がなくなります。QualityCheck ステップでは、ドリフトチェック用のベースラインをモデルレジストリからプルすることもできます。QualityCheckこのステップでは、Amazon SageMaker Model Monitor の事前構築済みコンテナを利用します。このコンテナには、特定のベースラインに対する制約の提案、統計の生成、制約の検証など、さまざまなモデルモニタリング機能が含まれています。詳細については、「Amazon SageMaker モデルモニターのビルド済みコンテナ」を参照してください。

QualityCheck ステップの設定

パイプラインで使用するたびに以下のチェックタイプの 1 つのみを実行するように QualityCheck ステップを設定できます。

• データの品質チェック

• モデルの品質チェック

実行するには、quality_check_config パラメータに次のいずれかのチェックタイプ値を指定します。

• DataQualityCheckConfig

• ModelQualityCheckConfig

QualityCheck ステップでは、Model Monitor の構築済みコンテナを実行し、チェックと処理ジョブのための専用の設定を必要とする処理ジョブが起動します。QualityCheckConfig と CheckJobConfig はこれらの設定のヘルパー関数であり、Model Monitor におけるモデルの品質またはデータの品質をモニタリングするためのベースラインの作成方法に合わせて調整されます。Model Monitor のベースラインの提案については、「ベースラインの作成」と「モデル品質のベースラインを作成する」を参照してください。

ドリフトチェックのステップ動作を制御する

QualityCheck ステップには、その動作を制御するために、以下の 2 つのブールフラグが必要です。

• skip_check: このパラメータでは、前のベースラインに対するドリフトチェックをスキップするかどうかを指定します。False に設定されている場合、設定されたチェックタイプの前のベースラインが使用可能である必要があります。

• register_new_baseline: このパラメータでは、新しく計算されたベースラインにステッププロパティ BaselineUsedForDriftCheckConstraints と BaselineUsedForDriftCheckStatistics を通じてアクセスできるかどうかを指定します。False に設定されている場合、設定されたチェックタイプの前のベースラインも使用可能である必要があります。これらには、BaselineUsedForDriftCheckConstraints と BaselineUsedForDriftCheckStatistics のプロパティを通じてアクセスできます。

詳細については、「Amazon SageMaker Model Building Pipelines を使用したベースライン計算 ClarifyCheck 、ドリフト検出、ライフサイクル、 QualityCheck 手順」を参照してください。

ベースラインを操作する

前のベースラインは、supplied_baseline_statistics と supplied_baseline_constraints のパラメータを通じて直接指定できます。単に model_package_group_name を指定すると、QualityCheck ステップでモデルパッケージグループ内の最新の承認済みモデルパッケージの DriftCheckBaselines がプルされます。model_package_group_name、supplied_baseline_constraints、supplied_baseline_statistics を指定すると、QualityCheck ステップでは、実行する QualityCheck ステップのチェックタイプの supplied_baseline_constraints と supplied_baseline_statistics で指定したベースラインが使用されます。

QualityCheckステップ要件の使用方法の詳細については、sagemaker.workflow.steps を参照してください。 QualityCheckStepPython 用アマゾン SageMaker SageMaker SDK にあります。QualityCheck SageMaker パイプラインでステップの使用方法を解説した Amazon SageMaker Studio ノートブックについては、「sagemaker-pipeline-model-monitor-clarify-steps.ipynb」を参照してください。

例 データ品質チェックのための QualityCheck ステップを作成する

from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.quality_check_step import DataQualityCheckConfig, QualityCheckStep
from sagemaker.workflow.execution_variables import ExecutionVariables

check_job_config = CheckJobConfig(
    role=role,
    instance_count=1,
    instance_type="ml.c5.xlarge",
    volume_size_in_gb=120,
    sagemaker_session=sagemaker_session,
)

data_quality_check_config = DataQualityCheckConfig(
    baseline_dataset=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    dataset_format=DatasetFormat.csv(header=False, output_columns_position="START"),
    output_s3_uri=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'dataqualitycheckstep'])
)

data_quality_check_step = QualityCheckStep(
    name="DataQualityCheckStep",
    skip_check=False,
    register_new_baseline=False,
    quality_check_config=data_quality_check_config,
    check_job_config=check_job_config,
    supplied_baseline_statistics="s3://sagemaker-us-west-2-555555555555/baseline/statistics.json",
    supplied_baseline_constraints="s3://sagemaker-us-west-2-555555555555/baseline/constraints.json",
    model_package_group_name="MyModelPackageGroup"
)

Amazon EMR ステップ

Amazon SageMaker Model Building Pipelines の Amazon EMR ステップを使用して、実行中の Amazon EMR クラスターに対する EMR ステップを処理できます。詳細については、「Amazon EMR の開始方法」を参照してください。

Amazon EMR ステップでは、Amazon EMR クラスターで使用する JAR の Amazon S3 の場所、渡す引数、Amazon EMR クラスター ID が EMRStepConfig に必要です。

注記

• Amazon EMR ステップでは、パイプラインに渡すロールに追加のアクセス許可が必要です。パイプラインロールにAWS管理ポリシー: AmazonSageMakerPipelinesIntegrationsをアタッチするか、ロールにそのポリシーのアクセス許可が含まれていることを確認してください。

• Amazon EMR on EKS はサポートされていません。

• EMR ステップは、STARTING、BOOTSTRAPPING、RUNNING、WAITING のいずれかの状態にあるクラスターでのみ実行できます。

• EMR クラスターでは、PENDING で最大 256 の EMR ステップを使用できます。この制限を超えて EMR ステップが送信されると、パイプラインの実行は失敗します。パイプラインのステップの再試行ポリシー の使用を検討できます。

例 EMR クラスターで新しいジョブを起動する Amazon EMR ステップ定義を作成します。

from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_config = EMRStepConfig(
    jar="s3://path/to/jar/MyJar.jar", # required, S3 path to jar
    args=["--verbose", "--force"], # optional list of arguments to pass to the jar
    main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest 
    properties=[ # optional list of Java properties that are set when the step runs
    {
        "key": "mapred.tasktracker.map.tasks.maximum",
        "value": "2"
    },
    {
        "key": "mapreduce.map.sort.spill.percent",
        "value": "0.90"
   },
   {
       "key": "mapreduce.tasktracker.reduce.tasks.maximum",
       "value": "5"
    }
  ]
)

step_emr = EMRStep (
    name="EMRSampleStep", # required
    cluster_id="j-1YONHTCP3YZKC", # required
    step_config=emr_config, # required
    display_name="My EMR Step",
    description="Pipeline step to execute EMR job"
)

失敗ステップ

FailStepを使用して、必要な条件または状態が達成されなかったときに Amazon SageMaker Model Building Pipelines の実行を停止し、そのパイプラインの実行を失敗としてマークします。FailStep を使用すると、カスタムエラーメッセージを入力し、パイプライン実行失敗の原因を示すこともできます。

注記

FailStep および他のパイプラインステップを同時に実行すると、すべての同時ステップが完了するまでパイプラインは終了しません。

FailStep の使用に関する制限

• FailStep は他のステップの DependsOn リストに追加できません。詳細については、「ステップ間のカスタム依存関係」を参照してください。

• 他のステップは FailStep を参照できません。それは、常にパイプラインの実行の最後のステップです。

• FailStep で終わるパイプライン実行を再試行することはできません。

FailStep ErrorMessage は、静的テキスト文字列の形式で作成できます。または、パイプラインパラメータの Join オペレーションや他のステッププロパティを使用して、より具体的なエラーメッセージを作成することもできます。

次のコードスニペット例では、FailStep でパイプラインパラメータと Join オペレーションで設定した ErrorMessage を使用しています。

from sagemaker.workflow.fail_step import FailStep
from sagemaker.workflow.functions import Join
from sagemaker.workflow.parameters import ParameterInteger

mse_threshold_param = ParameterInteger(name="MseThreshold", default_value=5)
step_fail = FailStep(
    name="AbaloneMSEFail",
    error_message=Join(
        on=" ", values=["Execution failed due to MSE >", mse_threshold_param]
    ),
)

ステップのプロパティ

properties 属性は、パイプラインのステップ間にデータの依存関係を追加するために使用されます。これらのデータ依存関係は、 SageMaker パイプラインによって使用され、パイプライン定義から DAG を作成します。これらのプロパティはプレースホルダー値として参照でき、実行時に解決されます。

SageMaker Pipelinesproperties ステップの属性は、Describe SageMaker 対応するジョブタイプの呼び出しによって返されるオブジェクトと一致します。ジョブタイプごとに、Describe 呼び出しは次のレスポンスオブジェクトを返します。

• ProcessingStep – DescribeProcessingJob

• TrainingStep – DescribeTrainingJob

• TransformStep – DescribeTransformJob

データ依存関係の作成時に各ステップタイプでどのプロパティが参照可能かを確認するには、Amazon SageMaker Python SDK の「データ依存性-プロパティリファレンス」を参照してください。

ステップの並列処理

ステップは、他のステップに依存していない場合、パイプライン実行時に直ちに実行されます。ただし、同時に実行するパイプラインステップが多すぎると、使用可能なリソースがすぐに使い果たされる可能性があります。ParallelismConfiguration を使用して、パイプラインの同時実行ステップ数を制御します。

次の例では ParallelismConfiguration を使用して同時実行ステップの上限数を 5 に設定しています。

pipeline.create(
    parallelism_config=ParallelismConfiguration(5),
)

ステップ間のデータ依存関係

DAG の構造を定義するには、ステップ間のデータ関係を指定します。ステップ間にデータ依存関係を作成するには、パイプラインのあるステップのプロパティを別のステップに入力として渡します。入力を受け取るステップは、入力を提供するステップの実行が終了するまで開始されません。

データ依存関係は、 JsonPath 次の形式の表記で表されます。この形式は JSON プロパティファイルを走査します。つまり、ファイル内にネストされた目的のプロパティに到達するのに必要な数だけ <property> インスタンスを追加できます。 JsonPath 表記の詳細については、JsonPath リポジトリを参照してください。

<step_name>.properties.<property>.<property>

処理ステップの ProcessingOutputConfig プロパティを使用して Amazon S3 バケットを指定する方法を以下に示します。

step_process.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri

データ依存関係を作成するには、以下のようにバケットをトレーニングステップに渡します。

from sagemaker.workflow.pipeline_context import PipelineSession

sklearn_train = SKLearn(..., sagemaker_session=PipelineSession())

step_train = TrainingStep(
    name="CensusTrain",
    step_args=sklearn_train.fit(inputs=TrainingInput(
        s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
            "train_data"].S3Output.S3Uri
    ))
)

データ依存関係の作成時に各ステップタイプでどのプロパティが参照可能かを確認するには、Amazon SageMaker Python SDK の「データ依存性-プロパティリファレンス」を参照してください。

ステップ間のカスタム依存関係

データ依存関係を指定すると、 SageMaker Pipelines によってステップ間のデータが接続されます。また、 SageMaker Pipelines を直接使用せずに、あるステップから前のステップのデータにアクセスすることもできます。この場合、別のステップの実行が終了するまでステップを開始しないように SageMaker Pipelines に指示するカスタム依存関係を作成できます。カスタム依存関係を作成するには、ステップのDependsOn 属性を指定します。

以下の例では、ステップ A とステップ B の両方が実行を終了してから開始されるステップ C を定義します。

{
  'Steps': [
    {'Name':'A', ...},
    {'Name':'B', ...},
    {'Name':'C', 'DependsOn': ['A', 'B']}
  ]
}

SageMaker 依存関係でサイクリック依存性が作成されると、Pipelines で検証例外がスローされます。

以下の例では、処理ステップの実行が終了した後に開始されるトレーニングステップを作成します。

processing_step = ProcessingStep(...)
training_step = TrainingStep(...)

training_step.add_depends_on([processing_step])

以下の例では、2 つの異なる処理ステップの実行が終了するまで開始されないトレーニングステップを作成します。

processing_step_1 = ProcessingStep(...)
processing_step_2 = ProcessingStep(...)

training_step = TrainingStep(...)

training_step.add_depends_on([processing_step_1, processing_step_2])

以下は、カスタム依存関係を作成する別の方法です。

training_step.add_depends_on([processing_step_1])
training_step.add_depends_on([processing_step_2])

以下の例では、ある処理ステップから入力を受け取り、別の処理ステップの実行が終了するまで待機するトレーニングステップを作成します。

processing_step_1 = ProcessingStep(...)
processing_step_2 = ProcessingStep(...)

training_step = TrainingStep(
    ...,
    inputs=TrainingInput(
        s3_data=processing_step_1.properties.ProcessingOutputConfig.Outputs[
            "train_data"
        ].S3Output.S3Uri
    )

training_step.add_depends_on([processing_step_2])

以下の例では、ステップのカスタム依存関係の文字列リストを取得する方法を示しています。

custom_dependencies = training_step.depends_on

ステップでカスタムイメージを使用する

パイプラインでステップを作成する際には、すべての SageMaker深層学習コンテナイメージを使用できます。

パイプラインのステップでは独自のコンテナを使用することもできます。Amazon SageMaker Studio 内でイメージを作成することはできないため、Amazon SageMaker Model Building Pipelines で使用する前に、別の方法を使ってイメージを作成する必要があります。

パイプラインのステップの作成時に独自のコンテナを使用する場合は、推定器定義にイメージ URI を追加します。で独自のコンテナを使用する方法については SageMaker、「で Docker コンテナを使用する」を参照してください SageMaker。