ステップタイプステップのプロパティステップの並列処理ステップ間のデータ依存関係ステップ間のカスタム依存関係ステップでカスタムイメージを使用する

パイプラインのステップ

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

トピック

ステップタイプ
ステップのプロパティ
ステップの並列処理
ステップ間のデータ依存関係
ステップ間のカスタム依存関係
ステップでカスタムイメージを使用する

ステップタイプ

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

注記

デコレータを使ってローカルの機械学習コードからステップを SageMaker Pipelines ステップに変換して作成することもできます。@step詳細については、「@step デコレーター」を参照してください。

Amazon SageMaker モデルビルディングパイプラインは、以下のステップタイプをサポートしています。

Processing
トレーニング
チューニング
AutoML
Model
CreateModel
RegisterModel
変換
条件
Callback
Lambda
ClarifyCheck
QualityCheck
EMR
ノートブックJob
失敗

@step デコレーター

@stepデコレータを使用して、ローカルの機械学習コードからステップを作成できます。コードをテストしたら、 SageMaker デコレータでアノテーションを付けることで関数をパイプラインステップに変換できます。@step SageMaker @step-decorated 関数の出力をステップとしてパイプラインに渡すと、パイプラインはパイプラインを作成して実行します。また、従来のパイプラインステップだけでなく、@step装飾された関数を 1 つ以上含むマルチステップ DAG パイプラインを作成することもできます。 SageMaker @stepデコレータを使用してステップを作成する方法の詳細については、「」を参照してください。L @step デコレータ付きの ift-and-shift Python コード

処理ステップ

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

処理ステップには、プロセッサ、処理コードを定義する 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、AND のトレーニング入力が必要です。HyperparameterTuner の warm_start_config パラメータを指定すると、以前のチューニングジョブを再トレーニングできます。ハイパーパラメータのチューニングとウォームスタートの詳細については、「ウォームスタートのハイパーパラメータ調整ジョブを実行する」を参照してください。

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

重要

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

以下の例は、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.automl_step 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 メソッドと IAM role を使用してモデルアーティファクトにアクセスすることで、AutoML ジョブから最適な客観的メトリクスを持つモデルを取得できます。


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

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

モデルステップ

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

モデルを作成する

ModelStep SageMaker を使用してモデルを作成できます。A ModelStep には、モデルのアーティファクトと、 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"),
)

モデルの登録

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

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


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 ステップ 1

重要

Python SDK v2.90.0 以降では、モデルステップを使用してモデルを作成することをお勧めします。 SageMaker 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 ステップ 1

重要

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

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

モデルの登録方法の詳細については、「Model Registry を使用したモデルの登録とデプロイ」を参照してください。RegisterModelステップ要件の詳細については、sagemaker.workflow.step_collectionsを参照してください。 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
from sagemaker.workflow.functions import 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 パイプラインは、顧客指定の Amazon Simple Queue Service (Amazon SQS) キューにメッセージを送信します。メッセージには、 SageMaker Pipelines が生成したトークンと、お客様が提供した入力パラメータのリストが含まれています。メッセージの送信後、 SageMaker Pipelines は顧客からの応答を待ちます。
ユーザーは Amazon SQS キューからメッセージを取得し、カスタムプロセスを開始します。
プロセスが終了すると、顧客は次の API のいずれかを呼び出して、Pipelines が生成したトークンを送信します。 SageMaker
- SendPipelineExecutionStepSuccess、出力パラメータのリストとともに
- SendPipelineExecutionStepFailure、失敗の理由とともに
API 呼び出しにより、 SageMaker Pipelines はパイプラインプロセスを続行するか、プロセスを失敗させます。

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

重要

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

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


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 は、これらの呼び出しのいずれかを受信した場合にのみ、パイプラインプロセスを停止します。

Lambda ステップ

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

重要

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

SageMaker には、Lambda 関数を作成、更新、呼び出し、削除するための SageMaker.Lambda_Helper.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。 LambdaStepクラスには型の引数があります。lambda_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 の事前構築済みコンテナを活用します。このコンテナには、制約の提案や特定のベースラインに対する制約の検証など、さまざまなモデルモニタリング機能があります。詳細については、「Clarify コンテナ入門」を参照してください。 SageMaker

ステップの設定 ClarifyCheck

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

データバイアスのチェック
モデルバイアスのチェック
モデルの説明可能性のチェック

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

DataBiasCheckConfig
ModelBiasCheckConfig
ModelExplainabilityCheckConfig

ClarifyCheckこのステップでは、 SageMaker Clarify の事前構築済みコンテナを実行する処理ジョブを起動します。チェックジョブと処理ジョブには専用の設定が必要です。 ClarifyCheckConfigまたCheckJobConfig、モデルバイアス、データバイアス、またはモデルの説明可能性をチェックするための SageMaker 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 用Amazon 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 ステップ 1

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 のプロパティを通じてアクセスできます。

ベースラインを操作する

前のベースラインは、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 用Amazon 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"
)

EMR ステップ

Amazon SageMaker モデルビルディングパイプラインの EMR ステップを使用して、実行中の Amazon EMR クラスターで Amazon EMR ステップを処理することも、パイプラインに Amazon EMR クラスターの作成と管理を任せることもできます。Amazon EMR の詳細については、「Getting started with Amazon EMR」を参照してください。

EMR ステップでは、EMRStepConfig が Amazon EMR クラスターで使用する JAR ファイルの場所と渡す引数を含める必要があります。また、実行中の EMR クラスターでステップを実行する場合は Amazon EMR クラスター ID を指定し、作成、管理、終了するクラスターで EMR ステップを実行する場合はクラスター設定も指定します。以下のセクションには、両方の方法を示す例とサンプルノートブックへのリンクが含まれています。

注記

EMR ステップでは、パイプラインに渡すロールに追加のアクセス許可が必要です。パイプラインロールに AWS マネージドポリシー: AmazonSageMakerPipelinesIntegrations をアタッチするか、ロールにそのポリシーのアクセス許可が含まれていることを確認する必要があります。
EMR ステップは EMR サーバーレスでも、EKS 上の Amazon EMR でもサポートされていません。
実行中のクラスターで EMR ステップを処理する場合は、STARTING、BOOTSTRAPPING、RUNNING、WAITING のいずれかの状態にあるクラスターのみ使用できます。
実行中のクラスターで EMR ステップを処理する場合、EMR クラスター上の PENDING の状態に最大 256 の EMR ステップを含めることができます。この制限を超えて送信された EMR ステップは、パイプラインの実行に失敗します。パイプラインのステップの再試行ポリシーの使用を検討できます。
クラスター ID またはクラスター設定のいずれかを指定できますが、両方を指定することはできません。
EMR ステップでは、Amazon EventBridge が EMR ステップまたはクラスターの状態の変化を監視します。実行中のクラスターで Amazon EMR ジョブを処理する場合、EMR ステップは SageMakerPipelineExecutionEMRStepStatusUpdateRule ルールを使用して EMR ステップの状態をモニタリングします。EMR ステップによって自動的に作成されたクラスターでジョブを処理する場合、ステップは SageMakerPipelineExecutionEMRClusterStatusRule ルールを使用してクラスターの状態の変化をモニタリングします。 EventBridge AWS アカウントにこれらのルールのいずれかが表示されている場合は、削除しないでください。削除すると、EMR ステップが完了しない可能性があります。

実行中の Amazon EMR クラスターで新しいジョブを起動する

実行中の Amazon EMR クラスターで新しいジョブを起動する場合は、クラスター ID を文字列として EMRStep の cluster_id 引数に渡します。次の例で、この手順を説明します。


from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_config = EMRStepConfig(
    jar="jar-location", # required, path to jar file used
    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-1ABCDEFG2HIJK", # include cluster_id to use a running cluster
    step_config=emr_config, # required
    display_name="My EMR Step",
    description="Pipeline step to execute EMR job"
)

完全な例を示すサンプルノートブックについては、「EMR クラスターを実行する SageMaker Pipelines EMR ステップ」を参照してください。

新しい Amazon EMR クラスターで新しいジョブを起動する

EMRStep作成した新しいクラスターで新しいジョブを開始する場合は、クラスター構成をリクエストと同じ構造の辞書として提供してください。RunJobFlowただし、以下のフィールドはクラスター設定に含めないでください。

[Name]
[Steps]
[AutoTerminationPolicy]
[Instances][KeepJobFlowAliveWhenNoSteps]
[Instances][TerminationProtected]

その他の RunJobFlow 引数はすべて、クラスター設定で使用できます。リクエスト構文の詳細については、を参照してくださいRunJobFlow。

次の例では、クラスター設定を EMR ステップ定義に渡します。これにより、ステップは新しい EMR クラスターで新しいジョブを起動するよう求められます。この例の EMR クラスター設定には、プライマリー EMR クラスターノードとコア EMR クラスターノードの仕様が含まれています。Amazon EMR ノードタイプの詳細については、「Understand node types: primary, core, and task nodes」を参照してください。


from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig

emr_step_config = EMRStepConfig(
    jar="jar-location", # required, path to jar file used
    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"
    }
  ]
)

# include your cluster configuration as a dictionary
emr_cluster_config = {
    "Applications": [
        {
            "Name": "Spark", 
        }
    ],
    "Instances":{
        "InstanceGroups":[
            {
                "InstanceRole": "MASTER",
                "InstanceCount": 1,
                "InstanceType": "m5.2xlarge"
            },
            {
                "InstanceRole": "CORE",
                "InstanceCount": 2,
                "InstanceType": "m5.2xlarge"
            }
        ]
    },
    "BootstrapActions":[],
    "ReleaseLabel": "emr-6.6.0",
    "JobFlowRole": "job-flow-role",
    "ServiceRole": "service-role"
}

emr_step = EMRStep(
    name="emr-step",
    cluster_id=None,
    display_name="emr_step",
    description="MyEMRStepDescription",
    step_config=emr_step_config,
    cluster_config=emr_cluster_config
)

完全な例を示すサンプルノートブックについては、「クラスターライフサイクル管理による SageMaker Pipelines EMR ステップ」を参照してください。

ノートブックJob ステップ

SageMaker ノートブックJob NotebookJobStep をパイプラインステップとして非インタラクティブに実行するには、を使用します。 SageMaker ノートブックジョブの詳細については、「」を参照してください。SageMaker ノートブックジョブ

A NotebookJobStep では、少なくともノートブック、イメージ URI、カーネル名を入力する必要があります。ノートブックJob ステップ要件と、ステップをカスタマイズするために設定できるその他のパラメータの詳細については、sagemaker.workflow.steps を参照してください。 NotebookJobStep。

次の例では、NotebookJobStep最小限の引数を使用してを定義しています。


from sagemaker.workflow.notebook_job_step import NotebookJobStep


notebook_job_step = NotebookJobStep(
    input_notebook=input_notebook,
    image_uri=image_uri,
    kernel_name=kernel_name
)

NotebookJobStep SageMaker パイプラインステップはノートブックジョブとして扱われるため、tags引数に特定のタグを含めると、Studio Classic UI のノートブックジョブダッシュボードで実行ステータスを追跡できます。含めるタグの詳細については、を参照してくださいStudio UI ダッシュボードでノートブックジョブを表示します。。

また、 SageMaker Python SDK を使用してノートブックジョブをスケジュールする場合、ノートブックジョブを実行する特定のイメージのみを指定できます。詳細については、「 SageMakerPython SDK ノートブックジョブのイメージ制約」を参照してください。

失敗ステップ

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 Pipelines properties ステップの属性は、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
    ))
)

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

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

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


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

SageMaker 依存関係によって循環依存関係が作成される場合、パイプラインは検証例外を投げます。

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


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 Classic 内からイメージを作成することはできないため、Amazon SageMaker モデルビルディングパイプラインで使用する前に、別の方法でイメージを作成する必要があります。

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

ブラウザで JavaScript が無効になっているか、使用できません。

AWS ドキュメントを使用するには、JavaScript を有効にする必要があります。手順については、使用するブラウザのヘルプページを参照してください。

ドキュメントの表記規則

パイプラインのパラメータ

L @step デコレータ付きの ift-and-shift Python コード

パイプラインのステップ

トピック

ステップタイプ

注記

@step デコレーター

処理ステップ

トレーニングステップ

チューニングステップ

重要

AutoML ステップ

注記

モデルステップ

モデルを作成する

モデルの登録

CreateModel ステップ 1

重要

RegisterModel ステップ 1

重要

変換ステップ

条件ステップ

Callback ステップ

重要

注記

Lambda ステップ

重要

注記

ClarifyCheck ステップ

ステップの設定 ClarifyCheck

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

ベースラインを操作する

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

QualityCheck ステップ 1

QualityCheck ステップの設定

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

ベースラインを操作する

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

EMR ステップ

注記

ノートブックJob ステップ

失敗ステップ

注記

FailStep の使用に関する制限

ステップのプロパティ

ステップの並列処理

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

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

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

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

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

`FailStep` の使用に関する制限