自动化机器学习工作流

教程

概览

在此教程中,了解如何使用 Amazon SageMaker Pipelines、Amazon SageMaker 模型注册表和 Amazon SageMaker Clarify 创建和自动化端到端机器学习(ML)工作流。

SageMaker Pipelines 是首个为 ML 专门构建的持续集成和持续交付(CI/CD)服务。使用 SageMaker Pipelines,您可以自动化 ML 工作流的不同步骤,包括数据加载、数据转换、训练、优化、评估以及部署。通过 SageMaker 模型注册表,您可以在一个中央存储库中跟踪模型版本、它们的元数据,比如使用案例分组,以及模型性能指标基准,在这个中央存储库中很容易根据您的业务需求选择适当的部署模型。SageMaker Clarify 可让您更清楚地了解您的训练数据和模型,以便您可以识别和限制偏差,并解释预测。

在此教程中,您将实施 SageMaker 管道以构建、训练和部署 XGBoost 二进制分类,以预测汽车保险索赔被欺诈的可能性。您将使用合成的汽车保险索赔数据集。原始输入是两个保险数据表:一个索赔表和一个客户表。索赔表有一个名为 fraud 的列,表明索赔是欺诈性的还是其他形式的。您管道将处理原始数据;创建训练、验证和测试数据集;建立并评价二进制分类模型。然后,它将使用 SageMaker Clarify 来测试模型偏差和可解释性,并最后部署模型进行推断。

您将学到的内容

在本指南中,您将:

  • 构建并运行 SageMaker 管道以自动化端到端 ML 生命周期
  • 使用部署的模型生成预测

先决条件

在开始学习本指南之前,您需要:

  • 一个 AWS 账户:如果您还没有账户,请遵循设置 AWS 环境入门指南中的说明获取快速概览。

 AWS 使用经验

中级

 完成时间

120 分钟

 所需费用

请参阅 SageMaker 定价估算此教程的费用。

 需要

您必须登录 AWS 账户。

 使用的服务

Amazon SageMaker Studio、Amazon SageMaker Pipelines、Amazon SageMaker Clarify、Amazon SageMaker 模型注册表

 上次更新日期

2022 年 6 月 24 日

实施

第 1 步:设置 Amazon SageMaker Studio 域

一个 AWS 账户在一个区域只能有一个 SageMaker Studio 域。如果您在美国东部(弗吉尼亚州北部)区域已经有一个 SageMaker Studio 域,请遵照 ML 工作流的 SageMaker Studio 设置指南将所需的 AWS IAM 策略附加到您的 SageMaker Studio 账户,然后跳过步骤 1,并直接继续步骤 2。

如果您没有现有的 SageMaker Studio 域,第 1 步以运行 AWS CloudFormation 模板,从而创建 SageMaker Studio 域并添加本教程剩余部分所需的权限。

选择 AWS CloudFormation 堆栈链接。此链接将打开 AWS CloudFormation 控制台并创建您的 SageMaker Studio 域和名为 studio-user 的用户。它还将添加所需权限到您的 SageMaker Studio 账户。在 CloudFormation 控制台中,确认美国东部(弗吉尼亚州北部) 是右上角显示的区域。 堆栈名称应为 CFN-SM-IM-Lambda-catalog,且不应更改。 此堆栈需要花费 10 分钟左右才能创建所有资源。

此堆栈假设您已经在账户中设置了一个公有 VPC。如果您没有公有 VPC,请参阅具有单个公有子网的 VPC以了解如何创建公有 VPC。 

选择 I acknowledge that AWS CloudFormation might create IAM resources(我确认,AWS CloudFormation 可能创建 IAM 资源),然后选择 Create stack(创建堆栈)。

在 CloudFormation 窗格上,选择 Stacks(堆栈)。堆栈约需要 10 分钟才能完成创建。创建堆栈时,堆栈状态从 CREATE_IN_PROGRESS 变为 CREATE_COMPLETE。 

步骤 2:设置 SageMaker Studio 笔记本并参数化管道

在此设置中,您将启动一个新的 SageMaker Studio 笔记本,并配置与 Amazon Simple Storage Service(Amazon S3)交互所需的 SageMaker 变量。

在 AWS 管理控制台搜索栏中输入 SageMaker Studio,然后选择 SageMaker Studio。 从控制台右上角的 Region(区域)下拉列表中选择美国东部(弗吉尼亚州北部)

对于启动应用程序,选择 Studio 以使用 studio-user 配置文件打开 SageMaker Studio。

在 SageMaker Studio 导航栏上,选择 File(文件)、New(新建)、Notebook(笔记本)。 

Set up notebook environment(设置笔记本环境)对话框中的 Image(图像)下,选择 Data Science(数据科学)。将自动选择 Python 3 内核。选择 Select(选择)。

笔记本右上角上的内核现在应显示 Python 3 (Data Science)(Python 3(数据科学))。

要导入所需的库,将以下代码复制并粘贴到笔记本中的单元格中并运行单元格。

import pandas as pd
import json
import boto3
import pathlib
import io
import sagemaker


from sagemaker.deserializers import CSVDeserializer
from sagemaker.serializers import CSVSerializer

from sagemaker.xgboost.estimator import XGBoost
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.processing import (
    ProcessingInput, 
    ProcessingOutput, 
    ScriptProcessor
)
from sagemaker.inputs import TrainingInput

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import (
    ProcessingStep, 
    TrainingStep, 
    CreateModelStep
)
from sagemaker.workflow.check_job_config import CheckJobConfig
from sagemaker.workflow.parameters import (
    ParameterInteger, 
    ParameterFloat, 
    ParameterString, 
    ParameterBoolean
)
from sagemaker.workflow.clarify_check_step import (
    ModelBiasCheckConfig, 
    ClarifyCheckStep, 
    ModelExplainabilityCheckConfig
)
from sagemaker.workflow.step_collections import RegisterModel
from sagemaker.workflow.conditions import ConditionGreaterThanOrEqualTo
from sagemaker.workflow.properties import PropertyFile
from sagemaker.workflow.condition_step import ConditionStep
from sagemaker.workflow.functions import JsonGet

from sagemaker.workflow.lambda_step import (
    LambdaStep,
    LambdaOutput,
    LambdaOutputTypeEnum,
)
from sagemaker.lambda_helper import Lambda

from sagemaker.model_metrics import (
    MetricsSource, 
    ModelMetrics, 
    FileSource
)
from sagemaker.drift_check_baselines import DriftCheckBaselines

from sagemaker.image_uris import retrieve

将以下代码块复制并粘贴到单元格中,并进行运行,以使用 SageMaker 和 AWS 开发工具包设置 SageMaker 和 S3 客户端对象。SageMaker 需要这些对象来执行各种操作,比如部署和调用端点,以及与 Amazon S3 和 AWS Lambda 交互。该代码还设置了存储原始数据集和处理过的数据集和模型构件的 S3 存储桶位置。请注意,读取和写入存储桶是独立的。读取存储桶是名为 sagemaker-sample-files 的公共 S3 存储桶,其中包含原始数据集。 写入存储桶是与您名为 sagemaker-<your- Region>-<your-account-id> 的账户关联的默认 S3 存储桶,本教程稍后部分将使用它来存储处理过的数据集和构件。

# Instantiate AWS services session and client objects
sess = sagemaker.Session()
write_bucket = sess.default_bucket()
write_prefix = "fraud-detect-demo"

region = sess.boto_region_name
s3_client = boto3.client("s3", region_name=region)
sm_client = boto3.client("sagemaker", region_name=region)
sm_runtime_client = boto3.client("sagemaker-runtime")

# Fetch SageMaker execution role
sagemaker_role = sagemaker.get_execution_role()


# S3 locations used for parameterizing the notebook run
read_bucket = "sagemaker-sample-files"
read_prefix = "datasets/tabular/synthetic_automobile_claims" 

# S3 location where raw data to be fetched from
raw_data_key = f"s3://{read_bucket}/{read_prefix}"

# S3 location where processed data to be uploaded
processed_data_key = f"{write_prefix}/processed"

# S3 location where train data to be uploaded
train_data_key = f"{write_prefix}/train"

# S3 location where validation data to be uploaded
validation_data_key = f"{write_prefix}/validation"

# S3 location where test data to be uploaded
test_data_key = f"{write_prefix}/test"


# Full S3 paths
claims_data_uri = f"{raw_data_key}/claims.csv"
customers_data_uri = f"{raw_data_key}/customers.csv"
output_data_uri = f"s3://{write_bucket}/{write_prefix}/"
scripts_uri = f"s3://{write_bucket}/{write_prefix}/scripts"
estimator_output_uri = f"s3://{write_bucket}/{write_prefix}/training_jobs"
processing_output_uri = f"s3://{write_bucket}/{write_prefix}/processing_jobs"
model_eval_output_uri = f"s3://{write_bucket}/{write_prefix}/model_eval"
clarify_bias_config_output_uri = f"s3://{write_bucket}/{write_prefix}/model_monitor/bias_config"
clarify_explainability_config_output_uri = f"s3://{write_bucket}/{write_prefix}/model_monitor/explainability_config"
bias_report_output_uri = f"s3://{write_bucket}/{write_prefix}/clarify_output/pipeline/bias"
explainability_report_output_uri = f"s3://{write_bucket}/{write_prefix}/clarify_output/pipeline/explainability"

# Retrieve training image
training_image = retrieve(framework="xgboost", region=region, version="1.3-1")

复制并粘贴以下代码来设置各种 SageMaker 管道组件的名称,例如模型和端点,并指定训练和推断实例类型和计数。这些值将用于参数化您的管道。

# Set names of pipeline objects
pipeline_name = "FraudDetectXGBPipeline"
pipeline_model_name = "fraud-detect-xgb-pipeline"
model_package_group_name = "fraud-detect-xgb-model-group"
base_job_name_prefix = "fraud-detect"
endpoint_config_name = f"{pipeline_model_name}-endpoint-config"
endpoint_name = f"{pipeline_model_name}-endpoint"

# Set data parameters
target_col = "fraud"

# Set instance types and counts
process_instance_type = "ml.c5.xlarge"
train_instance_count = 1
train_instance_type = "ml.m4.xlarge"
predictor_instance_count = 1
predictor_instance_type = "ml.m4.xlarge"
clarify_instance_count = 1
clarify_instance_type = "ml.m4.xlarge"

SageMaker Pipelines 支持参数化,允许您在运行时指定输入参数,而无需更改管道代码。您可以使用 sagemaker.workflow.parameters 模块下提供的模块(例如 ParameterIntegerParameterFloatParameterStringParameterBoolean)来指定不同数据类型的管道参数。复制、粘贴并运行以下代码来设置多个输入参数,包括 SageMaker Clarify 配置。

# Set up pipeline input parameters

# Set processing instance type
process_instance_type_param = ParameterString(
    name="ProcessingInstanceType",
    default_value=process_instance_type,
)

# Set training instance type
train_instance_type_param = ParameterString(
    name="TrainingInstanceType",
    default_value=train_instance_type,
)

# Set training instance count
train_instance_count_param = ParameterInteger(
    name="TrainingInstanceCount",
    default_value=train_instance_count
)

# Set deployment instance type
deploy_instance_type_param = ParameterString(
    name="DeployInstanceType",
    default_value=predictor_instance_type,
)

# Set deployment instance count
deploy_instance_count_param = ParameterInteger(
    name="DeployInstanceCount",
    default_value=predictor_instance_count
)

# Set Clarify check instance type
clarify_instance_type_param = ParameterString(
    name="ClarifyInstanceType",
    default_value=clarify_instance_type,
)

# Set model bias check params
skip_check_model_bias_param = ParameterBoolean(
    name="SkipModelBiasCheck", 
    default_value=False
)

register_new_baseline_model_bias_param = ParameterBoolean(
    name="RegisterNewModelBiasBaseline",
    default_value=False
)

supplied_baseline_constraints_model_bias_param = ParameterString(
    name="ModelBiasSuppliedBaselineConstraints", 
    default_value=""
)

# Set model explainability check params
skip_check_model_explainability_param = ParameterBoolean(
    name="SkipModelExplainabilityCheck", 
    default_value=False
)

register_new_baseline_model_explainability_param = ParameterBoolean(
    name="RegisterNewModelExplainabilityBaseline",
    default_value=False
)

supplied_baseline_constraints_model_explainability_param = ParameterString(
    name="ModelExplainabilitySuppliedBaselineConstraints", 
    default_value=""
)

# Set model approval param
model_approval_status_param = ParameterString(
    name="ModelApprovalStatus", default_value="Approved"
)

步骤 3:构建管道组件

管道是一系列步骤,这些步骤可以单独构建,然后放在一起形成 ML 工作流。 下图显示了管道的高级别步骤。

在本教程中,您将使用以下步骤构建管道:

  1. 数据处理步骤:使用 S3 中的输入原始数据运行 SageMaker 处理任务,并将训练、验证和测试分割输出到 S3。
  2. 训练步骤:将 S3 中的训练和验证数据作为输入,使用 SageMaker 训练作业训练 XGBoost 模型,并将训练好的模型构件存储在 S3 中。
  3. 评估步骤:通过使用 S3 中的测试数据和模型构件作为输入,运行 SageMaker 处理任务来评估测试数据集上的模型,并将输出模型性能评估报告存储在 S3 中。
  4. 条件步骤:将测试数据集上的模型性能与阈值相比较。使用 S3 中的模型性能评估报告作为输入来运行 SageMaker Pipelines 预定义步骤,并存储将在模型性能可接受时执行的管道步骤的输出列表。
  5. 创建模型步骤:使用 S3 中的模型构件作为输入运行 SageMaker Pipelines 预定义步骤,并将输出 SageMaker 模型存储在 S3 中。
  6. 偏差检查步骤:将 S3 中的训练数据和模型构件用作输入,以使用 SageMaker Clarify 检查模型偏差,并将模型偏差报告和基线指标存储在 S3 中。
  7. 偏差可解释性步骤:将 S3 中的训练数据和模型构件用作输入,以运行 SageMaker Clarify,并将模型可解释性报告和基线指标存储在 S3 中。
  8. 注册步骤:将模型、偏差和可解释性基准指标用作输入,以运行 SageMaker Pipelines 预定义步骤,以在 SageMaker 模型注册表中注册模型。
  9. 部署步骤:将 AWS Lambda 处理程序函数、模型和端点配置用作输入,以运行 SageMaker Pipelines 预定义步骤,从而将模型部署到 SageMaker 实时推理端点。

SageMaker Pipelines 提供很多预定义的步骤类型,例如数据处理步骤、模型训练步骤、模型调优步骤以及批量转换步骤。有关更多信息,请参阅《Amazon SageMaker 开发人员指南》中的管道步骤。在接下来的步骤中,您将分别配置和定义每个管道步骤,然后通过将管道步骤与输入参数结合来定义管道本身。

数据处理步骤:在此步骤中,准备一个 Python 脚本来摄取原始文件;进行缺失值插补、特征工程等处理;并管理用于模型构建的培训训练、验证和测试拆分。复制、粘贴并运行以下代码,以构建您的处理脚本。

%%writefile preprocessing.py

import argparse
import pathlib
import boto3
import os
import pandas as pd
import logging
from sklearn.model_selection import train_test_split

logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler())


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--train-ratio", type=float, default=0.8)
    parser.add_argument("--validation-ratio", type=float, default=0.1)
    parser.add_argument("--test-ratio", type=float, default=0.1)
    args, _ = parser.parse_known_args()
    logger.info("Received arguments {}".format(args))
    
    # Set local path prefix in the processing container
    local_dir = "/opt/ml/processing"    
    
    input_data_path_claims = os.path.join("/opt/ml/processing/claims", "claims.csv")
    input_data_path_customers = os.path.join("/opt/ml/processing/customers", "customers.csv")
    
    logger.info("Reading claims data from {}".format(input_data_path_claims))
    df_claims = pd.read_csv(input_data_path_claims)
    
    logger.info("Reading customers data from {}".format(input_data_path_customers))
    df_customers = pd.read_csv(input_data_path_customers)
    
    logger.debug("Formatting column names.")
    # Format column names
    df_claims = df_claims.rename({c : c.lower().strip().replace(' ', '_') for c in df_claims.columns}, axis = 1)
    df_customers = df_customers.rename({c : c.lower().strip().replace(' ', '_') for c in df_customers.columns}, axis = 1)
    
    logger.debug("Joining datasets.")
    # Join datasets
    df_data = df_claims.merge(df_customers, on = 'policy_id', how = 'left')

    # Drop selected columns not required for model building
    df_data = df_data.drop(['customer_zip'], axis = 1)
    
    # Select Ordinal columns
    ordinal_cols = ["police_report_available", "policy_liability", "customer_education"]

    # Select categorical columns and filling with na
    cat_cols_all = list(df_data.select_dtypes('object').columns)
    cat_cols = [c for c in cat_cols_all if c not in ordinal_cols]
    df_data[cat_cols] = df_data[cat_cols].fillna('na')
    
    logger.debug("One-hot encoding categorical columns.")
    # One-hot encoding categorical columns
    df_data = pd.get_dummies(df_data, columns = cat_cols)
    
    logger.debug("Encoding ordinal columns.")
    # Ordinal encoding
    mapping = {
               "Yes": "1",
               "No": "0" 
              }
    df_data['police_report_available'] = df_data['police_report_available'].map(mapping)
    df_data['police_report_available'] = df_data['police_report_available'].astype(float)

    mapping = {
               "15/30": "0",
               "25/50": "1", 
               "30/60": "2",
               "100/200": "3"
              }
    
    df_data['policy_liability'] = df_data['policy_liability'].map(mapping)
    df_data['policy_liability'] = df_data['policy_liability'].astype(float)

    mapping = {
               "Below High School": "0",
               "High School": "1", 
               "Associate": "2",
               "Bachelor": "3",
               "Advanced Degree": "4"
              }
    
    df_data['customer_education'] = df_data['customer_education'].map(mapping)
    df_data['customer_education'] = df_data['customer_education'].astype(float)
    
    df_processed = df_data.copy()
    df_processed.columns = [c.lower() for c in df_data.columns]
    df_processed = df_processed.drop(["policy_id", "customer_gender_unkown"], axis=1)
    
    # Split into train, validation, and test sets
    train_ratio = args.train_ratio
    val_ratio = args.validation_ratio
    test_ratio = args.test_ratio
    
    logger.debug("Splitting data into train, validation, and test sets")
    
    y = df_processed['fraud']
    X = df_processed.drop(['fraud'], axis = 1)
    X_train_val, X_test, y_train_val, y_test = train_test_split(X, y, test_size=test_ratio, random_state=42)
    X_train, X_val, y_train, y_val = train_test_split(X_train_val, y_train_val, test_size=val_ratio, random_state=42)

    train_df = pd.concat([y_train, X_train], axis = 1)
    val_df = pd.concat([y_val, X_val], axis = 1)
    test_df = pd.concat([y_test, X_test], axis = 1)
    dataset_df = pd.concat([y, X], axis = 1)
    
    logger.info("Train data shape after preprocessing: {}".format(train_df.shape))
    logger.info("Validation data shape after preprocessing: {}".format(val_df.shape))
    logger.info("Test data shape after preprocessing: {}".format(test_df.shape))
    
    # Save processed datasets to the local paths in the processing container.
    # SageMaker will upload the contents of these paths to S3 bucket
    logger.debug("Writing processed datasets to container local path.")
    train_output_path = os.path.join(f"{local_dir}/train", "train.csv")
    validation_output_path = os.path.join(f"{local_dir}/val", "validation.csv")
    test_output_path = os.path.join(f"{local_dir}/test", "test.csv")
    full_processed_output_path = os.path.join(f"{local_dir}/full", "dataset.csv")

    logger.info("Saving train data to {}".format(train_output_path))
    train_df.to_csv(train_output_path, index=False)
    
    logger.info("Saving validation data to {}".format(validation_output_path))
    val_df.to_csv(validation_output_path, index=False)

    logger.info("Saving test data to {}".format(test_output_path))
    test_df.to_csv(test_output_path, index=False)
    
    logger.info("Saving full processed data to {}".format(full_processed_output_path))
    dataset_df.to_csv(full_processed_output_path, index=False)

接下来,复制、粘贴并运行以下代码块以实例化处理器和 SageMaker Pipelines 步骤来运行处理脚本。由于处理脚本是在 Pandas 中编写的,您将使用 SKLearnProcessor。SageMaker Pipelines ProcessingStep 函数采用以下参数:处理器、原始数据集的输入 S3 位置,以及保存已处理数据集的输出 S3 位置。训练、验证和测试拆分比等其他参数通过 job_arguments 参数提供。

from sagemaker.workflow.pipeline_context import PipelineSession
# Upload processing script to S3
s3_client.upload_file(
    Filename="preprocessing.py", Bucket=write_bucket, Key=f"{write_prefix}/scripts/preprocessing.py"
)

# Define the SKLearnProcessor configuration
sklearn_processor = SKLearnProcessor(
    framework_version="0.23-1",
    role=sagemaker_role,
    instance_count=1,
    instance_type=process_instance_type,
    base_job_name=f"{base_job_name_prefix}-processing",
)

# Define pipeline processing step
process_step = ProcessingStep(
    name="DataProcessing",
    processor=sklearn_processor,
    inputs=[
        ProcessingInput(source=claims_data_uri, destination="/opt/ml/processing/claims"),
        ProcessingInput(source=customers_data_uri, destination="/opt/ml/processing/customers")
    ],
    outputs=[
        ProcessingOutput(destination=f"{processing_output_uri}/train_data", output_name="train_data", source="/opt/ml/processing/train"),
        ProcessingOutput(destination=f"{processing_output_uri}/validation_data", output_name="validation_data", source="/opt/ml/processing/val"),
        ProcessingOutput(destination=f"{processing_output_uri}/test_data", output_name="test_data", source="/opt/ml/processing/test"),
        ProcessingOutput(destination=f"{processing_output_uri}/processed_data", output_name="processed_data", source="/opt/ml/processing/full")
    ],
    job_arguments=[
        "--train-ratio", "0.8", 
        "--validation-ratio", "0.1",
        "--test-ratio", "0.1"
    ],
    code=f"s3://{write_bucket}/{write_prefix}/scripts/preprocessing.py"
)

复制、粘贴并运行以下代码块以准备训练脚本。此脚本封装了 XGBoost 二进制分类器的训练逻辑。模型训练中使用的超参数将在本教程后面的训练步骤定义中提供。

%%writefile xgboost_train.py

import argparse
import os
import joblib
import json
import pandas as pd
import xgboost as xgb
from sklearn.metrics import roc_auc_score

if __name__ == "__main__":
    parser = argparse.ArgumentParser()

    # Hyperparameters and algorithm parameters are described here
    parser.add_argument("--num_round", type=int, default=100)
    parser.add_argument("--max_depth", type=int, default=3)
    parser.add_argument("--eta", type=float, default=0.2)
    parser.add_argument("--subsample", type=float, default=0.9)
    parser.add_argument("--colsample_bytree", type=float, default=0.8)
    parser.add_argument("--objective", type=str, default="binary:logistic")
    parser.add_argument("--eval_metric", type=str, default="auc")
    parser.add_argument("--nfold", type=int, default=3)
    parser.add_argument("--early_stopping_rounds", type=int, default=3)
    

    # SageMaker specific arguments. Defaults are set in the environment variables
    # Set location of input training data
    parser.add_argument("--train_data_dir", type=str, default=os.environ.get("SM_CHANNEL_TRAIN"))
    # Set location of input validation data
    parser.add_argument("--validation_data_dir", type=str, default=os.environ.get("SM_CHANNEL_VALIDATION"))
    # Set location where trained model will be stored. Default set by SageMaker, /opt/ml/model
    parser.add_argument("--model_dir", type=str, default=os.environ.get("SM_MODEL_DIR"))
    # Set location where model artifacts will be stored. Default set by SageMaker, /opt/ml/output/data
    parser.add_argument("--output_data_dir", type=str, default=os.environ.get("SM_OUTPUT_DATA_DIR"))
    
    args = parser.parse_args()

    data_train = pd.read_csv(f"{args.train_data_dir}/train.csv")
    train = data_train.drop("fraud", axis=1)
    label_train = pd.DataFrame(data_train["fraud"])
    dtrain = xgb.DMatrix(train, label=label_train)
    
    
    data_validation = pd.read_csv(f"{args.validation_data_dir}/validation.csv")
    validation = data_validation.drop("fraud", axis=1)
    label_validation = pd.DataFrame(data_validation["fraud"])
    dvalidation = xgb.DMatrix(validation, label=label_validation)
    
    # Choose XGBoost model hyperparameters
    params = {"max_depth": args.max_depth,
              "eta": args.eta,
              "objective": args.objective,
              "subsample" : args.subsample,
              "colsample_bytree":args.colsample_bytree
             }
    
    num_boost_round = args.num_round
    nfold = args.nfold
    early_stopping_rounds = args.early_stopping_rounds
    
    # Cross-validate train XGBoost model
    cv_results = xgb.cv(
        params=params,
        dtrain=dtrain,
        num_boost_round=num_boost_round,
        nfold=nfold,
        early_stopping_rounds=early_stopping_rounds,
        metrics=["auc"],
        seed=42,
    )
    
    model = xgb.train(params=params, dtrain=dtrain, num_boost_round=len(cv_results))
    
    train_pred = model.predict(dtrain)
    validation_pred = model.predict(dvalidation)
    
    train_auc = roc_auc_score(label_train, train_pred)
    validation_auc = roc_auc_score(label_validation, validation_pred)
    
    print(f"[0]#011train-auc:{train_auc:.2f}")
    print(f"[0]#011validation-auc:{validation_auc:.2f}")

    metrics_data = {"hyperparameters" : params,
                    "binary_classification_metrics": {"validation:auc": {"value": validation_auc},
                                                      "train:auc": {"value": train_auc}
                                                     }
                   }
              
    # Save the evaluation metrics to the location specified by output_data_dir
    metrics_location = args.output_data_dir + "/metrics.json"
    
    # Save the trained model to the location specified by model_dir
    model_location = args.model_dir + "/xgboost-model"

    with open(metrics_location, "w") as f:
        json.dump(metrics_data, f)

    with open(model_location, "wb") as f:
        joblib.dump(model, f)

使用 SageMaker XGBoost 估算器和 SageMaker Pipelines TrainingStep 函数设置模型训练。

# Set XGBoost model hyperparameters 
hyperparams = {  
"eval_metric" : "auc",
"objective": "binary:logistic",
"num_round": "5",
"max_depth":"5",
"subsample":"0.75",
"colsample_bytree":"0.75",
"eta":"0.5"
}

# Set XGBoost estimator
xgb_estimator = XGBoost(
entry_point="xgboost_train.py", 
output_path=estimator_output_uri,
code_location=estimator_output_uri,
hyperparameters=hyperparams,
role=sagemaker_role,
# Fetch instance type and count from pipeline parameters
instance_count=train_instance_count,
instance_type=train_instance_type,
framework_version="1.3-1"
)

# Access the location where the preceding processing step saved train and validation datasets
# Pipeline step properties can give access to outputs which can be used in succeeding steps
s3_input_train = TrainingInput(
s3_data=process_step.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri, 
content_type="csv", 
s3_data_type="S3Prefix"
)
s3_input_validation = TrainingInput(
s3_data=process_step.properties.ProcessingOutputConfig.Outputs["validation_data"].S3Output.S3Uri,
content_type="csv",
s3_data_type="S3Prefix"
)

# Set pipeline training step
train_step = TrainingStep(
name="XGBModelTraining",
estimator=xgb_estimator,
inputs={
"train":s3_input_train, # Train channel 
"validation": s3_input_validation # Validation channel
}
)

复制、粘贴并运行以下代码块,该代码块将用于使用 SageMaker Pipelines CreateModelStep 函数创建 SageMaker 模型。此步骤利用训练步骤的输出来打包模型以进行部署。 请注意,实例类型参数的值使用您在教程前面定义的 SageMaker Pipelines 参数传递。

# Create a SageMaker model
model = sagemaker.model.Model(
    image_uri=training_image,
    model_data=train_step.properties.ModelArtifacts.S3ModelArtifacts,
    sagemaker_session=sess,
    role=sagemaker_role
)

# Specify model deployment instance type
inputs = sagemaker.inputs.CreateModelInput(instance_type=deploy_instance_type_param)

create_model_step = CreateModelStep(name="FraudDetModel", model=model, inputs=inputs)

在 ML 工作流中,重要的是评估一个训练过的模型的潜在偏差,并了解输入数据中的各种特征如何影响模型预测。SageMaker Pipelines 通过一种可用于执行三种类型检查的 ClarifyCheckStep 函数:数据偏差检查(训练前)、模型偏差检查(训练后)和模型可解释性检测。为了减少运行时间,在本教程中,您只实施了偏差和可解释性检查。复制、粘贴并运行以下代码块来设置 SageMaker Clarify 进行模型偏差检查。请注意,此步骤通过 properties 属性获取资产,如训练数据和上一步中创建的 SageMaker 模型。当管道被执行时,这个步骤直到提供输入的步骤运行完成后才开始。有关更多详细信息,请参阅《Amazon SageMaker 开发人员指南》中的步骤之间的数据依赖关系。为了管理成本和教程运行时间,将 ModelBiasCheckConfig 函数配置为仅计算一个偏差指标 DPPL。有关 SageMaker Clarify 中提供的偏差指标的更多信息,请参阅《Amazon SageMaker 开发人员指南》中的衡量训练后数据和模型偏差

# Set up common configuration parameters to be used across multiple steps
check_job_config = CheckJobConfig(
    role=sagemaker_role,
    instance_count=1,
    instance_type=clarify_instance_type,
    volume_size_in_gb=30,
    sagemaker_session=sess,
)

# Set up configuration of data to be used for model bias check
model_bias_data_config = sagemaker.clarify.DataConfig(
    # Fetch S3 location where processing step saved train data
    s3_data_input_path=process_step.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri,
    s3_output_path=bias_report_output_uri,
    label=target_col,
    dataset_type="text/csv",
    s3_analysis_config_output_path=clarify_bias_config_output_uri
)

# Set up details of the trained model to be checked for bias
model_config = sagemaker.clarify.ModelConfig(
    # Pull model name from model creation step
    model_name=create_model_step.properties.ModelName,
    instance_count=train_instance_count,
    instance_type=train_instance_type
)

# Set up column and categories that are to be checked for bias
model_bias_config = sagemaker.clarify.BiasConfig(
    label_values_or_threshold=[0],
    facet_name="customer_gender_female",
    facet_values_or_threshold=[1]
)

# Set up model predictions configuration to get binary labels from probabilities
model_predictions_config = sagemaker.clarify.ModelPredictedLabelConfig(probability_threshold=0.5)

model_bias_check_config = ModelBiasCheckConfig(
    data_config=model_bias_data_config,
    data_bias_config=model_bias_config,
    model_config=model_config,
    model_predicted_label_config=model_predictions_config,
    methods=["DPPL"]
)

# Set up pipeline model bias check step
model_bias_check_step = ClarifyCheckStep(
    name="ModelBiasCheck",
    clarify_check_config=model_bias_check_config,
    check_job_config=check_job_config,
    skip_check=skip_check_model_bias_param,
    register_new_baseline=register_new_baseline_model_bias_param,
    supplied_baseline_constraints=supplied_baseline_constraints_model_bias_param
)

复制、粘贴并运行以下代码块以设置模型可解释性检查。此步骤提供功能重要性之类的洞察(输入功能如何影响模型预测)。

# Set configuration of data to be used for model explainability check
model_explainability_data_config = sagemaker.clarify.DataConfig(
    # Fetch S3 location where processing step saved train data
    s3_data_input_path=process_step.properties.ProcessingOutputConfig.Outputs["train_data"].S3Output.S3Uri,
    s3_output_path=explainability_report_output_uri,
    label=target_col,
    dataset_type="text/csv",
    s3_analysis_config_output_path=clarify_explainability_config_output_uri 
)

# Set SHAP configuration for Clarify to compute global and local SHAP values for feature importance
shap_config = sagemaker.clarify.SHAPConfig(
    seed=42, 
    num_samples=100,
    agg_method="mean_abs",
    save_local_shap_values=True
)

model_explainability_config = ModelExplainabilityCheckConfig(
    data_config=model_explainability_data_config,
    model_config=model_config,
    explainability_config=shap_config
)

# Set pipeline model explainability check step
model_explainability_step = ClarifyCheckStep(
    name="ModelExplainabilityCheck",
    clarify_check_config=model_explainability_config,
    check_job_config=check_job_config,
    skip_check=skip_check_model_explainability_param,
    register_new_baseline=register_new_baseline_model_explainability_param,
    supplied_baseline_constraints=supplied_baseline_constraints_model_explainability_param
)

在生产系统中,并非所有训练模型都会被部署。通常,只部署性能优于所选评估指标的阈值的模型。在此步骤中,您将构建 Python 脚本,以使用受试者工作特征曲线下面积(ROC-AUC)指标在测试集中对模型进行评分。在随后的步骤中,将根据此指标使用模型的性能,以确定是否应该注册和部署模型。复制、粘贴并运行以下代码,以构建评估脚本来摄取测试数据集并生成 AUC 指标。

%%writefile evaluate.py

import json
import logging
import pathlib
import pickle
import tarfile

import numpy as np
import pandas as pd
import xgboost as xgb

from sklearn.metrics import roc_auc_score

logger = logging.getLogger()
logger.setLevel(logging.INFO)
logger.addHandler(logging.StreamHandler())


if __name__ == "__main__":
    model_path = "/opt/ml/processing/model/model.tar.gz"
    with tarfile.open(model_path) as tar:
        tar.extractall(path=".")

    logger.debug("Loading xgboost model.")
    # The name of the file should match how the model was saved in the training script
    model = pickle.load(open("xgboost-model", "rb"))

    logger.debug("Reading test data.")
    test_local_path = "/opt/ml/processing/test/test.csv"
    df_test = pd.read_csv(test_local_path)
    
    # Extract test set target column
    y_test = df_test.iloc[:, 0].values
   
    cols_when_train = model.feature_names
    # Extract test set feature columns
    X = df_test[cols_when_train].copy()
    X_test = xgb.DMatrix(X)

    logger.info("Generating predictions for test data.")
    pred = model.predict(X_test)
    
    # Calculate model evaluation score
    logger.debug("Calculating ROC-AUC score.")
    auc = roc_auc_score(y_test, pred)
    metric_dict = {
        "classification_metrics": {"roc_auc": {"value": auc}}
    }
    
    # Save model evaluation metrics
    output_dir = "/opt/ml/processing/evaluation"
    pathlib.Path(output_dir).mkdir(parents=True, exist_ok=True)

    logger.info("Writing evaluation report with ROC-AUC: %f", auc)
    evaluation_path = f"{output_dir}/evaluation.json"
    with open(evaluation_path, "w") as f:
        f.write(json.dumps(metric_dict))

接下来,复制、粘贴并运行下面的代码块来实例化处理器和 SageMaker Pipelines 步骤,以运行评估脚本。要处理自定义脚本,您可以使用 ScriptProcessor。SageMaker Pipelines ProcessingStep 函数采用以下参数:处理器、测试数据集的 S3 输入位置、模型构件和存储评估结果的输出位置。此外,还提供 property_files 参数。您可以使用属性文件存储来自处理步骤输出的信息,在本例中,该输出是一个带有模型性能指标的 json 文件。如教程后面所示,这对于确定什么时候应该运行条件步骤特别有用。

# Upload model evaluation script to S3
s3_client.upload_file(
    Filename="evaluate.py", Bucket=write_bucket, Key=f"{write_prefix}/scripts/evaluate.py"
)

eval_processor = ScriptProcessor(
    image_uri=training_image,
    command=["python3"],
    instance_type=predictor_instance_type,
    instance_count=predictor_instance_count,
    base_job_name=f"{base_job_name_prefix}-model-eval",
    sagemaker_session=sess,
    role=sagemaker_role,
)
evaluation_report = PropertyFile(
    name="FraudDetEvaluationReport",
    output_name="evaluation",
    path="evaluation.json",
)

# Set model evaluation step
evaluation_step = ProcessingStep(
    name="XGBModelEvaluate",
    processor=eval_processor,
    inputs=[
        ProcessingInput(
            # Fetch S3 location where train step saved model artifacts
            source=train_step.properties.ModelArtifacts.S3ModelArtifacts,
            destination="/opt/ml/processing/model",
        ),
        ProcessingInput(
            # Fetch S3 location where processing step saved test data
            source=process_step.properties.ProcessingOutputConfig.Outputs["test_data"].S3Output.S3Uri,
            destination="/opt/ml/processing/test",
        ),
    ],
    outputs=[
        ProcessingOutput(destination=f"{model_eval_output_uri}", output_name="evaluation", source="/opt/ml/processing/evaluation"),
    ],
    code=f"s3://{write_bucket}/{write_prefix}/scripts/evaluate.py",
    property_files=[evaluation_report],
)

使用 SageMaker 模型注册表,您可以对模型进行编目、管理模型版本和选择性地将模型部署到生产环境中。复制、粘贴并运行以下代码块以设置模型注册表步骤。两个参数 model_metricsdrift_check_baselines 含有以前在教程中由 ClarifyCheckStep 函数计算的基线指标。您还可以提供自己的自定义基线指标。这些参数背后的目的是提供一种配置与模型相关联的基线的方法,以便在漂移检查和模型监控任务中使用它们。每次执行管道时,您都可以选择使用新计算的基线来更新这些参数。

# Fetch baseline constraints to record in model registry
model_metrics = ModelMetrics(
    bias_post_training=MetricsSource(
        s3_uri=model_bias_check_step.properties.CalculatedBaselineConstraints,
        content_type="application/json"
    ),
    explainability=MetricsSource(
        s3_uri=model_explainability_step.properties.CalculatedBaselineConstraints,
        content_type="application/json"
    ),
)

# Fetch baselines to record in model registry for drift check
drift_check_baselines = DriftCheckBaselines(
    bias_post_training_constraints=MetricsSource(
        s3_uri=model_bias_check_step.properties.BaselineUsedForDriftCheckConstraints,
        content_type="application/json",
    ),
    explainability_constraints=MetricsSource(
        s3_uri=model_explainability_step.properties.BaselineUsedForDriftCheckConstraints,
        content_type="application/json",
    ),
    explainability_config_file=FileSource(
        s3_uri=model_explainability_config.monitoring_analysis_config_uri,
        content_type="application/json",
    ),
)

# Define register model step
register_step = RegisterModel(
    name="XGBRegisterModel",
    estimator=xgb_estimator,
    # Fetching S3 location where train step saved model artifacts
    model_data=train_step.properties.ModelArtifacts.S3ModelArtifacts,
    content_types=["text/csv"],
    response_types=["text/csv"],
    inference_instances=[predictor_instance_type],
    transform_instances=[predictor_instance_type],
    model_package_group_name=model_package_group_name,
    approval_status=model_approval_status_param,
    # Registering baselines metrics that can be used for model monitoring
    model_metrics=model_metrics,
    drift_check_baselines=drift_check_baselines
)

使用 Amazon SageMaker,可以通过几种方式部署已注册的模型进行推理。在此步骤中,使用 LambdaStep 函数部署模型。虽然您通常应该将 SageMaker 项目用于遵循 CI/CD 最佳实践的可靠模型部署,但在某些情况下,将 LambdaStep 用于轻量级模型部署到开发、测试和服务于低流量的内部端点是有意义的。LambdaStep 函数提供与 AWS Lambda 的本地集成,因此您可以在管道中实现自定义逻辑,而无需预调配或管理服务器。在 SageMaker Pipelines 的上下文中,LambdaStep 允许您向管道添加 AWS Lambda 函数,以支持任意计算操作,特别是持续时间较短的轻量级操作。请记住,在 SageMaker Pipelines LambdaStep 中,Lambda 函数的最大运行时间限制为 10 分钟,可修改的默认超时为 2 分钟。 

您有两种方式可将 LambdaStep 添加到您的管道中。首先,您可以提供使用 AWS Cloud Development Kit(AWS CDK)、AWS 管理控制台或其他方式创建的现有 Lambda 函数的 ARN。其次,高级的 SageMaker Python 开发工具包具有一个 Lambda 助手便利类,您可以使用它与定义管道的其他代码一起创建一个新的 Lambda 函数。在本教程中,您使用的是第二种方法。 复制、粘贴并运行以下代码以定义 Lambda 处理程序函数。这是一个自定义 Python 脚本,它在模型属性(比如模型名称)中采用,并部署到实时端点。

%%writefile lambda_deployer.py

"""
Lambda function creates an endpoint configuration and deploys a model to real-time endpoint. 
Required parameters for deployment are retrieved from the event object
"""

import json
import boto3


def lambda_handler(event, context):
    sm_client = boto3.client("sagemaker")

    # Details of the model created in the Pipeline CreateModelStep
    model_name = event["model_name"]
    model_package_arn = event["model_package_arn"]
    endpoint_config_name = event["endpoint_config_name"]
    endpoint_name = event["endpoint_name"]
    role = event["role"]
    instance_type = event["instance_type"]
    instance_count = event["instance_count"]
    primary_container = {"ModelPackageName": model_package_arn}

    # Create model
    model = sm_client.create_model(
        ModelName=model_name,
        PrimaryContainer=primary_container,
        ExecutionRoleArn=role
    )

    # Create endpoint configuration
    create_endpoint_config_response = sm_client.create_endpoint_config(
        EndpointConfigName=endpoint_config_name,
        ProductionVariants=[
        {
            "VariantName": "Alltraffic",
            "ModelName": model_name,
            "InitialInstanceCount": instance_count,
            "InstanceType": instance_type,
            "InitialVariantWeight": 1
        }
        ]
    )

    # Create endpoint
    create_endpoint_response = sm_client.create_endpoint(
        EndpointName=endpoint_name, 
        EndpointConfigName=endpoint_config_name
    )

复制、粘贴并运行以下代码块以创建 LambdaStep。模型、端点名称和部署实例类型和计数等所有参数都使用 inputs 参数提供。

# The function name must contain sagemaker
function_name = "sagemaker-fraud-det-demo-lambda-step"
# Define Lambda helper class can be used to create the Lambda function required in the Lambda step
func = Lambda(
    function_name=function_name,
    execution_role_arn=sagemaker_role,
    script="lambda_deployer.py",
    handler="lambda_deployer.lambda_handler",
    timeout=600,
    memory_size=10240,
)

# The inputs used in the lambda handler are passed through the inputs argument in the 
# LambdaStep and retrieved via the `event` object within the `lambda_handler` function

lambda_deploy_step = LambdaStep(
    name="LambdaStepRealTimeDeploy",
    lambda_func=func,
    inputs={
        "model_name": pipeline_model_name,
        "endpoint_config_name": endpoint_config_name,
        "endpoint_name": endpoint_name,
        "model_package_arn": register_step.steps[0].properties.ModelPackageArn,
        "role": sagemaker_role,
        "instance_type": deploy_instance_type_param,
        "instance_count": deploy_instance_count_param
    }
)

在此步骤中,您使用 ConditionStep 根据曲线下面积(AUC)指标来比较当前模型的性能。只有当性能大于或等于阈值 AUC(这里选择为 0.7)时,管道才会执行偏差和可解释性检查,注册模型,并部署它。像这样的条件步骤有助于选择性地将最佳模型部署到生产环境中。复制、粘贴并运行以下代码以定义条件步骤。

# Evaluate model performance on test set
cond_gte = ConditionGreaterThanOrEqualTo(
    left=JsonGet(
        step_name=evaluation_step.name,
        property_file=evaluation_report,
        json_path="classification_metrics.roc_auc.value",
    ),
    right=0.7, # Threshold to compare model performance against
)
condition_step = ConditionStep(
    name="CheckFraudDetXGBEvaluation",
    conditions=[cond_gte],
    if_steps=[create_model_step, model_bias_check_step, model_explainability_step, register_step, lambda_deploy_step], 
    else_steps=[]
)

第 4 步:构建并运行管道

定义所有的组件步骤后,您可以将它们组合到一个 SageMaker Pipelines 对象中。您不需要指定执行顺序,因为 SageMaker Pipelines 会根据步骤之间的依赖关系自动推断执行顺序。

复制、粘贴并运行以下代码块以设置管道。管道定义采用您在第 2 步中定义的所有参数和组件步骤列表。创建模型、偏差和可解释性检查、模型注册和 lambda 部署等步骤没有在管道定义中列出,因为它们不会运行,除非条件步骤的计算结果为 true。如果条件步骤为 true,则后续步骤将根据指定的输入和输出按顺序运行。

# Create the Pipeline with all component steps and parameters
pipeline = Pipeline(
    name=pipeline_name,
    parameters=[process_instance_type_param, 
                train_instance_type_param, 
                train_instance_count_param, 
                deploy_instance_type_param,
                deploy_instance_count_param,
                clarify_instance_type_param,
                skip_check_model_bias_param,
                register_new_baseline_model_bias_param,
                supplied_baseline_constraints_model_bias_param,
                skip_check_model_explainability_param,
                register_new_baseline_model_explainability_param,
                supplied_baseline_constraints_model_explainability_param,
                model_approval_status_param],
    steps=[
        process_step,
        train_step,
        evaluation_step,
        condition_step
    ],
    sagemaker_session=sess
    
)

在您的笔记本的单元格中复制、粘贴并运行以下代码。如果管道已存在,则代码将更新管道。如果管道不存在,则它将创建一个新管道。忽略任何 SageMaker 开发工具包警告,如“No finished training job found associated with this estimator.Please make sure this estimator is only used for building workflow config.”(未找到与此估算器关联的已完成训练任务。请确保此估算器仅用于构建工作流配置。)

# Create a new or update existing Pipeline
pipeline.upsert(role_arn=sagemaker_role)

# Full Pipeline description
pipeline_definition = json.loads(pipeline.describe()['PipelineDefinition'])
pipeline_definition

SageMaker 将用有向无环图(DAG)编码管道,其中每个节点都表示一个步骤,且节点之间的连接表示依赖关系。要从 SageMaker Studio 界面检查管道 DAG,请在左侧面板上选择 SageMaker Resources(SageMaker 资源)选项卡,从下拉列表中选择 Pipelines(管道),然后选择 FraudDetectXGBPipelineGraph(图表)。可以看到,您创建的管道步骤由图中的节点表示,节点之间的连接由 SageMaker 根据步骤定义中提供的输入和输出推断。

要执行管道,请运行以下代码语句。在此步骤中,管道执行参数作为参数提供。请转至左侧面板上的 SageMaker Resources(SageMaker 资源)选项卡,从下拉列表中选择 Pipelines(管道),然后选择 FraudDetectXGBPipelineExecutions(执行)。管道的当前运行被列出。

 

 

# Execute Pipeline
start_response = pipeline.start(parameters=dict(
        SkipModelBiasCheck=True,
        RegisterNewModelBiasBaseline=True,
        SkipModelExplainabilityCheck=True,
        RegisterNewModelExplainabilityBaseline=True)
                               )

要审查管道执行,请选择 Status(状态)选项卡。当所有步骤都成功执行时,图表中的节点变为绿色。

在 SageMaker Studio 界面中,选择左侧面板中的 SageMaker Resources(SageMaker 资源),然后从下拉列表中选择 Model registry(模型注册表)。注册的模型列在左侧面板的 Model group name(模型组名称)下。选择模型组名称以显示模型版本的列表。每当您运行管道时,如果模型版本满足评估的条件阈值,新模型版本都会添加到注册表中。选择一个模型版本以查看详细信息,例如模型端点和模型可解释性报告。

 

 

第 5 步:通过调用端点来测试管道

在此教程中,该模型的得分高于所选择的 0.7 AUC 的阈值。因此,条件步骤会将模型注册并将部署到实时推理端点中。

在 SageMaker Studio 界面上,选择左侧窗格中的 SageMaker Resources(SageMaker 资源)选项卡,选择 Endpoints(端点),并等到 fraud-detect-xgb-pipeline-endpoint 的状态变为 InService(使用中)。

端点状态变为 InService(使用中)后,复制、粘贴并运行以下代码以调用端点并运行示例推理。代码将返回测试数据集中前五个样本的模型预测。

# Fetch test data to run predictions with the endpoint
test_df = pd.read_csv(f"{processing_output_uri}/test_data/test.csv")

# Create SageMaker Predictor from the deployed endpoint
predictor = sagemaker.predictor.Predictor(endpoint_name, 
                                          sagemaker_session=sess,
                                          serializer=CSVSerializer(),
                                          deserializer=CSVDeserializer()
                                         )
# Test endpoint with payload of 5 samples
payload = test_df.drop(["fraud"], axis=1).iloc[:5]
result = predictor.predict(payload.values)
prediction_df = pd.DataFrame()
prediction_df["Prediction"] = result
prediction_df["Label"] = test_df["fraud"].iloc[:5].values
prediction_df

第 6 步:清理资源

最佳实践是删除您不再使用的资源,以免产生意外费用。

复制并粘贴以下代码块,以删除 Lambda 函数、模型、端点配置、端点和您在本教程中创建的管道。

# Delete the Lambda function
func.delete()

# Delete the endpoint
sm_client.delete_endpoint(EndpointName=endpoint_name)

# Delete the EndpointConfig
sm_client.delete_endpoint_config(EndpointConfigName=endpoint_config_name)

# Delete the model
sm_client.delete_model(ModelName=pipeline_model_name)

# Delete the pipeline
sm_client.delete_pipeline(PipelineName=pipeline_name)

要删除 S3 桶,请执行以下操作: 

  • 打开 Amazon S3 控制台。在导航栏上,选择 Buckets(桶)、sagemaker-<your-Region>-<your-account-id>,然后选择 fraud-detect-demo 旁的复选框。然后选择 Delete(删除)。
  • Delete objects(删除对象)对话框中,确认您是否已选中要删除的正确对象,并在 Permanently delete objects(永久删除对象)确认框中输入 permanently delete
  • 当此操作完成且桶为空时,您可以通过再次遵循相同程序来删除 sagemaker-<your-Region>-<your-account-id> 桶。

本教程中用于运行笔记本图像的数据科学内核将不断累积费用,直到您停止内核或执行以下步骤删除应用程序。 有关更多信息,请参阅《Amazon SageMaker 开发人员指南》中的关闭资源

要删除 SageMaker Studio 应用程序,请执行以下操作:在 SageMaker Studio 控制台中,选择 studio-user,然后通过选择 Delete app(删除应用程序)来删除 Apps(应用程序)下列出的所有应用程序。等待片刻直到状态更改为 Deleted(已删除)。

如果您在第 1 步中使用了一个现有的 SageMaker Studio 域,则跳过第 6 步的其余部分,直接进入结论部分。 

如果您在第 1 步运行 CloudFormation 模板来创建新的 SageMaker Studio 域,请继续执行以下步骤以删除由 CloudFormation 模板创建的域、用户和资源。  

要打开 CloudFormation 控制台,请在 AWS 管理控制台搜索栏中输入 CloudFormation,然后从搜索结果中选择 CloudFormation

CloudFormation 窗格上,选择 Stacks(堆栈)。从 Status(状态)下拉列表中,选择 Active(活动)。在 Stack name(堆栈名称)下方,选择 CFN-SM-IM-Lambda-catalog 以打开堆栈详细信息页面。

CFN-SM-IM-Lambda-catalog 堆栈详细信息页面上,选择 Delete(删除)以删除堆栈以及在第 1 步中创建的资源。

结论

恭喜! 您已完成自动化机器学习工作流教程。 

您已成功使用 Amazon SageMaker Pipelines 从数据处理、模型训练、模型评估、偏差和可解释性检查、条件模型注册和部署开始自动化端到端的 ML 工作流。最后,您使用 SageMaker 开发工具包将模型部署到实时推理端点,并使用示例有效负载对其进行测试。

您可以按照下面的步骤使用 Amazon SageMaker 继续机器学习之旅。

部署机器学习模型

了解如何部署 ML 模型以进行推理。
下一步 »

训练深度学习模型

了解如何构建、训练和调优 TensorFlow 深度学习模型。
下一步 »

查找更多操作教程

探索其他机器学习教程以进行深入了解。
下一步 »