翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
パイプラインのステップ
SageMaker パイプラインはステップで構成されています。これらのステップでは、プロパティを使用して、パイプラインが実行するアクションとステップ間の関係を定義します。
ステップタイプ
次に、各ステップタイプの要件について説明し、ステップの実装例を紹介します。これらは、必要なリソースと入力を提供しないため、動作可能な実装ではありません。これらのステップの実装に関するチュートリアルについては、「 SageMaker パイプラインを作成して管理する」を参照してください。
Amazon SageMaker Model Building Pipelines では、次のステップタイプをサポートしています。
処理ステップ
処理ステップを使用すると、データ処理のための処理ジョブを作成できます。ジョブの処理の詳細については、「データを処理して、モデルを評価する」を参照してください。
処理ステップには、プロセッサ、処理コードを定義する 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。
トレーニングステップには、推定器と、トレーニングおよび検証データの入力が必要です。以下の例は、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 の統合」を参照してください。
チューニングステップには、HyperparameterTunerHyperparameterTuner
の warm_start_config
パラメータを指定すると、以前のチューニングジョブを再トレーニングできます。ハイパーパラメータのチューニングとウォームスタートの詳細については、「ウォームスタートのハイパーパラメータ調整ジョブを実行する」を参照してください。
sagemaker.workflow.steps の get_top_model_s3_uri
チューニングステップは、Amazon SageMaker Python SDK v3.8.0 と Amazon SageMaker Studio v3.8.0 で導入されました。チューニングステップを使用する前には Studio を更新する必要があります。更新しないと、パイプライン DAG が表示されません。Studio の更新方法については、「 SageMaker Studio をシャットダウンしてアップデート」を参照してください。
以下の例は、TuningStep
定義の作成方法を示しています。
from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.tuner import HyperparameterTuner from sagemaker.inputs import TrainingInput from sagemaker.workflow.steps import TuningStep tuner = HyperparameterTuner(..., sagemaker_session=PipelineSession()) step_tuning = TuningStep( name = "HPTuning", step_args = tuner.fit(inputs=TrainingInput(s3_data="s3://
my-bucket/my-data
")) )
最適なモデルバージョンを取得する
以下の例は、get_top_model_s3_uri
メソッドを使用して、チューニングジョブから最適なモデルバージョンを取得する方法を示しています。最大でパフォーマンス上位 50 件のバージョンが利用でき、ランク付けされていますHyperParameterTuningJobObjective。top_k
引数はバージョンのインデックスです。top_k=0
がパフォーマンスが最も高いバージョン、top_k=49
がパフォーマンスが最も低いバージョンになります。
best_model = Model( image_uri=image_uri, model_data=step_tuning.get_top_model_s3_uri( top_k=0, s3_bucket=sagemaker_session.default_bucket() ), ... )
チューニングステップの要件の詳細については、「sagemaker.workflow.steps」を参照してください。 TuningStep
AutoML ステップ
AutoML
現在、AutoML ステップはアンサンブルトレーニングモードのみをサポートしています。
次のコード例は、を使用して定義を作成する方法を示していますAutoMLStep
。
from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.workflow.steps import AutoMLStep pipeline_session = PipelineSession() auto_ml = AutoML(..., role="
<role>
", target_attribute_name="my_target_attribute_name
", mode="ENSEMBLING
", sagemaker_session=pipeline_session) input_training = AutoMLInput( inputs="s3://my-bucket/my-training-data
", target_attribute_name="my_target_attribute_name
", channel_type="training", ) input_validation = AutoMLInput( inputs="s3://my-bucket/my-validation-data
", target_attribute_name="my_target_attribute_name
", channel_type="validation", ) step_args = auto_ml.fit( inputs=[input_training, input_validation] ) step_automl = AutoMLStep( name="AutoMLStep", step_args=step_args, )
最適なモデルバージョンを取得する
AutoML ステップは、複数のモデル候補を自動的にトレーニングします。次のように、get_best_auto_ml_model
メソッドと IAMrole
を使用してモデルアーティファクトにアクセスすることで、AutoML ジョブから最適な客観的メトリックを含むモデルを取得できます。
best_model = step_automl.get_best_auto_ml_model(
role=<role>
)
詳細については、 SageMaker Python SDK の AutoML
モデルステップ
SageMaker を使用してモデルを作成または登録します。ModelStep
ModelStep
要件の詳細については、sagemaker.workflow.model_step を参照してください。 ModelStep
モデルの作成
ModelStep
SageMaker を使用してモデルを作成できます。AModelStep
には、モデルアーティファクトと、 SageMaker モデルを作成するために使用する必要があるインスタンスタイプに関する情報が必要です。モデルの詳細については、「Amazon SageMaker でモデルをトレーニングする」を参照してください SageMaker。
以下の例は、ModelStep
定義の作成方法を示しています。
from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.model import Model from sagemaker.workflow.model_step import ModelStep step_train = TrainingStep(...) model = Model( image_uri=pytorch_estimator.training_image_uri(), model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts, sagemaker_session=PipelineSession(), role=role, ) step_model_create = ModelStep( name="MyModelCreationStep", step_args=model.create(instance_type="ml.m5.xlarge"), )
モデルの登録
ModelStep
sagemaker.model.Model
sagemaker.pipeline.PipelineModel
SageMaker aを使用してaまたはaをAmazonモデルレジストリに登録できます。PipelineModel
は推論パイプラインを表します。これは、推論要求を処理する連続したコンテナで構成されるモデルです。モデルの登録方法の詳細については、「モデルレジストリを使用したモデルの登録とデプロイ」を参照してください。
次の例では、を登録する aModelStep
を作成する方法を示していますPipelineModel
。
import time from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.sklearn import SKLearnModel from sagemaker.xgboost import XGBoostModel pipeline_session = PipelineSession() code_location = 's3://{0}/{1}/code'.format(
bucket_name
, prefix) sklearn_model = SKLearnModel( model_data=processing_step.properties.ProcessingOutputConfig.Outputs['model'].S3Output.S3Uri, entry_point='inference.py', source_dir='sklearn_source_dir/', code_location=code_location, framework_version='1.0-1', role=role, sagemaker_session=pipeline_session, py_version='py3' ) xgboost_model = XGBoostModel( model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts, entry_point='inference.py', source_dir='xgboost_source_dir/', code_location=code_location, framework_version='0.90-2', py_version='py3', sagemaker_session=pipeline_session, role=role ) from sagemaker.workflow.model_step import ModelStep from sagemaker import PipelineModel pipeline_model = PipelineModel( models=[sklearn_model, xgboost_model], role=role,sagemaker_session=pipeline_session, ) register_model_step_args = pipeline_model.register( content_types=["application/json"], response_types=["application/json"], inference_instances=["ml.t2.medium", "ml.m5.xlarge"], transform_instances=["ml.m5.xlarge"], model_package_group_name='sipgroup', ) step_model_registration = ModelStep( name="AbaloneRegisterModel", step_args=register_model_step_args, )
CreateModel ステップ
SageMaker Python SDK の v2.90.0 以降では、モデルステップを使用してモデルを作成することをお勧めします。 CreateModelStep
以前のバージョンの SageMaker Python SDK でも引き続き動作しますが、現在積極的にサポートされていません。
CreateModel
SageMaker ステップを使用してモデルを作成します。モデルの詳細については、「Amazon SageMaker でモデルをトレーニングする」を参照してください SageMaker。
モデル作成ステップには、モデルアーティファクトと、 SageMaker モデルの作成に使用する必要のあるインスタンスタイプに関する情報が必要です。次の例は、CreateModel
ステップの定義の作成方法を示しています。CreateModel
ステップ要件の詳細については、sagemaker.workflow.steps を参照してください。 CreateModelStep
from sagemaker.workflow.steps import CreateModelStep step_create_model = CreateModelStep( name="AbaloneCreateModel", model=best_model, inputs=inputs )
RegisterModel ステップ
SageMaker Python SDK の v2.90.0 以降では、モデルステップを使用してモデルを登録することをお勧めします。 RegisterModel
以前のバージョンの SageMaker Python SDK でも引き続き動作しますが、現在積極的にサポートされていません。
RegisterModel
ステップを使用して SageMaker.Model.ModelPipelineModel
は推論パイプラインを表します。これは、推論要求を処理する連続したコンテナで構成されるモデルです。
モデルの登録方法の詳細については、「モデルレジストリを使用したモデルの登録とデプロイ」を参照してください。RegisterModel
ステップ要件の詳細については、sagemaker.workflow.step_collection を参照してください。 RegisterModel
次の例は、PipelineModel
を登録する RegisterModel
ステップの作成方法を示しています。
import time from sagemaker.sklearn import SKLearnModel from sagemaker.xgboost import XGBoostModel code_location = 's3://{0}/{1}/code'.format(
bucket_name
, prefix) sklearn_model = SKLearnModel(model_data=processing_step.properties.ProcessingOutputConfig.Outputs['model'].S3Output.S3Uri, entry_point='inference.py', source_dir='sklearn_source_dir/
', code_location=code_location, framework_version='1.0-1', role=role, sagemaker_session=sagemaker_session, py_version='py3') xgboost_model = XGBoostModel(model_data=training_step.properties.ModelArtifacts.S3ModelArtifacts, entry_point='inference.py', source_dir='xgboost_source_dir/
', code_location=code_location, framework_version='0.90-2', py_version='py3', sagemaker_session=sagemaker_session, role=role) from sagemaker.workflow.step_collections import RegisterModel from sagemaker import PipelineModel pipeline_model = PipelineModel(models=[sklearn_model,xgboost_model],role=role,sagemaker_session=sagemaker_session) step_register = RegisterModel( name="AbaloneRegisterModel
", model=pipeline_model, content_types=["application/json"], response_types=["application/json"], inference_instances=["ml.t2.medium
", "ml.m5.xlarge
"], transform_instances=["ml.m5.xlarge
"], model_package_group_name='sipgroup
', )
model
が指定されていない場合、モデル登録ステップには、以下の例に示すような推定器が必要です。
from sagemaker.workflow.step_collections import RegisterModel step_register = RegisterModel( name="
AbaloneRegisterModel
", estimator=xgb_train, model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts, content_types=["text/csv"], response_types=["text/csv"], inference_instances=["ml.t2.medium", "ml.m5.xlarge
"], transform_instances=["ml.m5.xlarge
"], model_package_group_name=model_package_group_name, approval_status=model_approval_status, model_metrics=model_metrics )
変換ステップ
変換ステップは、バッチ変換でデータセット全体に対して推論を実行するために使用します。バッチ変換の詳細については、「推論パイプラインでバッチ変換を実行する」を参照してください。
変換ステップには、変換器と、バッチ変換を実行する対象のデータが必要です。次の例は、Transform
ステップの定義の作成方法を示しています。Transform
ステップ要件の詳細については、sagemaker.workflow.steps を参照してください。 TransformStep
from sagemaker.workflow.pipeline_context import PipelineSession from sagemaker.transformer import Transformer from sagemaker.inputs import TransformInput from sagemaker.workflow.steps import TransformStep transformer = Transformer(..., sagemaker_session=PipelineSession()) step_transform = TransformStep( name="AbaloneTransform", step_args=transformer.transform(data="s3://
my-bucket/my-data
"), )
条件ステップ
条件ステップを使用すると、ステップのプロパティの条件を評価し、パイプラインで次に実行するアクションを評価できます。
条件ステップには、条件のリスト、条件が true
と評価された場合に実行するステップのリスト、条件が false
と評価された場合に実行するステップのリストが必要です。以下の例は、ConditionStep
定義の作成方法を示しています。
機能制限
-
SageMaker Pipelines は、ネストされた条件ステップの使用をサポートしていません。条件ステップを別の条件ステップの入力として渡すことはできません。
-
条件ステップの両方のブランチで同じステップを使用することはできません。両方のブランチに同じステップ機能が必要な場合は、ステップを複製して別の名前を付けます。
from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo from sagemaker.workflow.condition_step import ( ConditionStep, JsonGet ) cond_lte = ConditionLessThanOrEqualTo( left=JsonGet( step_name=step_eval.name, property_file=evaluation_report, json_path="regression_metrics.mse.value" ), right=6.0 ) step_cond = ConditionStep( name="AbaloneMSECond", conditions=[cond_lte], if_steps=[step_register, step_create_model, step_transform], else_steps=[] )
ConditionStep
要件の詳細については、sagemaker.workflow.condition_step を参照してください。 ConditionStep
Callback ステップ
Callback
ステップは、Amazon SageMaker Model Building Pipelines から直接提供されない追加のプロセスやAWSのサービスをワークフローに組み込むために使用します。Callback
ステップを実行すると、次の手順が実行されます。
-
SageMaker Pipelines が、ユーザーが指定した Amazon Simple Queue Service (Amazon SQS) キューにメッセージを送信します。メッセージには、 SageMaker Pipelines によって生成されたトークンと、ユーザーが用意した入力パラメータのリストが含まれます。メッセージを送信したら、 SageMaker Pipelines はユーザーからの応答を待ちます。
-
ユーザーは Amazon SQS キューからメッセージを取得し、カスタムプロセスを開始します。
-
プロセスが終了すると、ユーザーは次のいずれかの API を呼び出し、 SageMaker Pipelines によって生成された以下のトークンを送信します。
-
SendPipelineExecutionStepSuccess、および出力パラメータのリスト
-
SendPipelineExecutionStepFailure、および障害の理由
-
-
API 呼び出しにより、 SageMaker Pipelines はパイプラインプロセスを続行するか、プロセスを失敗させます。
Callback
ステップ要件の詳細については、sagemaker.workflow.callback_step を参照してください。 CallbackStep
Callback
ステップは、Amazon SageMaker Python SDK v2.45.0 と Amazon SageMaker Studio v3.6.2 で導入されました。Callback
ステップを使用する前に Studio を更新する必要があります。更新しないと、パイプライン DAG が表示されません。Studio の更新方法については、「 SageMaker Studio をシャットダウンしてアップデート」を参照してください。
以下の例は、前の手順の実装方法を示しています。
from sagemaker.workflow.callback_step import CallbackStep step_callback = CallbackStep( name="MyCallbackStep", sqs_queue_url="https://sqs.us-east-2.amazonaws.com/012345678901/MyCallbackQueue", inputs={...}, outputs=[...] ) callback_handler_code = ' import boto3 import json def handler(event, context): sagemaker_client=boto3.client("sagemaker") for record in event["Records"]: payload=json.loads(record["body"]) token=payload["token"] # Custom processing # Call SageMaker to complete the step sagemaker_client.send_pipeline_execution_step_success( CallbackToken=token, OutputParameters={...} ) '
CallbackStep
の出力パラメータはネストしないでください。たとえば、ネストされた辞書を出力パラメータとして使用すると、辞書は単一の文字列として扱われます (例: {"output1": "{\"nested_output1\":\"my-output\"}"}
)。 ネストされた値を指定した場合、特定の出力パラメータを参照しようとすると、再試行できないクライアントエラーが発生します。
動作を停止する
Callback
ステップの実行中は、パイプラインプロセスが停止しません。
StopPipelineExecutionCallback
ステップの実行中のパイプラインプロセスを呼び出すと、 SageMaker Pipelines から指定した SQS キューに対して、追加の Amazon SQS メッセージが送信されます。SQS メッセージの本文には、Stopping
に設定された [Status] (ステータス) フィールドが含まれます。SQS メッセージの本文の例は次のとおりです。
{ "token": "26vcYbeWsZ", "pipelineExecutionArn": "arn:aws:sagemaker:us-east-2:012345678901:pipeline/callback-pipeline/execution/7pinimwddh3a", "arguments": { "number": 5, "stringArg": "some-arg", "inputData": "s3://sagemaker-us-west-2-012345678901/abalone/abalone-dataset.csv" }, "status": "Stopping" }
メッセージの受信時に SendPipelineExecutionStepSuccess
または SendPipelineExecutionStepFailure
を呼び出して必要なアクション (リソースのクリーンアップなど) を実行するためのロジックを、Amazon SQS メッセージコンシューマーに追加してください。
SageMaker Pipelines がこれらの呼び出しのいずれかを受信した場合にのみ、Pipelines はパイプラインプロセスを停止します。
Lambda ステップ
Lambda ステップを使用すると、AWS Lambda 関数を実行できます。既存の Lambda SageMaker 関数を実行できるほか、新しい Lambda 関数を作成して実行することもできます。 SageMaker パイプラインでの Lambda ステップの使用方法を解説したノートブックについては、「sagemaker-pipelines-lambda-step.ipynb
Lambda ステップは、Amazon SageMaker Python SDK v3.9.1 で導入されました。 SageMaker Lambda ステップを使用する前には Studio を更新する必要があります。更新しないと、パイプライン DAG が表示されません。Studio の更新方法については、「 SageMaker Studio をシャットダウンしてアップデート」を参照してください。
SageMaker には、Lambda 関数を作成、更新、呼び出し、削除するための sagemaker.lambdaLambda
には次の署名があります。
Lambda( function_arn, # Only required argument to invoke an existing Lambda function # The following arguments are required to create a Lambda function: function_name, execution_role_arn, zipped_code_dir, # Specify either zipped_code_dir and s3_bucket, OR script s3_bucket, # S3 bucket where zipped_code_dir is uploaded script, # Path of Lambda function script handler, # Lambda handler specified as "lambda_script.lambda_handler" timeout, # Maximum time the Lambda function can run before the lambda step fails ... )
sagemaker.workflow.lambda_step。 LambdaSteplambda_func
クラスには型の引数がありますLambda
。既存の Lambda 関数は、関数の Amazon リソースネーム (ARN) を function_arn
に指定するだけで呼び出すことができます。function_arn
の値を指定しない場合は、handler
と以下のいずれかを指定する必要があります。
-
zipped_code_dir
- 圧縮された Lambda 関数のパスs3_bucket
-zipped_code_dir
がアップロードされる Amazon S3 バケット -
script
- Lambda 関数スクリプトファイルのパス
以下の例は、既存の Lambda 関数を呼び出す Lambda
ステップ定義の作成方法を示しています。
from sagemaker.workflow.lambda_step import LambdaStep from sagemaker.lambda_helper import Lambda step_lambda = LambdaStep( name="ProcessingLambda", lambda_func=Lambda( function_arn="arn:aws:lambda:us-west-2:012345678910:function:split-dataset-lambda" ), inputs={ s3_bucket = s3_bucket, data_file = data_file }, outputs=[ "train_file", "test_file" ] )
以下の例は、Lambda 関数スクリプトを使用して Lambda 関数を作成して呼び出す Lambda
ステップ定義の作成方法を示しています。
from sagemaker.workflow.lambda_step import LambdaStep from sagemaker.lambda_helper import Lambda step_lambda = LambdaStep( name="ProcessingLambda", lambda_func=Lambda( function_name="split-dataset-lambda", execution_role_arn=execution_role_arn, script="lambda_script.py", handler="lambda_script.lambda_handler", ... ), inputs={ s3_bucket = s3_bucket, data_file = data_file }, outputs=[ "train_file", "test_file" ] )
入力と出力
Lambda
関数に入力または出力がある場合、これらも Lambda
ステップで定義する必要があります。
入力パラメータと出力パラメータはネストしないでください。たとえば、ネストされた辞書を出力パラメータとして使用すると、辞書は単一の文字列として扱われます (例: {"output1": "{\"nested_output1\":\"my-output\"}"}
)。 ネストされた値を指定して後で参照しようとすると、再試行できないクライアントエラーが発生します。
Lambda
ステップを定義する場合、inputs
はキーと値のペアの辞書でなければなりません。inputs
辞書の各値は、プリミティブ型 (文字列、整数、または浮動小数点) であることが必要です。ネストされたオブジェクトはサポートされません。定義しない場合、inputs
値はデフォルトで None
になります。
outputs
値は、キーのリストでなければなりません。これらのキーは、Lambda
関数の出力に定義されている辞書を参照します。inputs
と同様に、これらのキーはプリミティブ型でなければならず、ネストされたオブジェクトはサポートされません。
タイムアウトと停止動作
Lambda
クラスには、Lambda 関数の実行が可能な最大時間を指定する timeout
引数があります。デフォルト値は 120 秒で、最大値は 10 分です。Lambda 関数の実行中にタイムアウトに達すると、Lambda ステップは失敗しますが、Lambda 関数は引き続き実行されます。
Lambda ステップによって呼び出された Lambda 関数を停止できないため、Lambda ステップの実行中はパイプラインプロセスを停止できません。Lambda 関数の実行中にプロセスの停止を試みると、パイプラインは Lambda 関数が終了するか、タイムアウトに達するまで待機し、いずれかが発生すると停止します。Lambda 関数が終了した場合、パイプラインプロセスのステータスは Stopped
になります。タイムアウトに達した場合、パイプラインプロセスのステータスは Failed
になります。
ClarifyCheck ステップ
ClarifyCheck
ステップを使用すると、バイアス分析とモデルの説明可能性に関するベースラインドリフトチェックを前のベースラインに対して実行できます。その後、ベースラインを生成してメソッドに登録し、model.register()
モデルステップそのメソッドの出力をを使用してに渡すことができますstep_args
。これらのドリフトチェック用のベースラインは、Amazon SageMaker Model Monitor によってモデルエンドポイントで使用できるため、ベースラインを別途提案する必要がなくなります。ClarifyCheck
ステップでは、ドリフトチェック用のベースラインをモデルレジストリからプルすることもできます。ClarifyCheck
このステップでは、特定のベースラインに対する制約の提案や制約の検証など、さまざまなモデルモニタリング機能を提供する Amazon SageMaker Clarify の構築済みコンテナを利用します。詳細については、「 SageMaker Clarify コンテナの使用を開始する」を参照してください。
ClarifyCheck ステップの設定
パイプラインで使用するたびに以下のチェックタイプの 1 つのみを実行するように ClarifyCheck
ステップを設定できます。
-
データバイアスのチェック
-
モデルバイアスのチェック
-
モデルの説明可能性のチェック
実行するには、clarify_check_config
パラメータに次のいずれかのチェックタイプ値を指定します。
-
DataBiasCheckConfig
-
ModelBiasCheckConfig
-
ModelExplainabilityCheckConfig
ClarifyCheck
このステップでは、 SageMaker Clarify の構築済みコンテナを実行し、チェックと処理ジョブのための専用の設定を必要とする処理ジョブが起動します。 ClarifyCheckConfig
CheckJobConfig
これらの構成のヘルパー関数は、 SageMaker Clarify 処理ジョブがモデルバイアス、またはモデルの説明可能性をチェックするためのClarify処理ジョブの計算方法に合わせて調整されます。詳細については、「バイアス分析と説明可能性のための SageMaker Clarify 処理ジョブを実行する」を参照してください。
ドリフトチェックのステップ動作を制御する
ClarifyCheck
ステップには、その動作を制御するために、以下の 2 つのブールフラグが必要です。
-
skip_check
: このパラメータでは、前のベースラインに対するドリフトチェックをスキップするかどうかを指定します。False
に設定されている場合、設定されたチェックタイプの前のベースラインが使用可能である必要があります。 -
register_new_baseline
: このパラメータでは、新しく計算されたベースラインにステッププロパティBaselineUsedForDriftCheckConstraints
を通じてアクセスできるかどうかを指定します。False
に設定されている場合、設定されたチェックタイプの前のベースラインも使用可能である必要があります。このベースラインにはBaselineUsedForDriftCheckConstraints
プロパティを通じてアクセスできます。
詳細については、「Amazon SageMaker Model Building Pipelines を使用したベースライン計算 ClarifyCheck 、ドリフト検出、ライフサイクル、 QualityCheck 手順」を参照してください。
ベースラインを操作する
オプションで model_package_group_name
を指定すると、既存のベースラインを検索でき、ClarifyCheck
ステップでは、モデルパッケージグループ内の最新の承認済みモデルパッケージの DriftCheckBaselines
がプルされます。supplied_baseline_constraints
パラメータを通じて前のベースラインを指定することもできます。model_package_group_name
と supplied_baseline_constraints
の両方を指定すると、ClarifyCheck
ステップでは supplied_baseline_constraints
パラメータで指定されたベースラインが使用されます。
ClarifyCheck
ステップ要件の使用方法の詳細については、sagemaker.workflow.steps を参照してください。 ClarifyCheck
SageMaker パイプラインでステップの使用方法を解説した Amazon SageMaker Studio ノートブックについては、「sagemaker-pipeline-model-monitor-clarify-steps.ipynb
例 データバイアスチェック用の ClarifyCheck
ステップを作成する
from sagemaker.workflow.check_job_config import CheckJobConfig from sagemaker.workflow.clarify_check_step import DataBiasCheckConfig, ClarifyCheckStep from sagemaker.workflow.execution_variables import ExecutionVariables check_job_config = CheckJobConfig( role=role, instance_count=1, instance_type="
ml.c5.xlarge
", volume_size_in_gb=120
, sagemaker_session=sagemaker_session, ) data_bias_data_config = DataConfig( s3_data_input_path=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri, s3_output_path=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'databiascheckstep
']), label=0, dataset_type="text/csv
", s3_analysis_config_output_path=data_bias_analysis_cfg_output_path, ) data_bias_config = BiasConfig( label_values_or_threshold=[15.0
], facet_name=[8
], facet_values_or_threshold=[[0.5
]] ) data_bias_check_config = DataBiasCheckConfig( data_config=data_bias_data_config, data_bias_config=data_bias_config, )h data_bias_check_step = ClarifyCheckStep( name="DataBiasCheckStep
", clarify_check_config=data_bias_check_config, check_job_config=check_job_config, skip_check=False, register_new_baseline=False supplied_baseline_constraints="s3://sagemaker-us-west-2-111122223333/baseline/analysis.json
", model_package_group_name="MyModelPackageGroup
" )
QualityCheck ステップ
QualityCheck
ステップを使用すると、パイプラインのデータの品質とモデルの品質についてのベースラインの提案とベースラインドリフトチェックを前のベースラインに対して実行できます。その後、ベースラインを生成してメソッドに登録し、model.register()
モデルステップそのメソッドの出力をを使用してに渡すことができますstep_args
。Model Monitor では、これらのドリフトチェック用のベースラインをモデルエンドポイントで使用できるため、ベースラインを別途提案する必要がなくなります。QualityCheck
ステップでは、ドリフトチェック用のベースラインをモデルレジストリからプルすることもできます。QualityCheck
このステップでは、Amazon SageMaker Model Monitor の事前構築済みコンテナを利用します。このコンテナには、特定のベースラインに対する制約の提案、統計の生成、制約の検証など、さまざまなモデルモニタリング機能が含まれています。詳細については、「Amazon SageMaker モデルモニターのビルド済みコンテナ」を参照してください。
QualityCheck ステップの設定
パイプラインで使用するたびに以下のチェックタイプの 1 つのみを実行するように QualityCheck
ステップを設定できます。
-
データの品質チェック
-
モデルの品質チェック
実行するには、quality_check_config
パラメータに次のいずれかのチェックタイプ値を指定します。
-
DataQualityCheckConfig
-
ModelQualityCheckConfig
QualityCheck
ステップでは、Model Monitor の構築済みコンテナを実行し、チェックと処理ジョブのための専用の設定を必要とする処理ジョブが起動します。QualityCheckConfig
と CheckJobConfig
はこれらの設定のヘルパー関数であり、Model Monitor におけるモデルの品質またはデータの品質をモニタリングするためのベースラインの作成方法に合わせて調整されます。Model Monitor のベースラインの提案については、「ベースラインの作成」と「モデル品質のベースラインを作成する」を参照してください。
ドリフトチェックのステップ動作を制御する
QualityCheck
ステップには、その動作を制御するために、以下の 2 つのブールフラグが必要です。
-
skip_check
: このパラメータでは、前のベースラインに対するドリフトチェックをスキップするかどうかを指定します。False
に設定されている場合、設定されたチェックタイプの前のベースラインが使用可能である必要があります。 -
register_new_baseline
: このパラメータでは、新しく計算されたベースラインにステッププロパティBaselineUsedForDriftCheckConstraints
とBaselineUsedForDriftCheckStatistics
を通じてアクセスできるかどうかを指定します。False
に設定されている場合、設定されたチェックタイプの前のベースラインも使用可能である必要があります。これらには、BaselineUsedForDriftCheckConstraints
とBaselineUsedForDriftCheckStatistics
のプロパティを通じてアクセスできます。
詳細については、「Amazon SageMaker Model Building Pipelines を使用したベースライン計算 ClarifyCheck 、ドリフト検出、ライフサイクル、 QualityCheck 手順」を参照してください。
ベースラインを操作する
前のベースラインは、supplied_baseline_statistics
と supplied_baseline_constraints
のパラメータを通じて直接指定できます。単に model_package_group_name
を指定すると、QualityCheck
ステップでモデルパッケージグループ内の最新の承認済みモデルパッケージの DriftCheckBaselines
がプルされます。model_package_group_name
、supplied_baseline_constraints
、supplied_baseline_statistics
を指定すると、QualityCheck
ステップでは、実行する QualityCheck
ステップのチェックタイプの supplied_baseline_constraints
と supplied_baseline_statistics
で指定したベースラインが使用されます。
QualityCheck
ステップ要件の使用方法の詳細については、sagemaker.workflow.steps を参照してください。 QualityCheck
SageMaker パイプラインでステップの使用方法を解説した Amazon SageMaker Studio ノートブックについては、「sagemaker-pipeline-model-monitor-clarify-steps.ipynb
例 データ品質チェックのための QualityCheck
ステップを作成する
from sagemaker.workflow.check_job_config import CheckJobConfig from sagemaker.workflow.quality_check_step import DataQualityCheckConfig, QualityCheckStep from sagemaker.workflow.execution_variables import ExecutionVariables check_job_config = CheckJobConfig( role=role, instance_count=1, instance_type="
ml.c5.xlarge
", volume_size_in_gb=120
, sagemaker_session=sagemaker_session, ) data_quality_check_config = DataQualityCheckConfig( baseline_dataset=step_process.properties.ProcessingOutputConfig.Outputs["train
"].S3Output.S3Uri, dataset_format=DatasetFormat.csv(header=False, output_columns_position="START"), output_s3_uri=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'dataqualitycheckstep']) ) data_quality_check_step = QualityCheckStep( name="DataQualityCheckStep", skip_check=False, register_new_baseline=False, quality_check_config=data_quality_check_config, check_job_config=check_job_config, supplied_baseline_statistics="s3://sagemaker-us-west-2-555555555555/baseline/statistics.json
", supplied_baseline_constraints="s3://sagemaker-us-west-2-555555555555/baseline/constraints.json
", model_package_group_name="MyModelPackageGroup
" )
Amazon EMR ステップ
Amazon SageMaker Model Building Pipelines の Amazon EMR ステップを使用して、実行中の Amazon EMR クラスターに対する EMR ステップを処理できます。詳細については、「Amazon EMR の開始方法」を参照してください。
Amazon EMR ステップでは、Amazon EMR クラスターで使用する JAR の Amazon S3 の場所、渡す引数、Amazon EMR クラスター ID が EMRStepConfig
に必要です。
-
Amazon EMR ステップでは、パイプラインに渡すロールに追加のアクセス許可が必要です。パイプラインロールにAWS管理ポリシー: AmazonSageMakerPipelinesIntegrationsをアタッチするか、ロールにそのポリシーのアクセス許可が含まれていることを確認してください。
-
Amazon EMR on EKS はサポートされていません。
-
EMR ステップは、
STARTING
、BOOTSTRAPPING
、RUNNING
、WAITING
のいずれかの状態にあるクラスターでのみ実行できます。 -
EMR クラスターでは、
PENDING
で最大 256 の EMR ステップを使用できます。この制限を超えて EMR ステップが送信されると、パイプラインの実行は失敗します。パイプラインのステップの再試行ポリシー の使用を検討できます。
例 EMR クラスターで新しいジョブを起動する Amazon EMR ステップ定義を作成します。
from sagemaker.workflow.emr_step import EMRStep, EMRStepConfig emr_config = EMRStepConfig( jar="s3://
path/to/jar/MyJar.jar
", # required, S3 path to jar args=["--verbose", "--force"], # optional list of arguments to pass to the jar main_class="com.my.Main1", # optional main class, this can be omitted if jar above has a manifest properties=[ # optional list of Java properties that are set when the step runs { "key": "mapred.tasktracker.map.tasks.maximum", "value": "2" }, { "key": "mapreduce.map.sort.spill.percent", "value": "0.90" }, { "key": "mapreduce.tasktracker.reduce.tasks.maximum", "value": "5" } ] ) step_emr = EMRStep ( name="EMRSampleStep", # required cluster_id="j-1YONHTCP3YZKC", # required step_config=emr_config, # required display_name="My EMR Step", description="Pipeline step to execute EMR job" )
失敗ステップ
FailStep
を使用して、必要な条件または状態が達成されなかったときに Amazon SageMaker Model Building Pipelines の実行を停止し、そのパイプラインの実行を失敗としてマークします。FailStep
を使用すると、カスタムエラーメッセージを入力し、パイプライン実行失敗の原因を示すこともできます。
FailStep
および他のパイプラインステップを同時に実行すると、すべての同時ステップが完了するまでパイプラインは終了しません。
FailStep
の使用に関する制限
-
FailStep
は他のステップのDependsOn
リストに追加できません。詳細については、「ステップ間のカスタム依存関係」を参照してください。 -
他のステップは
FailStep
を参照できません。それは、常にパイプラインの実行の最後のステップです。 -
FailStep
で終わるパイプライン実行を再試行することはできません。
FailStep
ErrorMessage
は、静的テキスト文字列の形式で作成できます。または、パイプラインパラメータの Join
次のコードスニペット例では、FailStep
でパイプラインパラメータと Join
オペレーションで設定した ErrorMessage
を使用しています。
from sagemaker.workflow.fail_step import FailStep from sagemaker.workflow.functions import Join from sagemaker.workflow.parameters import ParameterInteger mse_threshold_param = ParameterInteger(name="MseThreshold", default_value=5) step_fail = FailStep( name="
AbaloneMSEFail
", error_message=Join( on=" ", values=["Execution failed due to MSE >
", mse_threshold_param] ), )
ステップのプロパティ
properties
属性は、パイプラインのステップ間にデータの依存関係を追加するために使用されます。これらのデータ依存関係は、 SageMaker パイプラインによって使用され、パイプライン定義から DAG を作成します。これらのプロパティはプレースホルダー値として参照でき、実行時に解決されます。
SageMaker Pipelinesproperties
ステップの属性は、Describe
SageMaker 対応するジョブタイプの呼び出しによって返されるオブジェクトと一致します。ジョブタイプごとに、Describe
呼び出しは次のレスポンスオブジェクトを返します。
-
ProcessingStep
– DescribeProcessingJob -
TrainingStep
– DescribeTrainingJob -
TransformStep
– DescribeTransformJob
データ依存関係の作成時に各ステップタイプでどのプロパティが参照可能かを確認するには、Amazon SageMaker Python SDK
ステップの並列処理
ステップは、他のステップに依存していない場合、パイプライン実行時に直ちに実行されます。ただし、同時に実行するパイプラインステップが多すぎると、使用可能なリソースがすぐに使い果たされる可能性があります。ParallelismConfiguration
を使用して、パイプラインの同時実行ステップ数を制御します。
次の例では ParallelismConfiguration
を使用して同時実行ステップの上限数を 5 に設定しています。
pipeline.create( parallelism_config=ParallelismConfiguration(5), )
ステップ間のデータ依存関係
DAG の構造を定義するには、ステップ間のデータ関係を指定します。ステップ間にデータ依存関係を作成するには、パイプラインのあるステップのプロパティを別のステップに入力として渡します。入力を受け取るステップは、入力を提供するステップの実行が終了するまで開始されません。
データ依存関係は、 JsonPath 次の形式の表記で表されます。この形式は JSON プロパティファイルを走査します。つまり、ファイル内にネストされた目的のプロパティに到達するのに必要な数だけ <property>
インスタンスを追加できます。 JsonPath 表記の詳細については、JsonPath リポジトリを参照してください
<step_name>
.properties.<property>
.<property>
処理ステップの ProcessingOutputConfig
プロパティを使用して Amazon S3 バケットを指定する方法を以下に示します。
step_process.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri
データ依存関係を作成するには、以下のようにバケットをトレーニングステップに渡します。
from sagemaker.workflow.pipeline_context import PipelineSession sklearn_train = SKLearn(..., sagemaker_session=PipelineSession()) step_train = TrainingStep( name="CensusTrain", step_args=sklearn_train.fit(inputs=TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "train_data"].S3Output.S3Uri )) )
データ依存関係の作成時に各ステップタイプでどのプロパティが参照可能かを確認するには、Amazon SageMaker Python SDK
ステップ間のカスタム依存関係
データ依存関係を指定すると、 SageMaker Pipelines によってステップ間のデータが接続されます。また、 SageMaker Pipelines を直接使用せずに、あるステップから前のステップのデータにアクセスすることもできます。この場合、別のステップの実行が終了するまでステップを開始しないように SageMaker Pipelines に指示するカスタム依存関係を作成できます。カスタム依存関係を作成するには、ステップのDependsOn
属性を指定します。
以下の例では、ステップ A
とステップ B
の両方が実行を終了してから開始されるステップ C
を定義します。
{ 'Steps': [ {'Name':'A', ...}, {'Name':'B', ...}, {'Name':'C', 'DependsOn': ['A', 'B']} ] }
SageMaker 依存関係でサイクリック依存性が作成されると、Pipelines で検証例外がスローされます。
以下の例では、処理ステップの実行が終了した後に開始されるトレーニングステップを作成します。
processing_step = ProcessingStep(...) training_step = TrainingStep(...) training_step.add_depends_on([processing_step])
以下の例では、2 つの異なる処理ステップの実行が終了するまで開始されないトレーニングステップを作成します。
processing_step_1 = ProcessingStep(...) processing_step_2 = ProcessingStep(...) training_step = TrainingStep(...) training_step.add_depends_on([processing_step_1, processing_step_2])
以下は、カスタム依存関係を作成する別の方法です。
training_step.add_depends_on([processing_step_1]) training_step.add_depends_on([processing_step_2])
以下の例では、ある処理ステップから入力を受け取り、別の処理ステップの実行が終了するまで待機するトレーニングステップを作成します。
processing_step_1 = ProcessingStep(...) processing_step_2 = ProcessingStep(...) training_step = TrainingStep( ..., inputs=TrainingInput( s3_data=processing_step_1.properties.ProcessingOutputConfig.Outputs[ "train_data" ].S3Output.S3Uri ) training_step.add_depends_on([processing_step_2])
以下の例では、ステップのカスタム依存関係の文字列リストを取得する方法を示しています。
custom_dependencies = training_step.depends_on
ステップでカスタムイメージを使用する
パイプラインでステップを作成する際には、すべての SageMaker深層学習コンテナイメージ
パイプラインのステップでは独自のコンテナを使用することもできます。Amazon SageMaker Studio 内でイメージを作成することはできないため、Amazon SageMaker Model Building Pipelines で使用する前に、別の方法を使ってイメージを作成する必要があります。
パイプラインのステップの作成時に独自のコンテナを使用する場合は、推定器定義にイメージ URI を追加します。で独自のコンテナを使用する方法については SageMaker、「で Docker コンテナを使用する」を参照してください SageMaker。