Amazon SageMaker Model Building Pipelines のステップ

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

ステップタイプ

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

Amazon SageMaker Model Building Pipelines では、以下のステップタイプがサポートされています。

@step デコレータ

@step デコレータを使用して、ローカル機械学習コードからステップを作成できます。コードをテストした後、 @stepデコレータで注釈を付けることで、関数を SageMaker パイプラインステップに変換できます。 SageMaker Pipelines @stepは、 デコレーションされた関数の出力をステップとしてパイプラインに渡すときにパイプラインを作成して実行します。また、1 つ以上の デコレーション関数と従来のDAGパイプラインステップを含む複数ステップ@stepの SageMaker パイプラインを作成することもできます。デ@stepコレータを使用してステップを作成する方法の詳細については、「」を参照してください@step デコレータを使用した L 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.stepsProcessingStep」のドキュメントを参照してください。詳細な例については、「Orchestrate Jobs to Train and Evaluate Models with 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.stepsTuningStep. クラスの get_top_model_s3_uri メソッドを使用して、最もパフォーマンスの高いモデルバージョンの 1 つからモデルアーティファクトを取得します。 SageMaker パイプラインでチューニングステップを使用する方法を示すノートブックについては、sagemaker-pipelines-tuning-step「.ipynb」を参照してください。

以下の例は、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 によるモデル開発の自動化」を参照してください。

以下の例は、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メソッドを使用して、AutoML ジョブから最適な目標メトリクスを持つモデルを取得します。また、モデルアーティファクトにアクセスするにはIAMrole、 を使用する必要があります。

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

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

モデルステップ

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

モデルを作成する

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

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

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

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

from sagemaker.workflow.steps import CreateModelStep

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

RegisterModel ステップ

RegisterModel ステップを使用して、sagemaker.model.Model または sagemaker.pipeline PipelineModelを Amazon 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、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"),
)

条件ステップ

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

条件ステップには以下が必要です。

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

制約事項

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.ConditionStep API reference」を参照してください。サポートされている条件の詳細については、 SageMaker Python SDKドキュメントの「Amazon SageMaker Model Building Pipelines - Conditions」を参照してください。

コールバックステップ

Callback ステップを使用して、Amazon SageMaker Model Building Pipelines によって直接提供されていないプロセスと AWS サービスをワークフローに追加します。Callback ステップを実行すると、次の手順が実行されます。

Callback ステップ要件の詳細については、sagemaker.workflow.callback_step CallbackStepのドキュメントを参照してください。完全なソリューションについては、 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={...}
            )
'

動作を停止する

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

実行中のCallbackステップでパイプラインプロセスStopPipelineExecutionで を呼び出すと、 SageMaker Pipelines は Amazon SQS メッセージをSQSキューに送信します。SQS メッセージの本文には、 に設定されたステータスフィールドが含まれていますStopping。メッセージ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"
}

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

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

Lambda ステップ

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

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 と以下のいずれかを指定する必要があります。

以下の例は、既存の 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 ステップで定義する必要があります。

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

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

タイムアウトと停止動作

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

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

ClarifyCheck ステップ

ClarifyCheck ステップを使用すると、バイアス分析とモデルの説明可能性に関するベースラインドリフトチェックを前のベースラインに対して実行できます。次に、model.register() メソッドを使用してベースラインを生成して登録し、step_args を使用してそのメソッドの出力を モデルステップ に渡すことができます。これらのドリフトチェックのベースラインは、Amazon SageMaker Model Monitor でモデルエンドポイントに使用できます。そのため、ベースラインの提案を個別に行う必要はありません。

ClarifyCheck ステップでは、ドリフトチェック用のベースラインをモデルレジストリからプルすることもできます。ClarifyCheck ステップでは、 SageMaker Clarify 構築済みコンテナを使用します。このコンテナは、制約の提案や特定のベースラインに対する制約の検証など、さまざまなモデルモニタリング機能を提供します。詳細については、「 SageMaker Clarify コンテナの使用を開始する」を参照してください。

ClarifyCheck ステップの設定

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

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

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

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

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

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

ベースラインを操作する

オプションで を指定model_package_group_nameして、既存のベースラインを見つけることができます。次に、ClarifyCheckステップは、モデルパッケージグループ内のDriftCheckBaselines最新の承認済みモデルパッケージの をプルします。

supplied_baseline_constraints パラメータを通じて前のベースラインを指定することもできます。model_package_group_name と supplied_baseline_constraints の両方を指定すると、ClarifyCheck ステップでは supplied_baseline_constraints パラメータで指定されたベースラインが使用されます。

ClarifyCheck ステップ要件の使用の詳細については、Amazon SageMaker SageMaker SDK for Python の「sagemaker.workflow.stepsClarifyCheckStep」を参照してください。 SageMaker Pipelines で ClarifyCheck ステップを使用する方法を示す Amazon SageMaker Studio Classic ノートブックについては、sagemaker-pipeline-model-monitor「-clarify-steps.ipynb」を参照してください。

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 ステップを使用して、パイプライン内のデータ品質またはモデル品質のベースラインの提案とドリフトチェックを以前のベースラインに対して実行します。その後、ベースラインを生成して メソッドに登録し、そのメソッドの出力を モデルステップを使用して に渡すことができますstep_args。 model.register() 〕

Model Monitor では、これらのドリフトチェック用のベースラインをモデルエンドポイントで使用できるため、ベースラインを別途提案する必要がなくなります。QualityCheck ステップでは、ドリフトチェック用のベースラインをモデルレジストリからプルすることもできます。QualityCheck ステップでは、Amazon SageMaker Model Monitor のビルド済みコンテナを活用します。このコンテナには、制約の提案、統計の生成、ベースラインに対する制約の検証など、さまざまなモデルモニタリング機能があります。詳細については、「Amazon SageMaker Model Monitor のビルド済みコンテナ」を参照してください。

QualityCheck ステップの設定

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

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

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

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

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

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

ベースラインを操作する

以前のベースラインは、 supplied_baseline_statisticsおよび supplied_baseline_constraintsパラメータを使用して直接指定できます。また、 を指定model_package_group_nameし、 QualityCheck ステップはモデルパッケージグループ内の最新の承認済みモデルパッケージDriftCheckBaselinesで をプルすることもできます。

以下を指定すると、QualityCheckステップは supplied_baseline_constraintsおよび で指定されたベースラインをQualityCheckステップのチェックタイプsupplied_baseline_statisticsに使用します。

QualityCheck ステップ要件の使用の詳細については、Amazon SageMaker SageMaker SDK for Python の「sagemaker.workflow.stepsQualityCheckStep」を参照してください。 SageMaker Pipelines で QualityCheck ステップを使用する方法を示す Amazon SageMaker Studio Classic ノートブックについては、sagemaker-pipeline-model-monitor「-clarify-steps.ipynb」を参照してください。

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 Model Building Pipelines EMR ステップを使用して、次の操作を行います。

Amazon の詳細についてはEMR、「Amazon の開始方法EMR」を参照してください。

EMR ステップでは、 に Amazon EMRクラスターで使用されるJARファイルの場所と渡す引数EMRStepConfigを含める必要があります。実行中のEMRクラスターで ステップを実行する場合は、Amazon EMRクラスター ID も指定します。また、クラスター設定を渡して、ユーザーが作成、管理、終了するクラスターで 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 「Pipelines EMR Step With Running ClusterEMR」を参照してください。

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

がEMRStep作成する新しいクラスターで新しいジョブを起動するには、クラスター設定をディクショナリとして指定します。ディクショナリは、RunJobFlowリクエストと同じ構造である必要があります。ただし、以下のフィールドはクラスター設定に含めないでください。

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

次の例では、クラスター設定を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 「Pipelines EMR Step With Cluster Lifecycle Management」を参照してください。

ノートブックジョブステップ

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

には、少なくとも入力ノートブック、イメージURI、カーネル名NotebookJobStepが必要です。ノートブックジョブのステップ要件と、ステップをカスタマイズするために設定できるその他のパラメータの詳細については、「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、ノートブックジョブを実行する特定のイメージのみを指定できます。詳細については、「Python SDKノートブックジョブの画像制約 SageMaker 」を参照してください。

失敗ステップ

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

FailStep の使用に関する制限

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

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 Pipelines はこれらのデータ依存関係を使用して、パイプライン定義DAGから を構築します。これらのプロパティはプレースホルダー値として参照でき、実行時に解決されます。

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

データ依存関係の作成中に各ステップタイプで参照可能なプロパティを確認するには、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 はステップ間のデータ接続を提供します。または、パイプラインを直接使用せずに、1 SageMaker つのステップで前のステップのデータにアクセスすることもできます。この場合、別のステップの実行が完了するまでステップを開始しないように SageMaker パイプラインに指示するカスタム依存関係を作成できます。カスタム依存関係を作成するには、ステップの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深層学習コンテナイメージのいずれかを使用できます。

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

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