Amazon SageMaker 管道步驟

管道由步驟組成。這些步驟使用屬性定義管道採取的動作以及步驟之間的關係。

步驟類型

下文說明了每個步驟類型的要求，並提供步驟的範例實作。這些不是工作的實現，因為它們不提供所需的資源和輸入。有關實現這些步驟的教學課程，請參閱建立和管理管道。

Amazon SageMaker 管道支援下列步驟類型：

@step 裝飾器

如果您想要協調利用進階 SageMaker 功能或其他功能的自訂 ML 工作 AWS drag-and-drop 管道 UI 中的服務，請使用執行程式碼步驟.

您可以使用@step裝飾器從本機機器學習程式碼建立步驟。測試程式碼之後，您可以使用裝飾器註解函式，將函數轉換為 SageMaker 管線步驟。@step當您將 @step-飾函數的輸出作為步傳遞到管道時，管道會創建並運行管道。您還可以創建一個多步驟DAG管道，其中包括一個或多個@step裝飾功能以及傳統的 SageMaker 管道步驟。如需如何使用@step裝飾器建立步驟的詳細資訊，請參閱L 與 @step 裝飾器的 ift-and-shift Python 代碼。

執行程式碼步驟

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

「執行程式碼」步驟會將檔案上傳到 Amazon 的預設 Amazon S3 儲存貯體 SageMaker。此值區可能沒有設定必要的跨來源資源共用 (CORS) 權限。若要進一步瞭解如何設定CORS權限，請參閱CORS權限要求。

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

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

處理步驟

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

訓練步驟

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

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

調校步驟

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

調整工作與管道的 SageMaker 實驗相關聯，並將訓練工作建立為試驗。如需詳細資訊，請參閱Experiments 整合。

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

您可以使用箭頭. 工作流程步驟的 get_top_model_s3_ uri 方法。 TuningStepclass，以從其中一個效能最佳的模型版本取得模型加工品。如需示範如何在 SageMaker 管線中使用調整步驟的筆記本，請參閱 sagemaker-pipelines-tuning-step.ip ynb。

下列範例示範如何建立 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()
    ),
    ...
)

如需調整步驟需求的詳細資訊，請參閱下垂器工作流程。步驟。 TuningStep文件。

微調步驟

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

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

若要使用 drag-and-drop 編輯器將「微調」模型步驟新增至管線，請依照下列步驟執行：

AutoML 步驟

使用 AutoML 建API立 AutoML 工作以自動訓練模型。如需 AutoML 任務的詳細資訊，請參閱使用 Amazon 自動輔助駕駛 SageMaker 自動執行模型開發。

下列範例示範如何使用 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 SDK 中的 AutoML 步驟。

模型步驟

使用ModelStep建立或註冊 SageMaker 模型。如ModelStep需需求的詳細資訊，請參閱下垂器. 工作流程. 模型 _ 步驟。 ModelStep文件。

建立模型

您可以使用ModelStep來建立 SageMaker 模型。A 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.PipelineModelPipelineModel 表示推論管道，是一種由直線順序容器構成的模型，可處理推論請求。如需有關如何註冊模型的詳細資訊，請參閱使用模型註冊表註冊和部署模型。

下列範例示範如何建立註冊 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 執行個體類型的相關資訊。下列範例顯示如何建立「建立模型」步驟定義。如需有關「建立模型」步驟需求的詳細資訊，請參閱下垂器 .workflow .step。 CreateModelStep文件。

註冊模型步驟

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

部署模型 (端點) 步驟

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

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

若要深入瞭解的所有必要權限以 SageMaker 及如何設定權限，請參閱Amazon SageMaker API 許可：動作、許可和資源參考。

若要在 drag-and-drop 編輯器中將模型部署步驟新增至您的 Pipeline，請完成以下步驟：

轉換步驟

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

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

條件步驟

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

條件步驟需要：

回呼步驟

使用Callback步驟來新增其他程序和 AWS 您工作流程中不是由 Amazon SageMaker 管道直接提供的服務。Callback 步驟執行時，會發生下列程序：

如需有關步Callback驟需求的詳細資訊，請參閱步驟。 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。

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

Lambda 步驟

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

SageMaker 提供用於建立、更新、叫用和刪除 Lambda 函數的 SAGEMA_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
    ...
)

箭頭. 工作流程 .lambda_ 步驟。 LambdaStepclass 具有類型的lambda_func引數Lambda。若要叫用現有的 Lambda 函數，唯一的要求是將函數的 Amazon 資源名稱 (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 步驟

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

ClarifyCheck 步驟也可以從模型註冊中提取漂移檢查基準。該ClarifyCheck步驟使用 SageMaker 澄清預構建容器。此容器提供一系列模型監視功能，包括對指定基準線的約束建議和條件約束驗證。如需詳細資訊，請參閱開始使用 SageMaker 澄清容器。

設定 ClarifyCheck 步驟

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

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

此ClarifyCheck步驟會啟動執行 Clearing 預先建置容器的 SageMaker處理工作，並需要進行檢查和處理工作的專用組態。 ClarifyCheckConfig並且CheckJobConfig是這些配置的輔助函數。這些輔助函數與 Criven 處理工作計算方式以檢查模型偏差、資料偏差或模型說 SageMaker 明的方式保持一致。如需詳細資訊，請參閱執行 SageMaker 澄清處理工作以進行偏差分析和解釋。

控制漂移檢查的步驟行為

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

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

處理基準

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

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

如需有關使用ClarifyCheck步驟需求的詳細資訊，請參閱下垂器工作流程。步驟。 ClarifyCheckStep在 Amazon SageMaker SageMaker SDK 的 Python。對於顯示如何在管道中使用ClarifyCheck步驟的 Amazon SageMaker 工作室經典筆記本，請參閱sagemaker-pipeline-model-monitor明確步驟 .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 步驟

使用此QualityCheck步驟對管道中的資料品質或模型品質之前的基準執行基準建議和漂移檢查。然後，您可以模型步驟使用該model.register()方法生成並註冊基準線，並將該方法的輸出傳遞給 use step_args。]

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

設定 QualityCheck 步驟

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

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

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

控制漂移檢查的步驟行為

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

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

處理基準

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

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

如需有關使用QualityCheck步驟需求的詳細資訊，請參閱下垂器工作流程。步驟。 QualityCheckStep在 Amazon SageMaker SageMaker SDK 的 Python。對於顯示如何在管道中使用QualityCheck步驟的 Amazon SageMaker 工作室經典筆記本，請參閱sagemaker-pipeline-model-monitor明確步驟 .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"
)

EMR步驟

使用 Amazon SageMaker 管道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 筆記本 Job 作為管線步驟。如果您在 Pipeline drag-and-drop UI 中建置管道，請使用執行程式碼步驟來執行您的筆記本。如需 SageMaker 記事本工作的詳細資訊，請參閱SageMaker 筆記本工作。

A 至少NotebookJobStep需要輸入筆記本，映像URI和內核名稱。如需有關「記事本 Job」步驟需求和其他可設定來自訂步驟的參數的詳細資訊，請參閱 sagemaker.workflow.step。 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 筆記本工作。因此，透過包含tags引數的特定標籤，在 Studio Classic UI 筆記本工作儀表板中追蹤執行狀態。如需要包含之標籤的更多詳細資訊，請參閱在 Studio UI 儀表板中檢視筆記本工作。

此外，如果您使用 SageMaker Python 排程筆記本工作SDK，則只能指定執行筆記本工作的特定影像。如需詳細資訊，請參閱 SageMakerPython SDK 筆記型電腦工作的影像限制。

失敗步驟

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

使用「失敗」步驟的限制

您可以在靜態文本字符串的形式創建失敗步驟錯誤消息。或者，您也可以使用「管線參數」、「連接」作業或其他步驟屬性來建立資訊更詳盡的錯誤訊息 (如果您使用的話) SDK。

步驟屬性

使用properties屬性在管線中的步驟之間新增資料相依性。配管使用這些資料相依性DAG從配管定義建構。這些屬性可以作為預留位置值參考，並在執行期解析。

Pipeline 步驟的properties屬性與Describe呼叫對應 SageMaker 工作類型所傳回的物件相符。對於每個任務類型，Describe 呼叫都會傳回下列回應物件：

若要在建立資料相依性期間檢查每個步驟類型可參照哪些屬性，請參閱 Amazon SageMaker Python SDK 中的資料相依性-屬性參考。

步進平行

當步驟不依賴於任何其他步驟時，它會在管線執行時立即執行。不過，平行執行過多管道步驟可能會快速耗盡可用的資源。使用 ParallelismConfiguration 控制管道執行的並行步驟數量。

下列範例使用 ParallelismConfiguration 將並行步驟限制為 5 個。

pipeline.create(
    parallelism_config=ParallelismConfiguration(5),
)

步驟間的資料相依性

您可以透過指定步驟之間的資料關係DAG來定義您的結構。若要在步驟之間建立資料相依性，請將一個步驟的屬性作為輸入傳遞至管道中的另一個步驟。在提供輸入的步驟執行完後，接收輸入的步驟才會啟動。

數據依賴使用以下格式的 JsonPath 表示法。此格式會遍歷JSON屬性檔案。這意味著你可以追加盡可能多的 <property> 依需要執行個體，以達到檔案中所需的巢狀性質。有關 JsonPath 符號的更多信息，請參閱JsonPath 回購。

<step_name>.properties.<property>.<property>

下面示範如何使用處理步驟的 ProcessingOutputConfig 屬性來指定 Amazon S3 儲存貯體。

step_process.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri

若要建立資料相依性，請將儲存貯體傳遞至訓練步驟，如下所示。

from sagemaker.workflow.pipeline_context import PipelineSession

sklearn_train = SKLearn(..., sagemaker_session=PipelineSession())

step_train = TrainingStep(
    name="CensusTrain",
    step_args=sklearn_train.fit(inputs=TrainingInput(
        s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
            "train_data"].S3Output.S3Uri
    ))
)

若要在建立資料相依性期間檢查每個步驟類型可參照哪些屬性，請參閱 Amazon SageMaker Python SDK 中的資料相依性-屬性參考。

步驟間的自訂相依性

當您指定資料相依性時，Pipeline 會提供步驟之間的資料連線。或者，一個步驟可以存取上一個步驟中的資料，而無需直接使用 Pipeline。在此情況下，您可以建立自訂相依性，告知 Pipelines 在另一個步驟完成執行之前不要啟動步驟。您可以透過指定步驟的 DependsOn 屬性來建立自訂相依性。

例如，下列內容定義了僅在 A 步驟和 B 步驟執行完後才開始的 C 步驟。

{
  'Steps': [
    {'Name':'A', ...},
    {'Name':'B', ...},
    {'Name':'C', 'DependsOn': ['A', 'B']}
  ]
}

管道拋出一個驗證異常，如果依賴將創建一個循環依賴關係。

下列範例會建立在處理步驟執行完後開始的訓練步驟。

processing_step = ProcessingStep(...)
training_step = TrainingStep(...)

training_step.add_depends_on([processing_step])

下列範例會建立在兩個不同的處理步驟執行完後才開始的訓練步驟。

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深度學習容器映像檔。

您也可以在管道步驟中使用您自己的容器。因為您無法從 Studio Classic 中建立映像檔，因此必須先使用其他方法建立映像，才能將其與管道搭配使用。

若要在為管線建立步驟時使用自己的容器，請在估算器定義URI中包含影像。如需搭配使用您自己容器的詳細資訊 SageMaker，請參閱搭配使用 Docker 容器 SageMaker。