パイプラインのステップ

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

ステップタイプ

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

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

• 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 パイプラインノートブックの例。

トレーニングステップ

トレーニングステップを使用すると、モデルをトレーニングするためのトレーニングジョブを作成できます。トレーニングジョブの詳細については、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 パラメータを指定すると、以前のチューニングジョブを再トレーニングできます。ハイパーパラメータのチューニングとウォームスタートの詳細については、「ウォームスタートのハイパーパラメータ調整ジョブを実行する」を参照してください。

お使いいただくのはget_top_model_s3_uriのを使用できます。セージメーカー、ワークフロー、ステップ。TuningStep最もパフォーマンスの高いモデルバージョンの 1 つからモデルアーティファクトを取得するクラス。でのチューニングステップの使用方法を示すノートブックの場合 SageMaker パイプラインについては、を参照してください。sagemaker-pipelines-tuning-stepipynb。

重要

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

以下の例は、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 ステップ

を使うAutoMLAutoML ジョブで、モデルをトレーニングする際には、約 2 分の時間がかかります。AutoML のジョブの詳細については、「」を参照してください。Amazon によるモデル開発の自動化 SageMaker オートパイロット。

注記

現在、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 ステップは、複数のモデル候補を自動的にトレーニングします。AutoML ジョブから最も客観的な指標をもつモデルを取得するには、get_best_auto_ml_modelメソッドと IAMrole次のようにモデルアーティファクトにアクセスします。

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

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

モデルステップ

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

モデルの作成

次を使用できます。ModelStep作成するには SageMaker モデル。あるModelStepモデルアーティファクトとに関する情報が必要です SageMaker モデルの作成に使用する必要があるインスタンスタイプ。詳細については、詳細について SageMaker モデルについては、を参照してください。Amazon の開始方法 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"),
)

モデルの登録

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

次の例は、ModelStep登録する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 ステップ 1

重要

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

使用するCreateModel作成する SageMaker モデル。詳細については、詳細について SageMaker モデルについては、を参照してください。Amazon の開始方法 SageMaker。

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

from sagemaker.workflow.steps import CreateModelStep

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

RegisterModel ステップ 1

重要

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

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

モデルの登録方法の詳細については、「モデルレジストリを使用したモデルの登録とデプロイ」を参照してください。詳細については、詳細について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ステップ要件については、を参照してくださいセージメーカー. ワークフロー. ステップ。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 モデル構築パイプラインの SageMaker Python SDK の開始方法

Callback ステップ

使用するCallback追加のプロセスを組み込むためのステップとAWSAmazon が直接提供していないサービスをワークフローに組み込む SageMaker モデル構築パイプライン Callback ステップを実行すると、次の手順が実行されます。

• SageMaker パイプラインは、お客様指定の Amazon Simple Queue Service (Amazon SQS) キューにメッセージを送信します。メッセージには以下が含まれます。 SageMaker パイプライン — 生成したトークンと顧客で作成した入力のパラメータの製品がリストで作成した入力のパラメータの製品が削除されます。メッセージを送信すると、 SageMaker パイプラインは顧客からの応答を待ちます。

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

• プロセスが終了すると、顧客は次のAPIのいずれかを呼び出して、 SageMaker パイプライン — 生成トークン:

  • SendPipelineExecutionStepSuccessのサービスクォータの詳細については、

  • SendPipelineExecutionStepFailureUserProfile UserProfile

• API Canvas SageMaker パイプラインプロセスを継続するためには、次の手順を実行します。

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

重要

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

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

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 ステップの実行中は、パイプラインプロセスが停止しません。

を呼び出すとStopPipelineExecution実行中のパイプラインプロセスでCallbackステップ、 SageMaker パイプラインで作成した追加の Amazon SQS のメッセージがパイプラインで作成した 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 パイプラインはこれらの呼び出しのいずれかを受信し、パイプラインプロセスを停止します。

Lambda ステップ

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

重要

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

SageMaker を提供するSagemaker.Lambda_Helper.LambdaLambda 関数を作成、更新、呼び出し、削除するためのクラス。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 ステップ 1

ClarifyCheck ステップを使用すると、バイアス分析とモデルの説明可能性に関するベースラインドリフトチェックを前のベースラインに対して実行できます。その後、とを生成できます。ベースラインを登録とmodel.register()メソッドを実行し、そのメソッドの出力をに渡します。モデルステップを使用しますstep_args。これらのドリフトチェックのベースラインは Amazon でも使用できます。 SageMaker モデルエンドポイント用のモデルモニター。これにより、次のような作業を完了します。ベースライン別途提案 ClarifyCheck ステップでは、ドリフトチェック用のベースラインをモデルレジストリからプルすることもできます。ザ・ClarifyCheckステップはアマゾンを活用します。 SageMaker 制約提案や特定のベースラインに対する制約検証など、モデルモニタリング機能が提供されています。詳細については、次を参照してください。の系統を追跡する SageMaker コンテナをクラリファイ。

の設定 ClarifyCheck ステップ

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

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

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

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

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

• DataBiasCheckConfig

• ModelBiasCheckConfig

• ModelExplainabilityCheckConfig

ザ・ClarifyCheckステップは、を実行する処理ジョブを起動します。 SageMaker事前に構築されたコンテナを明確にし、専用のコンテナが必要ですチェックと処理ジョブの設定。ClarifyCheckConfigそしてCheckJobConfigこれらの構成のヘルパー関数は、以下のように調整されています。 SageMaker モデルバイアス、データバイアス、またはモデルの説明可能性をチェックするための処理ジョブの計算を明確にする。詳細については、「を実行する。 SageMaker バイアス分析と説明可能性を高めることを選択した場合」を参照してください。

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

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

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

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

詳細については、「ベースライン計算、ドリフト検出、ライフサイクル ClarifyCheck そして QualityCheck アマゾンのステップ SageMaker モデル構築パイプライン」を参照してください。

ベースラインを操作する

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

特徴の詳細はClarifyCheck手順の要件については、を参照してください。セージメーカー. ワークフロー. ステップ。ClarifyCheckStepでアマゾン SageMaker SageMakerSDK for Python。アマゾンの場合 SageMaker 使い方が分かるスタジオノートブックClarifyCheckステップイン SageMaker パイプライン、「」を参照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ステップはアマゾンを活用します。 SageMaker モデルモニターでは制約提論や統計生成、ベースラインに対する制約検証など、モデルモニタリング機能が提供されています。詳細については、「アマゾン 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 のプロパティを通じてアクセスできます。

詳細については、「ベースライン計算、ドリフト検出、ライフサイクル ClarifyCheck そして QualityCheck アマゾンのステップ SageMaker モデル構築パイプライン」を参照してください。

ベースラインを操作する

前のベースラインは、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手順の要件については、を参照してください。セージメーカー. ワークフロー. ステップ。QualityCheckStepでアマゾン SageMaker SageMakerSDK for Python。アマゾンの場合 SageMaker 使い方が分かるスタジオノートブックQualityCheckステップイン SageMaker パイプライン、「」を参照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ステップ・トゥ・プロセスアマゾン EMR Step実行中の Amazon EMR クラスターで実行するか、パイプラインに Amazon EMR クラスターの作成と管理を任せます。Amazon EMR の詳細については、Amazon EMR スキーマの例については、。

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

注記

• EMR ステップでは、パイプラインに渡されるロールに追加権限が必要です。添付してくださいAWS管理ポリシー:AmazonSageMakerPipelinesIntegrationsPipeline ロールで使用するか、ロールにそのポリシー内のアクセス権限が含まれていることを確認します。

• EMR ステップは EMR サーバーレスでも、EKS 上の Amazon EMR でもサポートされていません。

• 実行中のクラスターで EMR ステップを処理する場合、以下のいずれかの状態のクラスターのみを使用できます。STARTING、BOOTSTRAPPING、RUNNING、またはWAITING。

• 実行中のクラスターで EMR ステップを処理する場合、1 つのクラスターには最大 256 の EMR ステップを含めることができます。PENDINGEMR クラスター上の状態 この制限を超えて送信された EMR ステップは、パイプラインの実行に失敗します。パイプラインのステップの再試行ポリシー の使用を検討できます。

• クラスタ ID またはクラスタ構成のいずれかを指定できますが、両方は指定できません。

• EMR ステップはアマゾンに依存しています EventBridge EMR ステップまたはクラスターの状態の変化を監視します。実行中のクラスターで Amazon EMR ジョブを処理する場合、EMR ステップはSageMakerPipelineExecutionEMRStepStatusUpdateRuleEMR ステップ状態を監視するルール。EMR ステップで自動的に作成されるクラスターでジョブを処理する場合、ステップはSageMakerPipelineExecutionEMRClusterStatusRuleルールはクラスターの状態の変化を監視します。これらのいずれかが見えたら EventBridge ではを使用できます。AWSアカウントを削除しないでください。削除すると、EMR ステップが完了しない可能性があります。

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

実行中の Amazon EMR クラスターで新しいジョブを開始する場合は、クラスター ID を文字列として渡します。cluster_idの定義EMRStep。次の例は、この手順を示しています。

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"
)

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

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

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

• [Name]

• [Steps]

• [AutoTerminationPolicy]

• [Instances][KeepJobFlowAliveWhenNoSteps]

• [Instances][TerminationProtected]

その他すべてRunJobFlow引数はクラスター設定で使用できます。リクエスト構成の名前RunJobFlow。

次の例では、クラスター構成を EMR ステップ定義に渡します。これにより、ステップは新しい EMR クラスターで新しいジョブを起動するよう求められます。この例の EMR クラスター構成には、プライマリー EMR クラスターノードとコア EMR クラスターノードの仕様が含まれています。Amazon EMR のメトリクスを使用する方法については、を参照してください。ノードタイプ（プライマリ、コア、タスクノード）を理解する。

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 クラスターライフサイクル管理によるパイプライン EMR ステップ。

失敗ステップ

使用するFailStepアマゾンを止めるには SageMaker 希望する条件や状態に達しなかった場合のビルディングパイプラインの実行をモデル化し、パイプラインの実行を失敗とマークします。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 をビルドする際には、これらのプロパティはプレースホルダー値として参照でき、実行時に解決されます。

はpropertiesaのアトリビュート SageMaker パイプラインステップはによって返されたオブジェクトと一致します。Describe対応するものを呼び出します。 SageMaker ジョブの種類 ジョブタイプごとに、Describe 呼び出しは次のレスポンスオブジェクトを返します。

• ProcessingStep – DescribeProcessingJob

• TrainingStep – DescribeTrainingJob

• TransformStep – DescribeTransformJob

データ依存関係の作成中に各ステップタイプでどのプロパティが参照可能かを確認するには、を参照してください。データ依存関係-プロパティリファレンスのアマゾン 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 Python SDK。

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

データ依存関係を指定すると、 SageMaker Pipelines で、ステップ間のデータ接続をする際には、数分の時間がかかります。または、1 つのステップで、直接使用しなくても前のステップのデータにアクセスできます。 SageMaker パイプライン。この場合、次のようなカスタム依存関係を作成できます。 SageMaker パイプラインで作成したステップの製品が別のステップの製品が実行の製品が削除されます。カスタム依存関係を作成するには、ステップのDependsOn 属性を指定します。

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

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

SageMaker Pipeline では、依存関係によって循環依存関係が作成される場合、検証例外が発生します。

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

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 モデル構築パイプライン

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