新增步驟

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

執行程式碼步驟會將檔案上傳至 Amazon SageMaker AI 的預設 Amazon S3 儲存貯體。此儲存貯體可能未設定必要的跨來源資源共用 (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 文件。

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

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

若要使用拖放編輯器將微調模型步驟新增至您的管道，請遵循下列步驟：

使用 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 步驟。

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

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

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

註冊模型步驟會將模型註冊至 SageMaker 模型註冊庫。

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

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

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

若要在拖放編輯器中將模型部署步驟新增至您的管道，請完成下列步驟：

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

轉換步驟需要轉換器以及要對其執行批次轉換的資料。下列範例展示如何建立轉換步驟定義。如需轉換步驟要求的詳細資訊，請參閱 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 時，Pipelines 會將 Amazon SQS 訊息傳送到 SQS 佇列。SQS 訊息的主體包含設定為 Stopping 的狀態欄位。下列範例示範 SQS 訊息主體。

{
  "token": "26vcYbeWsZ",
  "pipelineExecutionArn": "arn:aws:sagemaker:us-east-2:012345678901:pipeline/callback-pipeline/execution/7pinimwddh3a",
  "arguments": {
    "number": 5,
    "stringArg": "some-arg",
    "inputData": "s3://sagemaker-us-west-2-012345678901/abalone/abalone-dataset.csv"
  },
  "status": "Stopping"
}

您應該將邏輯新增至 Amazon SQS 訊息取用者，以在收到訊息時採取任何必要的動作 (例如資源清理)。然後將呼叫新增至 SendPipelineExecutionStepSuccess 或 SendPipelineExecutionStepFailure。

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

您可以使用 Lambda 步驟來執行 AWS Lambda函數。您可以執行現有的 Lambda 函式，或 SageMaker AI 可以建立並執行新的 Lambda 函式。如果您選擇使用現有的 Lambda 函數，它必須與 SageMaker AI 管道AWS 區域位於相同的 中。如需顯示如何在 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。

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

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

設定 ClarifyCheck 步驟

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

若要這樣做，請使用下列其中一個檢查類型值來設定 clarify_check_config 參數：

ClarifyCheck 步驟會啟動執行 SageMaker Clarify AI 預先建置容器的處理任務，並需要專用的檢查和處理任務組態。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 AI SageMaker AI SDK for Python 中的 sagemaker.workflow.steps.ClarifyCheckStep。如需顯示如何在 Pipelines 中使用 ClarifyCheck 步驟的 Amazon SageMaker Studio Classic 筆記本，請參閱 sagemaker-pipeline-model-monitor-clarify-steps.ipynb。

from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.clarify_check_step import DataBiasCheckConfig, ClarifyCheckStep
from sagemaker.workflow.execution_variables import ExecutionVariables

check_job_config = CheckJobConfig(
    role=role,
    instance_count=1,
    instance_type="ml.c5.xlarge",
    volume_size_in_gb=120,
    sagemaker_session=sagemaker_session,
)

data_bias_data_config = DataConfig(
    s3_data_input_path=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    s3_output_path=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'databiascheckstep']),
    label=0,
    dataset_type="text/csv",
    s3_analysis_config_output_path=data_bias_analysis_cfg_output_path,
)

data_bias_config = BiasConfig(
    label_values_or_threshold=[15.0], facet_name=[8], facet_values_or_threshold=[[0.5]]  
)

data_bias_check_config = DataBiasCheckConfig(
    data_config=data_bias_data_config,
    data_bias_config=data_bias_config,
)h

data_bias_check_step = ClarifyCheckStep(
    name="DataBiasCheckStep",
    clarify_check_config=data_bias_check_config,
    check_job_config=check_job_config,
    skip_check=False,
    register_new_baseline=False
   supplied_baseline_constraints="s3://sagemaker-us-west-2-111122223333/baseline/analysis.json",
    model_package_group_name="MyModelPackageGroup"
)

使用 QualityCheck 步驟，針對管道中先前的資料品質或模型品質基準執行基準建議和漂移檢查。然後，您可以使用 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 是這些組態的協助程式函式。這些協助程式函式與 Model Monitor 為模型品質或資料品質監控建立基準的方式一致。如需有關模型監視器基線建議的詳細資訊，請參閱建立基準和建立模型品質基準。

控制漂移檢查的步驟行為

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

如需更多詳細資訊，請參閱Amazon SageMaker Pipelines 中具有 ClarifyCheck 和 QualityCheck 步驟的基準計算、漂移偵測和生命週期。

處理基準

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

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

如需使用 QualityCheck 步驟要求的詳細資訊，請參閱 Amazon SageMaker AI SageMaker AI SDK for Python 中的 sagemaker.workflow.steps.QualityCheckStep。如需顯示如何在 Pipelines 中使用 QualityCheck 步驟的 Amazon SageMaker Studio Classic 筆記本，請參閱 sagemaker-pipeline-model-monitor-clarify-steps.ipynb。

from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.quality_check_step import DataQualityCheckConfig, QualityCheckStep
from sagemaker.workflow.execution_variables import ExecutionVariables

check_job_config = CheckJobConfig(
    role=role,
    instance_count=1,
    instance_type="ml.c5.xlarge",
    volume_size_in_gb=120,
    sagemaker_session=sagemaker_session,
)

data_quality_check_config = DataQualityCheckConfig(
    baseline_dataset=step_process.properties.ProcessingOutputConfig.Outputs["train"].S3Output.S3Uri,
    dataset_format=DatasetFormat.csv(header=False, output_columns_position="START"),
    output_s3_uri=Join(on='/', values=['s3:/', your_bucket, base_job_prefix, ExecutionVariables.PIPELINE_EXECUTION_ID, 'dataqualitycheckstep'])
)

data_quality_check_step = QualityCheckStep(
    name="DataQualityCheckStep",
    skip_check=False,
    register_new_baseline=False,
    quality_check_config=data_quality_check_config,
    check_job_config=check_job_config,
    supplied_baseline_statistics="s3://sagemaker-us-west-2-555555555555/baseline/statistics.json",
    supplied_baseline_constraints="s3://sagemaker-us-west-2-555555555555/baseline/constraints.json",
    model_package_group_name="MyModelPackageGroup"
)

使用 Amazon SageMaker Pipelines EMR 步驟：

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

EMR 步驟要求 EMRStepConfig 包括 Amazon EMR 叢集所使用的 JAR 檔案的位置，以及要傳遞的任何引數。如果您想要在執行中的 EMR 叢集上執行步驟，您也會提供 Amazon EMR 叢集 ID。您也可以傳遞叢集組態，在其為您建立、管理和終止的叢集上執行 EMR 步驟。以下章節包括示範這兩種方法的範例和範例筆記本連結。

將 Amazon EMR 步驟新增至管道

若要將 EMR 步驟新增至管道，請執行下列動作：

在正在執行的 Amazon EMR 叢集上啟動新任務

若要在執行中的 Amazon EMR 叢集上啟動新任務，請將叢集 ID 作為字串傳遞至 EMRStep 的 cluster_id 引數。下列範例示範此程序。

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

如需逐步引導您完成整個範例的範例筆記本，請參閱 Pipelines EMR Step With Running EMR Cluster。

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

如需逐步引導您完成整個範例的範例筆記本，請參閱 Pipelines EMR Step With Cluster Lifecycle Management。

若要將 EMR 無伺服器步驟新增至管道，請執行下列動作：

使用 NotebookJobStep 以非互動方式將您的 SageMaker 筆記本任務做為管道步驟執行。如果您在 Pipelines 拖放 UI 中建置管道，請使用 執行程式碼步驟 來執行筆記本。如需 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 筆記本任務的映像限制條件。

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

使用失敗步驟的限制

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