定義管道 - Amazon SageMaker

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

定義管道

若要使用 Amazon SageMaker Pipelines 協調工作流程，您必須以JSON管道定義的形式產生定向非循環圖形 （DAG）。DAG 指定 ML 程序涉及的不同步驟，例如資料預先處理、模型訓練、模型評估和模型部署，以及這些步驟之間的相依性和資料流程。下列主題說明如何產生管道定義。

您可以使用 SageMaker Python SDK或 Amazon SageMaker Studio 中的視覺化 drag-and-drop管道設計工具功能來產生JSON管道定義。下圖是您在本教學課程中DAG建立管道的表示：

Studio 中管道的視覺化 drag-and-drop界面螢幕擷取畫面。

您在下列各節中定義的管道解決了迴歸問題，以根據鮑魚的實體測量來確定鮑魚的年齡。如需包含本教學課程中內容的可執行 Jupyter 筆記本，請參閱使用 Amazon SageMaker Model Building Pipelines 協調任務。

注意

您可以參考模型位置作為訓練步驟的屬性，如 Github 中 end-to-end的範例CustomerChurn 管道所示。

主題

    下列逐步解說會引導您使用管道設計工具建立準系統管道 drag-and-drop的步驟。如果您需要隨時暫停或結束視覺化設計工具中的管道編輯工作階段，請按一下匯出選項。這可讓您將管道的目前定義下載至本機環境。稍後，當您想要繼續管道編輯程序時，您可以將相同的JSON定義檔案匯入視覺化設計工具。

    建立處理步驟

    若要建立資料處理任務步驟，請執行下列動作：

    1. 按照 中的指示開啟 Studio 主控台推出 Amazon SageMaker 工作。

    2. 在左側導覽窗格中，選取管道 。

    3. 選擇 Create (建立)。

    4. 選擇空白 。

    5. 在左側側邊欄中，選擇處理資料並將其拖曳至畫布。

    6. 在畫布中，選擇您新增的流程資料步驟。

    7. 若要新增輸入資料集，請在右側列中選擇在資料 （輸入） 下新增，然後選擇資料集。

    8. 若要新增位置以儲存輸出資料集，請在右側列中選擇在資料 （輸出） 下新增，然後導覽至目的地。

    9. 完成右側列中的其餘欄位。如需這些索引標籤中欄位的相關資訊，請參閱 sagemaker.workflow.steps。ProcessingStep

    建立訓練步驟

    若要設定模型訓練步驟，請執行下列動作：

    1. 在左側側邊欄中，選擇訓練模型並將其拖曳至畫布。

    2. 在畫布中，選擇您新增的訓練模型步驟。

    3. 若要新增輸入資料集，請在右側列中選擇在資料 （輸入） 下新增，然後選擇資料集。

    4. 若要選擇儲存模型成品的位置，請在URI位置 （SURI3） 欄位中輸入 Amazon S3，或選擇瀏覽 S3 以導覽至目的地位置。

    5. 完成右側列中的其餘欄位。如需這些索引標籤中欄位的相關資訊，請參閱 sagemaker.workflow.steps。TrainingStep

    6. 按一下並將游標從您在上一節中新增的程序資料步驟拖曳至訓練模型步驟，以建立連接兩個步驟的邊緣。

    使用註冊模型步驟建立模型套件

    若要使用模型註冊步驟建立模型套件，請執行下列動作：

    1. 在左側側邊欄中，選擇註冊模型並將其拖曳至畫布。

    2. 在畫布中，選擇您新增的註冊模型步驟。

    3. 若要選取要註冊的模型，請選擇在模型下新增 （輸入）。

    4. 選擇建立模型群組，將模型新增至新的模型群組。

    5. 完成右側側邊欄中的其餘欄位。如需這些標籤中欄位的相關資訊，請參閱 sagemaker.workflow.step_collections。RegisterModel

    6. 按一下並將游標從您在上一節中新增的訓練模型步驟拖曳至註冊模型步驟，以建立連接兩個步驟的邊緣。

    使用部署模型 （端點） 步驟將模型部署至端點

    若要使用模型部署步驟部署模型，請執行下列動作：

    1. 在左側側邊欄中，選擇部署模型 （端點），並將其拖曳至畫布。

    2. 在畫布中，選擇您新增的部署模型 （端點） 步驟。

    3. 若要選擇要部署的模型，請選擇在模型 （輸入） 下新增。

    4. 選擇建立端點選項按鈕以建立新的端點。

    5. 輸入端點的名稱和描述。

    6. 按一下並將游標從您在上一節中新增的註冊模型步驟拖曳至部署模型 （端點） 步驟，以建立連接這兩個步驟的邊緣。

    7. 完成右側側邊欄中的其餘欄位。

    定義管道參數

    您可以設定一組管道參數，其值可以針對每次執行更新。若要定義管道參數並設定預設值，請按一下視覺化設計工具底部的齒輪圖示。

    儲存管道

    輸入建立管道所需的所有必要資訊後，按一下視覺化設計工具底部的儲存。這會驗證您的管道在執行階段是否有任何潛在錯誤，並通知您。在您解決自動驗證檢查標記的所有錯誤之前，儲存操作不會成功。如果您想要稍後繼續編輯，您可以將進行中管道儲存為本機環境中JSON的定義。您可以按一下視覺化設計工具底部的匯出按鈕，將管道匯出為JSON定義檔案。稍後，若要繼續更新管道，請按一下匯入按鈕上傳該JSON定義檔案。

    必要條件

    若要執行下列教學課程，請完成下列步驟：

    • 依照建立筆記本執行個體中所述的內容，設定筆記本執行個體。這可讓您的角色許可讀取和寫入 Amazon S3，並在 中建立訓練、批次轉換和處理任務 SageMaker。

    • 如修改角色許可政策所示，授予您的筆記本取得及傳遞其角色的權限。新增下列程式碼JSON片段，將此政策連接至您的角色。使用ARN用來建立筆記本執行個體的 取代<your-role-arn> 。

      {
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "iam:GetRole",
                "iam:PassRole"
            ],
            "Resource": "<your-role-arn>"
        }
    ]
}

    • 請依照修改角色信任政策 中的步驟來信任 SageMaker 服務主體。將下列陳述式片段新增至角色的信任關係：

      {
      "Sid": "",
      "Effect": "Allow",
      "Principal": {
        "Service": "sagemaker.amazonaws.com"
      },
      "Action": "sts:AssumeRole"
    }

    設定您的環境

    使用下列程式碼區塊建立新的 SageMaker 工作階段。這會傳回工作階段ARN的角色。此角色ARN應該是您設定為先決條件ARN的執行角色。

    import boto3
import sagemaker
import sagemaker.session
from sagemaker.workflow.pipeline_context import PipelineSession

region = boto3.Session().region_name
sagemaker_session = sagemaker.session.Session()
role = sagemaker.get_execution_role()
default_bucket = sagemaker_session.default_bucket()

pipeline_session = PipelineSession()

model_package_group_name = f"AbaloneModelPackageGroupName"

    建立管道

    重要

    允許 Amazon SageMaker Studio 或 Amazon SageMaker Studio Classic 建立 Amazon SageMaker 資源的自訂IAM政策也必須授予許可，才能將標籤新增至這些資源。需要將標籤新增至資源的許可，因為 Studio 和 Studio Classic 會自動標記他們建立的任何資源。如果IAM政策允許 Studio 和 Studio Classic 建立資源，但不允許標記，則嘗試建立資源時可能會發生「AccessDenied」錯誤。如需詳細資訊，請參閱提供標記 SageMaker資源的許可。

    AWS Amazon 的受管政策 SageMaker 提供建立 SageMaker 資源的許可，已包含在建立這些資源時新增標籤的許可。

    從 SageMaker 筆記本執行個體執行下列步驟，以建立包含步驟的管道：

    • 預處理

    • 訓練

    • 評估

    • 條件式評估

    • 模型註冊

    注意

    您可以使用 ExecutionVariables和 聯結函數來指定輸出位置。 ExecutionVariables 會在執行階段解析。例如，ExecutionVariables.PIPELINE_EXECUTION_ID 解析目前執行的 ID，該 ID 可以用來做為不同執行間的唯一識別碼。

    步驟 1：下載資料集

    此筆記本使用 UCI Machine Learning Abalone 資料集。此資料集含下列功能：

    • length – 測量的最長鮑魚外殼尺寸。

    • diameter – 與長度垂直的鮑魚直徑。

    • height – 鮑魚殼中鮑魚肉的高度。

    • whole_weight – 整個鮑魚的重量。

    • shucked_weight – 從鮑魚中取出的肉的重量。

    • viscera_weight – 鮑魚內臟經過出血處理後的重量。

    • shell_weight – 除肉和乾燥後鮑魚殼的重量。

    • sex – 鮑魚的性別。'M'、'F' 或 'I' 之一，其中 'I' 表示幼鮑。

    • rings – 鮑魚殼上的環數。

    使用公式 age=rings + 1.5，可以根據鮑魚殼上的環數得到鮑魚年齡的良好近似值。不過，取得此數字是一項耗時的任務。您必須沿著螺錐切割貝殼，染色切片，然後在顯微鏡下計算環數。不過，其他實體測量更容易取得。這個筆記本使用資料集，透過其他物理量測建立一個預測環數的模型。

    下載資料集

    1. 將資料集下載到帳戶的預設 Amazon S3 儲存貯體中。

      !mkdir -p data
local_path = "data/abalone-dataset.csv"

s3 = boto3.resource("s3")
s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file(
    "dataset/abalone-dataset.csv",
    local_path
)

base_uri = f"s3://{default_bucket}/abalone"
input_data_uri = sagemaker.s3.S3Uploader.upload(
    local_path=local_path, 
    desired_s3_uri=base_uri,
)
print(input_data_uri)

    2. 建立模型後，下載第二個資料集以進行批次轉換。

      local_path = "data/abalone-dataset-batch.csv"

s3 = boto3.resource("s3")
s3.Bucket(f"sagemaker-servicecatalog-seedcode-{region}").download_file(
    "dataset/abalone-dataset-batch",
    local_path
)

base_uri = f"s3://{default_bucket}/abalone"
batch_data_uri = sagemaker.s3.S3Uploader.upload(
    local_path=local_path, 
    desired_s3_uri=base_uri,
)
print(batch_data_uri)

    步驟 2：定義管道參數

    此程式碼區塊會為您的管道定義下列參數：

    • processing_instance_count – 處理任務的執行個體計數。

    • input_data – 輸入資料的 Amazon S3 位置。

    • batch_data – 用於批次轉換之輸入資料的 Amazon S3 位置。

    • model_approval_status – 將已訓練模型進行 CI/CD 註冊的核准狀態。如需更多資訊，請參閱MLOps 使用 SageMaker 專案自動化。

    from sagemaker.workflow.parameters import (
    ParameterInteger,
    ParameterString,
)

processing_instance_count = ParameterInteger(
    name="ProcessingInstanceCount",
    default_value=1
)
model_approval_status = ParameterString(
    name="ModelApprovalStatus",
    default_value="PendingManualApproval"
)
input_data = ParameterString(
    name="InputData",
    default_value=input_data_uri,
)
batch_data = ParameterString(
    name="BatchData",
    default_value=batch_data_uri,
)

    步驟 3：定義特徵工程的處理步驟

    本節展示如何建立處理步驟，以透過資料集準備用於進行訓練的資料。

    建立處理步驟

    1. 建立處理命令碼的目錄。

      !mkdir -p abalone

    2. 在 /abalone 目錄中，建立名為 preprocessing.py 的檔案，內含下列內容。此預先處理指令碼會傳入處理步驟，以便在輸入資料上執行。然後，訓練步驟會使用預先處理的訓練功能和標籤來訓練模型。評估步驟使用訓練過的模型和預先處理的測試功能和標籤來評估模型。此指令碼使用 scikit-learn 執行下列動作：

      • 填寫缺少的 sex 分類資料並進行編碼，以適合訓練之用。

      • 對除rings 和 sex 之外的所有數字欄位執行縮放和標準化處理。

      • 將資料分割為訓練、測試和驗證資料集。

      %%writefile abalone/preprocessing.py
import argparse
import os
import requests
import tempfile
import numpy as np
import pandas as pd


from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder


# Because this is a headerless CSV file, specify the column names here.
feature_columns_names = [
    "sex",
    "length",
    "diameter",
    "height",
    "whole_weight",
    "shucked_weight",
    "viscera_weight",
    "shell_weight",
]
label_column = "rings"

feature_columns_dtype = {
    "sex": str,
    "length": np.float64,
    "diameter": np.float64,
    "height": np.float64,
    "whole_weight": np.float64,
    "shucked_weight": np.float64,
    "viscera_weight": np.float64,
    "shell_weight": np.float64
}
label_column_dtype = {"rings": np.float64}


def merge_two_dicts(x, y):
    z = x.copy()
    z.update(y)
    return z


if __name__ == "__main__":
    base_dir = "/opt/ml/processing"

    df = pd.read_csv(
        f"{base_dir}/input/abalone-dataset.csv",
        header=None, 
        names=feature_columns_names + [label_column],
        dtype=merge_two_dicts(feature_columns_dtype, label_column_dtype)
    )
    numeric_features = list(feature_columns_names)
    numeric_features.remove("sex")
    numeric_transformer = Pipeline(
        steps=[
            ("imputer", SimpleImputer(strategy="median")),
            ("scaler", StandardScaler())
        ]
    )

    categorical_features = ["sex"]
    categorical_transformer = Pipeline(
        steps=[
            ("imputer", SimpleImputer(strategy="constant", fill_value="missing")),
            ("onehot", OneHotEncoder(handle_unknown="ignore"))
        ]
    )

    preprocess = ColumnTransformer(
        transformers=[
            ("num", numeric_transformer, numeric_features),
            ("cat", categorical_transformer, categorical_features)
        ]
    )
    
    y = df.pop("rings")
    X_pre = preprocess.fit_transform(df)
    y_pre = y.to_numpy().reshape(len(y), 1)
    
    X = np.concatenate((y_pre, X_pre), axis=1)
    
    np.random.shuffle(X)
    train, validation, test = np.split(X, [int(.7*len(X)), int(.85*len(X))])

    
    pd.DataFrame(train).to_csv(f"{base_dir}/train/train.csv", header=False, index=False)
    pd.DataFrame(validation).to_csv(f"{base_dir}/validation/validation.csv", header=False, index=False)
    pd.DataFrame(test).to_csv(f"{base_dir}/test/test.csv", header=False, index=False)

    3. 為 SKLearnProcessor 建立要傳入處理步驟的執行個體。

      from sagemaker.sklearn.processing import SKLearnProcessor


framework_version = "0.23-1"

sklearn_processor = SKLearnProcessor(
    framework_version=framework_version,
    instance_type="ml.m5.xlarge",
    instance_count=processing_instance_count,
    base_job_name="sklearn-abalone-process",
    sagemaker_session=pipeline_session,
    role=role,
)

    4. 建立處理步驟。此步驟會接受 SKLearnProcessor、輸入和輸出通道，以及您建立的 preprocessing.py 指令碼。這與 SageMaker Python 中的處理器執行個體run方法非常類似SDK。傳入 ProcessingStep 的 input_data 參數是步驟本身的輸入資料。此輸入資料會在處理器執行個體執行時使用。

      請注意在處理任務的輸出組態中指定的 "train、"validation 和 "test" 具名通道。這類步驟Properties可用於後續步驟，並在執行時間解析為其執行時間值。

      from sagemaker.processing import ProcessingInput, ProcessingOutput
from sagemaker.workflow.steps import ProcessingStep
   

processor_args = sklearn_processor.run(
    inputs=[
      ProcessingInput(source=input_data, destination="/opt/ml/processing/input"),  
    ],
    outputs=[
        ProcessingOutput(output_name="train", source="/opt/ml/processing/train"),
        ProcessingOutput(output_name="validation", source="/opt/ml/processing/validation"),
        ProcessingOutput(output_name="test", source="/opt/ml/processing/test")
    ],
    code="abalone/preprocessing.py",
) 

step_process = ProcessingStep(
    name="AbaloneProcess",
    step_args=processor_args
)

    步驟 4：定義訓練步驟

    本節說明如何使用 SageMaker XGBoost演算法，針對處理步驟的訓練資料輸出來訓練模型。

    定義訓練步驟

    1. 指定儲存訓練模型的模型路徑。

      model_path = f"s3://{default_bucket}/AbaloneTrain"

    2. 設定XGBoost演算法和輸入資料集的估算器。訓練執行個體類型會傳遞至此估算器。典型的訓練指令碼：

      • 從輸入通道載入資料

      • 使用超參數設定訓練

      • 訓練模型

      • 將模型model_dir儲存至 ，以便稍後託管

      SageMaker 在訓練任務結束時，以 model.tar.gz的形式將模型上傳至 Amazon S3。

      from sagemaker.estimator import Estimator


image_uri = sagemaker.image_uris.retrieve(
    framework="xgboost",
    region=region,
    version="1.0-1",
    py_version="py3",
    instance_type="ml.m5.xlarge"
)
xgb_train = Estimator(
    image_uri=image_uri,
    instance_type="ml.m5.xlarge",
    instance_count=1,
    output_path=model_path,
    sagemaker_session=pipeline_session,
    role=role,
)
xgb_train.set_hyperparameters(
    objective="reg:linear",
    num_round=50,
    max_depth=5,
    eta=0.2,
    gamma=4,
    min_child_weight=6,
    subsample=0.7,
    silent=0
)

    3. 使用 的估算器執行個體和屬性建立TrainingStep ProcessingStep。在 "train"和"validation"輸出通道S3Uri的 中將 傳遞至 TrainingStep。  

      from sagemaker.inputs import TrainingInput
from sagemaker.workflow.steps import TrainingStep


train_args = xgb_train.fit(
    inputs={
        "train": TrainingInput(
            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                "train"
            ].S3Output.S3Uri,
            content_type="text/csv"
        ),
        "validation": TrainingInput(
            s3_data=step_process.properties.ProcessingOutputConfig.Outputs[
                "validation"
            ].S3Output.S3Uri,
            content_type="text/csv"
        )
    },
)

step_train = TrainingStep(
    name="AbaloneTrain",
    step_args = train_args
)

    步驟 5：定義模型評估的處理步驟

    本節將介紹如何建立處理步驟來評估模型的準確性。此模型評估的結果會用於條件步驟，以決定要採取的執行路徑。

    定義模型評估的處理步驟

    1. 在 /abalone 目錄中建立名為 evaluation.py 的檔案。此指令碼用於處理步驟，以執行模型評估。它需要經過訓練的模型和測試資料集作為輸入，然後產生包含分類評估指標JSON的檔案。

      %%writefile abalone/evaluation.py
import json
import pathlib
import pickle
import tarfile
import joblib
import numpy as np
import pandas as pd
import xgboost


from sklearn.metrics import mean_squared_error


if __name__ == "__main__":
    model_path = f"/opt/ml/processing/model/model.tar.gz"
    with tarfile.open(model_path) as tar:
        tar.extractall(path=".")
    
    model = pickle.load(open("xgboost-model", "rb"))

    test_path = "/opt/ml/processing/test/test.csv"
    df = pd.read_csv(test_path, header=None)
    
    y_test = df.iloc[:, 0].to_numpy()
    df.drop(df.columns[0], axis=1, inplace=True)
    
    X_test = xgboost.DMatrix(df.values)
    
    predictions = model.predict(X_test)

    mse = mean_squared_error(y_test, predictions)
    std = np.std(y_test - predictions)
    report_dict = {
        "regression_metrics": {
            "mse": {
                "value": mse,
                "standard_deviation": std
            },
        },
    }

    output_dir = "/opt/ml/processing/evaluation"
    pathlib.Path(output_dir).mkdir(parents=True, exist_ok=True)
    
    evaluation_path = f"{output_dir}/evaluation.json"
    with open(evaluation_path, "w") as f:
        f.write(json.dumps(report_dict))

    2. 建立一個 ScriptProcessor 執行個體，用來建立 ProcessingStep。

      from sagemaker.processing import ScriptProcessor


script_eval = ScriptProcessor(
    image_uri=image_uri,
    command=["python3"],
    instance_type="ml.m5.xlarge",
    instance_count=1,
    base_job_name="script-abalone-eval",
    sagemaker_session=pipeline_session,
    role=role,
)

    3. 使用處理器執行個體、輸入和輸出通道和evaluation.py指令碼建立 ProcessingStep。傳入：

      • step_train 訓練步驟中的 S3ModelArtifacts 屬性

      • step_process 處理步驟S3Uri"test"輸出通道的

      這與 SageMaker Python 中的處理器執行個體run方法非常類似SDK。  

      from sagemaker.workflow.properties import PropertyFile


evaluation_report = PropertyFile(
    name="EvaluationReport",
    output_name="evaluation",
    path="evaluation.json"
)

eval_args = script_eval.run(
        inputs=[
        ProcessingInput(
            source=step_train.properties.ModelArtifacts.S3ModelArtifacts,
            destination="/opt/ml/processing/model"
        ),
        ProcessingInput(
            source=step_process.properties.ProcessingOutputConfig.Outputs[
                "test"
            ].S3Output.S3Uri,
            destination="/opt/ml/processing/test"
        )
    ],
    outputs=[
        ProcessingOutput(output_name="evaluation", source="/opt/ml/processing/evaluation"),
    ],
    code="abalone/evaluation.py",
)

step_eval = ProcessingStep(
    name="AbaloneEval",
    step_args=eval_args,
    property_files=[evaluation_report],
)

    步驟 6：定義批次轉換 CreateModelStep 的

    重要

    建議使用 模型步驟 建立 SageMaker Python 2.90.0 版以後的模型SDK。 CreateModelStep將繼續在 SageMaker Python 的先前版本中運作SDK，但不再主動支援。

    本節說明如何從訓練步驟的輸出建立 SageMaker 模型。此模型用於根據新資料集進行批次轉換。此步驟會傳遞至條件步驟，而且只有在條件步驟評估為 時才會執行true。

    定義批次轉換 CreateModelStep 的

    1. 建立 SageMaker 模型。從 step_train 訓練步驟傳入 S3ModelArtifacts 屬性。

      from sagemaker.model import Model


model = Model(
    image_uri=image_uri,
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=pipeline_session,
    role=role,
)

    2. 定義模型的 SageMaker 模型輸入。

      from sagemaker.inputs import CreateModelInput


inputs = CreateModelInput(
    instance_type="ml.m5.large",
    accelerator_type="ml.eia1.medium",
)

    3. CreateModelStep 使用您定義的 CreateModelInput和 SageMaker 模型執行個體建立您的 。

      from sagemaker.workflow.steps import CreateModelStep


step_create_model = CreateModelStep(
    name="AbaloneCreateModel",
    model=model,
    inputs=inputs,
)

    步驟 7：定義 TransformStep 以執行批次轉換

    本節展示如何在模型訓練後建立 TransformStep，以根據資料集執行批次轉換。此步驟會傳遞至條件步驟，而且只有在條件步驟評估為 時才會執行true。

    定義 TransformStep 以執行批次轉換

    1. 建立具有適當運算執行個體類型、執行個體計數和所需輸出 Amazon S3 儲存貯體的轉換器執行個體URI。從 step_create_model CreateModel 步驟傳入 ModelName 屬性。

      from sagemaker.transformer import Transformer


transformer = Transformer(
    model_name=step_create_model.properties.ModelName,
    instance_type="ml.m5.xlarge",
    instance_count=1,
    output_path=f"s3://{default_bucket}/AbaloneTransform"
)

    2. 使用您定義的轉換器執行個體和 batch_data 管道參數建立 TransformStep。

      from sagemaker.inputs import TransformInput
from sagemaker.workflow.steps import TransformStep


step_transform = TransformStep(
    name="AbaloneTransform",
    transformer=transformer,
    inputs=TransformInput(data=batch_data)
)

    步驟 8：定義建立模型套件 RegisterModel 的步驟

    重要

    建議使用 模型步驟 註冊 SageMaker Python 2.90.0 版以後的模型SDK。 RegisterModel將繼續在 SageMaker Python 的先前版本中運作SDK，但不再主動支援。

    本節說明如何建立 執行個體RegisterModel。在管道RegisterModel中執行的結果是模型套件。模型套件是可重複使用的模型成品抽象，可封裝推論所需的所有元件。它由推論規格以及可選模型加權位置組成，推論規格會定義要使用的推論映像。模型套件群組是模型套件的集合。您可以使用 ModelPackageGroup for Pipelines 將新版本和模型套件新增至群組，以便每次執行管道。若要取得有關模型註冊表的更多相關資訊，請參閱使用模型登錄檔進行模型註冊部署。

    此步驟會傳遞至條件步驟，而且只有在條件步驟評估為 時才會執行true。

    定義建立模型套件 RegisterModel 的步驟

    • 透過用於訓練步驟的估算器執行個體來建構 RegisterModel 步驟。從 step_train 訓練步驟傳入 S3ModelArtifacts 屬性並指定 ModelPackageGroup。管道ModelPackageGroup會為您建立此項目。

      from sagemaker.model_metrics import MetricsSource, ModelMetrics 
from sagemaker.workflow.step_collections import RegisterModel


model_metrics = ModelMetrics(
    model_statistics=MetricsSource(
        s3_uri="{}/evaluation.json".format(
            step_eval.arguments["ProcessingOutputConfig"]["Outputs"][0]["S3Output"]["S3Uri"]
        ),
        content_type="application/json"
    )
)
step_register = RegisterModel(
    name="AbaloneRegisterModel",
    estimator=xgb_train,
    model_data=step_train.properties.ModelArtifacts.S3ModelArtifacts,
    content_types=["text/csv"],
    response_types=["text/csv"],
    inference_instances=["ml.t2.medium", "ml.m5.xlarge"],
    transform_instances=["ml.m5.xlarge"],
    model_package_group_name=model_package_group_name,
    approval_status=model_approval_status,
    model_metrics=model_metrics
)

    步驟 9：定義條件步驟以驗證模型準確性

    ConditionStep 可讓管道DAG根據步驟屬性的條件，支援在管道中執行的條件式執行。在此情況下，只有在模型的準確性超過所需值時，您才想要註冊模型套件。模型的準確度由模型評估步驟決定。如果準確度超過所需的值，管道也會建立 SageMaker 模型，並在資料集上執行批次轉換。本節展示如何定義條件步驟。

    定義條件步驟以驗證模型準確性

    1. 使用模型評估計算處理步驟 step_eval 之輸出中的準確性值來定義 ConditionLessThanOrEqualTo 條件。使用您在處理步驟中編製索引的屬性檔案，以及平均平方錯誤值 JSONPath 的個別 取得此輸出"mse"。

      from sagemaker.workflow.conditions import ConditionLessThanOrEqualTo
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.functions import JsonGet


cond_lte = ConditionLessThanOrEqualTo(
    left=JsonGet(
        step_name=step_eval.name,
        property_file=evaluation_report,
        json_path="regression_metrics.mse.value"
    ),
    right=6.0
)

    2. 建構 ConditionStep。傳入 ConditionEquals 條件，然後將模型套件註冊和批次轉換步驟設定為條件通過時執行的後續步驟。

      step_cond = ConditionStep(
    name="AbaloneMSECond",
    conditions=[cond_lte],
    if_steps=[step_register, step_create_model, step_transform],
    else_steps=[], 
)

    第 10 步：建立管道

    您現在已建立了所有步驟，接下來將它們合併到一個管道中。

    建立管道

    1. 為管道定義下列內容：name、parameters、和 steps。(account, region) 對內的名稱必須是唯一的。

      注意

      一個步驟只能在管道的步驟清單或條件步驟的 if/else 步驟清單中出現一次。不能同時在這兩個清單中出現。

      from sagemaker.workflow.pipeline import Pipeline


pipeline_name = f"AbalonePipeline"
pipeline = Pipeline(
    name=pipeline_name,
    parameters=[
        processing_instance_count,
        model_approval_status,
        input_data,
        batch_data,
    ],
    steps=[step_process, step_train, step_eval, step_cond],
)

    2. （選用） 檢查JSON管道定義，以確保其格式良好。

      import json

json.loads(pipeline.definition())

    此管道定義已準備好提交給 SageMaker。在下一個教學課程中，您將此管道提交至 SageMaker 並開始執行。

    您也可以使用 boto3 或 AWS CloudFormation建立管道。建立管道需要管道定義，這是一個定義管道每個步驟的JSON物件。 SageMaker SDK 提供建構管道定義的簡單方法，您可以搭配APIs上述任何一種方法來建立管道本身。不使用 SDK，使用者必須撰寫原始JSON定義，才能建立管道，而不需要 SageMaker Python 提供的任何錯誤檢查SDK。若要查看管道JSON定義的結構描述，請參閱 SageMaker 管道定義JSON結構描述。下列程式碼範例顯示 SageMaker 管道定義JSON物件的範例：

    {'Version': '2020-12-01',
 'Metadata': {},
 'Parameters': [{'Name': 'ProcessingInstanceType',
   'Type': 'String',
   'DefaultValue': 'ml.m5.xlarge'},
  {'Name': 'ProcessingInstanceCount', 'Type': 'Integer', 'DefaultValue': 1},
  {'Name': 'TrainingInstanceType',
   'Type': 'String',
   'DefaultValue': 'ml.m5.xlarge'},
  {'Name': 'ModelApprovalStatus',
   'Type': 'String',
   'DefaultValue': 'PendingManualApproval'},
  {'Name': 'ProcessedData',
   'Type': 'String',
   'DefaultValue': 'S3_URL',
{'Name': 'InputDataUrl',
   'Type': 'String',
   'DefaultValue': 'S3_URL',
 'PipelineExperimentConfig': {'ExperimentName': {'Get': 'Execution.PipelineName'},
  'TrialName': {'Get': 'Execution.PipelineExecutionId'}},
 'Steps': [{'Name': 'ReadTrainDataFromFS',
   'Type': 'Processing',
   'Arguments': {'ProcessingResources': {'ClusterConfig': {'InstanceType': 'ml.m5.4xlarge',
      'InstanceCount': 2,
      'VolumeSizeInGB': 30}},
    'AppSpecification': {'ImageUri': 'IMAGE_URI',
     'ContainerArguments': [....]},
    'RoleArn': 'ROLE',
      'ProcessingInputs': [...],
    'ProcessingOutputConfig': {'Outputs': [.....]},
    'StoppingCondition': {'MaxRuntimeInSeconds': 86400}},
   'CacheConfig': {'Enabled': True, 'ExpireAfter': '30d'}},
   ...
   ...
   ...
  }

    後續步驟：執行管道