新增步驟

在管道drag-and-dropUI 中，您可以使用執行程式碼步驟，將您自己的程式碼執行為管道步驟。您可以上傳 Python 函數、指令碼或筆記本，做為管道的一部分執行。如果您想要協調利用進階 SageMaker AI 功能或其他 AWS 服務的自訂 ML 任務，您應該使用此步驟。

執行程式碼步驟會將檔案上傳至 Amazon SageMaker AI 的預設 Amazon S3 儲存貯體。 Amazon SageMaker 此儲存貯體可能沒有設定必要的跨來源資源共享 (CORS) 許可。若要進一步了解如何設定 CORS 許可，請參閱 輸入映像資料的 CORS 需求。

執行程式碼步驟使用 Amazon SageMaker 訓練任務來執行程式碼。確保您的 IAM 角色具有 sagemaker:DescribeTrainingJob和 sagemaker:CreateTrainingJob API 許可。若要進一步了解 Amazon SageMaker AI 的所有必要許可，以及如何設定這些許可，請參閱 Amazon SageMaker AI API 許可：動作、許可和資源參考。

若要使用管道設計工具將執行程式碼步驟新增至管道，請執行下列動作：

使用處理步驟建立用於資料處理的處理任務。如需有關處理工作的詳細資訊，請參閱處理資料和評估模型。

您使用訓練步驟建立訓練任務來訓練模型。如需訓練任務的詳細資訊，請參閱使用 Amazon SageMaker AI 訓練模型。

訓練步驟需要估算器以及訓練和驗證資料輸入。

您可以使用調校步驟來建立超參數調校任務，也稱為超參數最佳化 (HPO)。超參數調校任務會執行多個訓練任務，每個任務都會產生模型版本。如需有關超參數調校的詳細資訊，請參閱使用 SageMaker AI 自動調校模型。

調校任務與管道的 SageMaker AI 實驗相關聯，訓練任務建立為試驗。如需詳細資訊，請參閱Experiments 整合。

調校步驟需要 HyperparameterTuner 和訓練輸入。您可以指定 HyperparameterTuner 之 warm_start_config 參數來重新訓練先前的調校任務。如需有關超參數調整和熱啟動的詳細資訊，請參閱執行超參數調校任務的暖啟動。

您使用 sagemaker.workflow.steps.TuningStep 類別的 get_top_model_s3_uri 方法，從效能最佳的模型版本之一取得模型成品。如需說明如何在 SageMaker AI 管道中使用調校步驟的筆記本，請參閱 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://amzn-s3-demo-bucket/my-data"))
)

取得最佳模型版本

下列範例示範如何使用 get_top_model_s3_uri 方法從調校任務取得最佳模型版本。根據 HyperParameterTuningJobObjective，最多提供效能排名前 50 的版本。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 文件。

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://amzn-s3-demo-bucket/my-data"))
)

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()
    ),
    ...
)

微調會在新的資料集上，從 Amazon SageMaker JumpStart 訓練預先訓練的基礎模型。這個程序也稱為移轉學習，可以利用較小的資料集和較短的訓練時間來產生精確的模型。當您微調模型時，您可以使用預設資料集或選擇自己的資料。若要進一步了解如何從 JumpStart 微調基礎模型，請參閱 微調模型。

微調步驟使用 Amazon SageMaker 訓練任務來自訂模型。確保您的 IAM 角色具有 sagemaker:DescribeTrainingJob和 sagemaker:CreateTrainingJob API 許可，可在管道中執行微調任務。若要進一步了解 Amazon SageMaker AI 的必要許可以及如何設定這些許可，請參閱 Amazon SageMaker AI API 許可：動作、許可和資源參考。

若要使用drag-and-drop編輯器將微調模型步驟新增至管道，請遵循下列步驟：

使用 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://amzn-s3-demo-bucket/my-training-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="training",
)
input_validation = AutoMLInput(
    inputs="s3://amzn-s3-demo-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 任務取得具有最佳目標指標的模型，如下所示。您也必須使用 IAM role 來存取模型成品。

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

如需詳細資訊，請參閱 SageMaker Python SDK 中的 AutoML 步驟。

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://amzn-s3-demo-bucket/my-training-data",
    target_attribute_name="my_target_attribute_name",
    channel_type="training",
)
input_validation = AutoMLInput(
    inputs="s3://amzn-s3-demo-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,
)     

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

使用 ModelStep建立或註冊 SageMaker AI 模型。如需有關 ModelStep 要求的更多資訊，請參閱 sagemaker.workflow.model_step.ModelStep 文件。

建立模型

您可以使用 ModelStep來建立 SageMaker AI 模型。ModelStep 需要模型成品和有關 SageMaker AI 執行個體類型的資訊，您需要這些執行個體類型來建立模型。如需 SageMaker AI 模型的詳細資訊，請參閱使用 Amazon SageMaker AI 訓練模型。

下列範例示範如何建立 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向 Amazon SageMaker 模型登錄檔註冊 sagemaker.model.Model或 sagemaker.pipeline.PipelineModel 。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,
)

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

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

您可以使用建立模型步驟來建立 SageMaker AI 模型。如需 SageMaker AI 模型的詳細資訊，請參閱使用 Amazon SageMaker 訓練模型。

建立模型步驟需要模型成品，以及建立模型時需要使用的 SageMaker AI 執行個體類型相關資訊。下列範例示範如何建立建立模型步驟定義。如需建立模型步驟需求的詳細資訊，請參閱 sagemaker.workflow.steps.CreateModelStep 文件。

註冊模型步驟會將模型註冊到 SageMaker 模型登錄檔。

在管道設計工具中，使用部署模型 （端點） 步驟，將您的模型部署到端點。您可以建立新的端點或使用現有的端點。即時推論非常適合您具有即時、互動、低延遲需求的推論工作負載。您可以將模型部署到 SageMaker AI 託管服務，並取得可用於推論的即時端點。這些端點是完全受管的，並支援自動擴展。若要進一步了解 SageMaker AI 中的即時推論，請參閱即時推論。

在將部署模型步驟新增至管道之前，請確定您的 IAM 角色具有下列許可：

若要進一步了解 SageMaker AI 的所有必要許可，以及如何設定這些許可，請參閱 Amazon SageMaker AI API 許可：動作、許可和資源參考。

若要在drag-and-drop編輯器中將模型部署步驟新增至管道，請完成下列步驟：

您可以使用批次轉換的轉換步驟對整個資料集執行推論。有關批次轉換的更多資訊，請參閱使用推論管道進行批次轉換。

轉換步驟需要轉換器以及要對其執行批次轉換的資料。下列範例示範如何建立轉換步驟定義。如需轉換步驟需求的詳細資訊，請參閱 sagemaker.workflow.steps.TransformStep 文件。

您可以使用條件步驟來評估步驟屬性的條件，以評估接下來應該在管道中採取哪些動作。

條件步驟需要：

使用Callback步驟將非由 Amazon SageMaker Pipelines 直接提供的其他程序和 AWS 服務新增至工作流程。Callback 步驟執行時，會發生下列程序：

如需有關 Callback 步驟要求的詳細資訊，請參閱工作流程回調步驟回調步驟文件。有關完整的解決方案，請參閱擴展 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 AI to complete the step
            sagemaker_client.send_pipeline_execution_step_success(
                CallbackToken=token,
                OutputParameters={...}
            )
'

停止行為

Callback 步驟執行期間，管道程序不會停止。

當您使用執行Callback步驟呼叫管道程序上的 StopPipelineExecution 時，管道會將 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。

只有當管道收到其中一個呼叫時，才會停止管道程序。

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 AI to complete the step
            sagemaker_client.send_pipeline_execution_step_success(
                CallbackToken=token,
                OutputParameters={...}
            )
'

{
  "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"
}

您可以使用 Lambda 步驟來執行 AWS Lambda 函數。您可以執行現有的 Lambda 函數，或 SageMaker AI 可以建立和執行新的 Lambda 函數。如需說明如何在 SageMaker AI 管道中使用 Lambda 步驟的筆記本，請參閱 sagemaker-pipelines-lambda-step.ipynb。

SageMaker AI 提供 sagemaker.lambda_helper.Lambda 類別來建立、更新、叫用和刪除 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 類型的 lambda_func 引數。若要調用現有的 Lambda 函式，唯一的要求是將函式的 Amazon Resource Name (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 類別具有一個 timeout 引數，可指定 Lambda 函式可以執行的最長時間。預設值是 120 秒，上限值為 10 分鐘。如果 Lambda 函式在達到逾時時間時正在執行，則 Lambda 步驟會失敗；不過，Lambda 函式會繼續執行。

Lambda 步驟執行時，無法停止管道程序，因為 Lambda 步驟調用的 Lambda 函式無法停止。如果您在 Lambda 函數執行時停止程序，管道會等待函數完成或直到逾時為止。這取決於哪個先發生。程序接著會停止。如果 Lambda 函式完成，則管道程序狀態為 Stopped。如果超時到達，則管道程序狀態為 Failed。

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

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

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

您可以使用 ClarifyCheck 步驟可對照先前的基準執行基準漂移檢查，以便進行偏差分析和模型可解釋性。然後，您可以使用 model.register() 方法產生並註冊基準，然後使用 step_args 將該方法的輸出傳遞給 模型步驟。Amazon SageMaker Model Monitor 可以針對模型端點使用這些基準進行偏離檢查。因此，您不需要分別提出基準建議。

ClarifyCheck 步驟也可以從模型註冊中提取漂移檢查基準。步驟ClarifyCheck使用 SageMaker Clarify 預先建置的容器。此容器提供一系列模型監控功能，包括針對指定基準的限制建議和限制驗證。如需詳細資訊，請參閱預先建置的 SageMaker Clarify 容器。

設定 ClarifyCheck 步驟

您可以設定 ClarifyCheck 步驟，以便每次在管道中使用該步驟時僅執行下列其中一種可用的檢查類型。

若要執行此作業，請使用下列其中一個檢查類型值來設定 clarify_check_config 參數：

ClarifyCheck 步驟會啟動執行 SageMaker AI Clarify 預先建置容器的處理任務，並需要檢查和處理任務的專用組態。 ClarifyCheckConfig和 CheckJobConfig是這些組態的協助程式函數。這些協助程式函數符合 SageMaker Clarify 處理任務運算的方式，以檢查模型偏差、資料偏差或模型可解釋性。如需詳細資訊，請參閱執行 SageMaker Clarify 處理任務以進行偏差分析和解釋性。

控制漂移檢查的步驟行為

ClarifyCheck 步驟需要下列兩個布林標記來控制其行為：

如需更多詳細資訊，請參閱使用 Amazon SageMaker Pipelines 中的 ClarifyCheck 和 QualityCheck 步驟進行基準計算、偏離偵測和生命週期。

處理基準

您可以選擇性地指定 model_package_group_name來尋找現有的基準。然後，ClarifyCheck步驟會在模型套件群組中提取DriftCheckBaselines最新核准的模型套件上的 。

或者，您可以透過 supplied_baseline_constraints 參數提供先前的基準。如果同時指定 model_package_group_name 和 supplied_baseline_constraints，則 ClarifyCheck 步驟會使用 supplied_baseline_constraints 參數指定的基準。

如需使用ClarifyCheck步驟需求的詳細資訊，請參閱 Amazon SageMaker SageMaker SageMaker AI SageMaker AI SDK for Python 中的 sagemaker.workflow.steps.ClarifyCheckStep。如需顯示如何在管道中使用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"
)

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 步驟，針對管道中的資料品質或模型品質，對先前的基準執行基準建議和偏離檢查。然後，您可以使用 model.register()方法產生並註冊基準，並將該方法的輸出傳遞給 模型步驟 step_args。 】

Model Monitor 可針對模型端點使用這些基準進行漂移檢查，因此您不需要單獨執行基準建議。QualityCheck 步驟也可以從模型註冊中提取漂移檢查基準。QualityCheck 步驟會利用 Amazon SageMaker AI Model Monitor 預先建置的容器。此容器具有一系列模型監控功能，包括針對基準的限制建議、統計資料產生和限制驗證。如需詳細資訊，請參閱Amazon SageMaker Model Monitor 預建容器。

設定 QualityCheck 步驟

您可以將QualityCheck步驟設定為每次在管道中使用時，只執行下列其中一種檢查類型。

為了執行此動作，您可以使用下列其中一個檢查類型值來設定 quality_check_config 參數：

QualityCheck 步驟會啟動執行 Model Monitor 預先建置容器的處理任務，並需要檢查和處理任務的專用組態。QualityCheckConfig 和 CheckJobConfig是這些組態的協助程式函數。這些協助程式函數符合模型監控如何建立模型品質或資料品質監控的基準。如需有關模型監視器基線建議的詳細資訊，請參閱建立基準和建立模型品質基準。

控制漂移檢查的步驟行為

QualityCheck 步驟需要下列兩個布林標記來控制其行為：

如需更多詳細資訊，請參閱使用 Amazon SageMaker Pipelines 中的 ClarifyCheck 和 QualityCheck 步驟進行基準計算、偏離偵測和生命週期。

處理基準

您可以直接透過 supplied_baseline_statistics和 supplied_baseline_constraints 參數指定先前的基準。您也可以在模型套件群組中指定 model_package_group_name，QualityCheck步驟會在最新核准的模型套件DriftCheckBaselines上提取 。

當您指定下列項目時，QualityCheck步驟會使用 supplied_baseline_constraints和 supplied_baseline_statistics 所指定的基準，檢查QualityCheck步驟的類型。

如需使用QualityCheck步驟需求的詳細資訊，請參閱 Amazon SageMaker SageMaker AI SageMaker AI SDK for Python 中的 sagemaker.workflow.steps.QualityCheckStep。如需顯示如何在管道中使用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"
)

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 SageMaker 管道 EMR 步驟來：

如需有關 Amazon EMR 的詳細資訊，請參閱 Getting started with Amazon EMR。

EMR 步驟需要 EMRStepConfig 包含 Amazon EMR 叢集所使用的 JAR 檔案位置，以及要傳遞的任何引數。如果您想要在執行中的 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"
)

如需引導您完成完整範例的範例筆記本，請參閱管道 EMR 步驟與執行中的 EMR 叢集。

在新的 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
)

如需引導您完成完整範例的範例筆記本，請參閱管道 EMR 步驟與叢集生命週期管理。

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

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
)

使用 NotebookJobStep以非互動方式執行 SageMaker Notebook 任務作為管道步驟。如果您在管道drag-and-dropUI 中建置管道，請使用 執行程式碼步驟來執行筆記本。如需 SageMaker 筆記本任務的詳細資訊，請參閱 SageMaker 筆記本工作。

至少NotebookJobStep需要輸入筆記本、映像 URI 和核心名稱。如需有關筆記本任務步驟需求以及您可以設定為自訂步驟的其他參數的詳細資訊，請參閱 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 筆記本任務。因此，請在 Studio Classic UI 筆記本任務儀表板中追蹤執行狀態，方法是在tags引數中包含特定標籤。如需要包含之標籤的詳細資訊，請參閱在 Studio UI 儀表板中檢視您的筆記本任務。

此外，如果您使用 SageMaker Python SDK 排程筆記本任務，您只能指定特定映像來執行筆記本任務。如需詳細資訊，請參閱SageMaker AI Python SDK 筆記本任務的影像限制。

from sagemaker.workflow.notebook_job_step import NotebookJobStep


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

當未達到所需的條件或狀態時，請使用失敗步驟來停止 Amazon SageMaker 管道執行。Fail 步驟也可讓您輸入自訂錯誤訊息，指出管道執行失敗的原因。

使用失敗步驟的限制

您可以建立靜態文字字串形式的失敗步驟錯誤訊息。或者，如果您使用 SDK，也可以使用管道參數、聯結操作或其他步驟屬性來建立更明智的錯誤訊息。