新增步驟

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

Execute Code 步驟會將檔案上傳至 Amazon 的預設 Amazon S3 儲存貯體 SageMaker。此儲存貯體可能沒有必要的跨來源資源共用 （CORS） 許可集。若要進一步了解設定CORS許可，請參閱 CORS 輸入映像資料的需求。

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

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

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

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

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

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

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

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

您可以使用 sagemaker.workflow.steps.TuningStep 類別的 get_top_model_s3_uri 方法，從其中一個效能最佳的模型版本取得模型成品。如需顯示如何在 SageMaker 管道中使用調校步驟的筆記本，請參閱 sagemaker-pipelines-tuning-step.ipynb。

下列範例示範如何建立 TuningStep 定義。

from sagemaker.workflow.pipeline_context import PipelineSession

from sagemaker.tuner import HyperparameterTuner
from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TuningStep

tuner = HyperparameterTuner(..., sagemaker_session=PipelineSession())
    
step_tuning = TuningStep(
    name = "HPTuning",
    step_args = tuner.fit(inputs=TrainingInput(s3_data="s3://amzn-s3-demo-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 文件。

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

微調步驟使用 Amazon SageMaker 訓練任務來自訂模型。確保您的IAM角色具有 sagemaker:DescribeTrainingJob和 sagemaker:CreateTrainingJobAPI許可，可在管道中執行微調任務。若要進一步了解 Amazon 所需的許可 SageMaker 以及如何設定這些許可，請參閱 Amazon SageMaker 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 任務取得具有最佳目標指標的模型。您還必須使用 IAMrole來存取模型成品。

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

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

使用 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向 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 模型。如需 SageMaker 模型的詳細資訊，請參閱 用 Amazon 訓練模型 SageMaker。

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

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

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

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

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

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

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

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

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

條件步驟需要：

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

如需Callback步驟需求的詳細資訊，請參閱 sagemaker.workflow.callback_step.CallbackStep 文件。如需完整的解決方案，請參閱擴展 SageMaker 管道以使用回呼步驟 包含自訂步驟。

下列範例顯示上述程序的實作。

from sagemaker.workflow.callback_step import CallbackStep

step_callback = CallbackStep(
    name="MyCallbackStep",
    sqs_queue_url="https://sqs.us-east-2.amazonaws.com/012345678901/MyCallbackQueue",
    inputs={...},
    outputs=[...]
)

callback_handler_code = '
    import boto3
    import json

    def handler(event, context):
        sagemaker_client=boto3.client("sagemaker")

        for record in event["Records"]:
            payload=json.loads(record["body"])
            token=payload["token"]

            # Custom processing

            # Call SageMaker to complete the step
            sagemaker_client.send_pipeline_execution_step_success(
                CallbackToken=token,
                OutputParameters={...}
            )
'

停止行為

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

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

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

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

SageMaker 提供 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_func引數Lambda。若要叫用現有的 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啟動執行 Clarify 預先建置容器的 SageMaker處理任務，並需要檢查和處理任務 的專用組態。 ClarifyCheckConfig和 CheckJobConfig是這些組態的協助程式函數。這些協助程式函數與 SageMaker Clarify 處理任務如何運算，以檢查模型偏差、資料偏差或模型可解釋性。如需詳細資訊，請參閱執行 SageMaker 澄清處理任務以進行偏差分析和可解釋性。

控制漂移檢查的步驟行為

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

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

處理基準

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

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

如需使用ClarifyCheck步驟需求的詳細資訊，請參閱 Amazon SageMaker SageMaker SDK for Python 中的 sagemaker.workflow.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"
)

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

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

設定 QualityCheck 步驟

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

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

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

控制漂移檢查的步驟行為

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

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

處理基準

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

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

如需使用QualityCheck步驟需求的詳細資訊，請參閱 Amazon SageMaker SageMaker 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"
)

使用 Amazon SageMaker Pipelines EMR 步驟來：

如需 Amazon 的詳細資訊EMR，請參閱 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步驟 。

使用 NotebookJobStep以非互動式方式執行 SageMaker 筆記本任務，作為管道步驟。如果您在管道 drag-and-drop 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，您只能指定特定影像來執行筆記本任務。如需詳細資訊，請參閱Python SDK 筆記本任務的影像限制 SageMaker 。

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

使用失敗步驟的限制

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