翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。
パイプラインのステップ
SageMaker パイプラインは、ステップで構成されます。これらのステップでは、プロパティを使用して、パイプラインが実行するアクションとステップ間の関係を定義します。
ステップタイプ
次に、各ステップタイプの要件について説明し、ステップの実装例を紹介します。これらは、必要なリソースと入力を提供しないため、動作可能な実装ではありません。これらのステップの実装に関するチュートリアルについては、「作成と管理 SageMaker パイプライン」を参照してください。
アマゾン SageMaker モデルビルディングパイプラインでは、以下のステップタイプをサポートしています。
処理ステップ
処理ステップを使用して、データ処理用の処理ジョブを作成します。ジョブの処理の詳細については、「データを処理して、モデルを評価する」を参照してください。
処理ステップには、プロセッサ、処理コードを定義する 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
パラメータを指定すると、以前のチューニングジョブを再トレーニングできます。ハイパーパラメータのチューニングとウォームスタートの詳細については、「ウォームスタートのハイパーパラメータ調整ジョブを実行する」を参照してください。
お使いいただくのはget_top_model_s3_uri
重要
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 ステップ
を使うAutoML
注記
現在、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
モデルステップ
を使用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.modelPipelineModel
は推論パイプラインを表します。これは、推論要求を処理する連続したコンテナで構成されるモデルです。
モデルの登録方法の詳細については、「モデルレジストリを使用したモデルの登録とデプロイ」を参照してください。詳細については、詳細について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。ConditionStep
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
重要
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( 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 ステップ 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
手順の要件については、を参照してください。セージメーカー. ワークフロー. ステップ。ClarifyCheckStepClarifyCheck
ステップイン 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
手順の要件については、を参照してください。セージメーカー. ワークフロー. ステップ。QualityCheckStepQualityCheck
ステップイン 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 ステップでは次のことが必要です。EMRStepConfig
Amazon EMR クラスターが使用する JAR ファイルの場所と、渡される引数が含まれます。また、実行中の EMR クラスターでステップを実行する場合は Amazon EMR クラスター ID を指定し、EMR ステップが自動的に作成、管理、終了するクラスターで実行する場合はクラスター設定を指定します。以下のセクションには、両方の方法を示す例とサンプルノートブックへのリンクが含まれています。
注記
-
EMR ステップでは、パイプラインに渡されるロールに追加権限が必要です。添付してくださいAWS管理ポリシー:
AmazonSageMakerPipelinesIntegrations
Pipeline ロールで使用するか、ロールにそのポリシー内のアクセス権限が含まれていることを確認します。 -
EMR ステップは EMR サーバーレスでも、EKS 上の Amazon EMR でもサポートされていません。
-
実行中のクラスターで EMR ステップを処理する場合、以下のいずれかの状態のクラスターのみを使用できます。
STARTING
、BOOTSTRAPPING
、RUNNING
、またはWAITING
。 -
実行中のクラスターで EMR ステップを処理する場合、1 つのクラスターには最大 256 の EMR ステップを含めることができます。
PENDING
EMR クラスター上の状態 この制限を超えて送信された EMR ステップは、パイプラインの実行に失敗します。パイプラインのステップの再試行ポリシー の使用を検討できます。 クラスタ ID またはクラスタ構成のいずれかを指定できますが、両方は指定できません。
EMR ステップはアマゾンに依存しています EventBridge EMR ステップまたはクラスターの状態の変化を監視します。実行中のクラスターで Amazon EMR ジョブを処理する場合、EMR ステップは
SageMakerPipelineExecutionEMRStepStatusUpdateRule
EMR ステップ状態を監視するルール。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 をビルドする際には、これらのプロパティはプレースホルダー値として参照でき、実行時に解決されます。
はproperties
aのアトリビュート SageMaker パイプラインステップはによって返されたオブジェクトと一致します。Describe
対応するものを呼び出します。 SageMaker ジョブの種類 ジョブタイプごとに、Describe
呼び出しは次のレスポンスオブジェクトを返します。
-
ProcessingStep
– DescribeProcessingJob -
TrainingStep
– DescribeTrainingJob -
TransformStep
– DescribeTransformJob
データ依存関係の作成中に各ステップタイプでどのプロパティが参照可能かを確認するには、を参照してください。データ依存関係-プロパティリファレンス
ステップの並列処理
ステップは、他のステップに依存していない場合、パイプライン実行時に直ちに実行されます。ただし、同時に実行するパイプラインステップが多すぎると、使用可能なリソースがすぐに使い果たされる可能性があります。ParallelismConfiguration
を使用して、パイプラインの同時実行ステップ数を制御します。
次の例では ParallelismConfiguration
を使用して同時実行ステップの上限数を 5 に設定しています。
pipeline.create( parallelism_config=ParallelismConfiguration(5), )
ステップ間のデータ依存関係
DAG の構造を定義するには、ステップ間のデータ関係を指定します。ステップ間にデータ依存関係を作成するには、パイプラインのあるステップのプロパティを別のステップに入力として渡します。入力を受け取るステップは、入力を提供するステップの実行が終了するまで開始されません。
データ依存関係では以下を使用します。 JsonPath 次の形式での表記です。この形式は JSON プロパティファイルを走査します。つまり、ファイル内にネストされた目的のプロパティに到達するのに必要な数だけ <property>
インスタンスを追加できます。詳細については、詳細について JsonPath 表記法については、を参照してください。JsonPath レポ
<step_name>
.properties.<property>
.<property>
処理ステップの ProcessingOutputConfig
プロパティを使用して Amazon S3 バケットを指定する方法を以下に示します。
step_process.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri
データ依存関係を作成するには、以下のようにバケットをトレーニングステップに渡します。
from sagemaker.workflow.pipeline_context import PipelineSession sklearn_train = SKLearn(..., sagemaker_session=PipelineSession()) step_train = TrainingStep( name="CensusTrain", step_args=sklearn_train.fit(inputs=TrainingInput( s3_data=step_process.properties.ProcessingOutputConfig.Outputs[ "train_data"].S3Output.S3Uri )) )
データ依存関係の作成中に各ステップタイプでどのプロパティが参照可能かを確認するには、を参照してください。データ依存関係-プロパティリファレンス
ステップ間のカスタム依存関係
データ依存関係を指定すると、 SageMaker Pipelines で、ステップ間のデータ接続をする際には、数分の時間がかかります。または、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.