使用 Amazon 中的推理管道将预处理逻辑部署到单个终端节点的 ML 模型中 SageMaker

为回归任务准备数据集。

在 Amazon SageMaker Studio 中@@ 打开一本笔记本。

若要导入所有必要的库并初始化工作环境，请在笔记本中使用以下示例代码：

import sagemaker
from sagemaker import get_execution_role

sagemaker_session = sagemaker.Session()

# Get a SageMaker-compatible role used by this Notebook Instance.
role = get_execution_role()

# S3 prefix
bucket = sagemaker_session.default_bucket()
prefix = "Scikit-LinearLearner-pipeline-abalone-example"

若要下载示例数据集，请将以下代码添加至您的笔记本：

! mkdir abalone_data
! aws s3 cp s3://sagemaker-sample-files/datasets/tabular/uci_abalone/abalone.csv ./abalone_data

请注意： 此模式中的示例使用 UCI 机器学习存储库中的 Abalone 数据集。

将数据集上传至 S3 存储桶。

在此前准备数据集的笔记本中，添加以下代码，以将示例数据上传至 S3 存储桶：

WORK_DIRECTORY = "abalone_data"

train_input = sagemaker_session.upload_data(
    path="{}/{}".format(WORK_DIRECTORY, "abalone.csv"),
    bucket=bucket,
    key_prefix="{}/{}".format(prefix, "train"),
)

准备 preprocessor.py 脚本。

创建 SKLearn 预处理器对象。

要创建可以整合到最终推理管道中的 skLearn 预处理器对象（名为 skLearn Estimator），请在笔记本中运行以下代码： SageMaker

from sagemaker.sklearn.estimator import SKLearn

FRAMEWORK_VERSION = "0.23-1"
script_path = "sklearn_abalone_featurizer.py"

sklearn_preprocessor = SKLearn(
    entry_point=script_path,
    role=role,
    framework_version=FRAMEWORK_VERSION,
    instance_type="ml.c4.xlarge",
    sagemaker_session=sagemaker_session,
)
sklearn_preprocessor.fit({"train": train_input})

测试预处理器推理。

要确认您的预处理器定义正确，请在 SageMaker 笔记本中输入以下代码来启动批处理转换作业：

# Define a SKLearn Transformer from the trained SKLearn Estimator
transformer = sklearn_preprocessor.transformer(
    instance_count=1, instance_type="ml.m5.xlarge", assemble_with="Line", accept="text/csv"
)


# Preprocess training input
transformer.transform(train_input, content_type="text/csv")
print("Waiting for transform job: " + transformer.latest_transform_job.job_name)
transformer.wait()
preprocessed_train = transformer.output_path

创建模型对象。

要基于线性学习器算法创建模型对象，请在 SageMaker 笔记本中输入以下代码：

import boto3
from sagemaker.image_uris import retrieve

ll_image = retrieve("linear-learner", boto3.Session().region_name)
s3_ll_output_key_prefix = "ll_training_output"
s3_ll_output_location = "s3://{}/{}/{}/{}".format(
    bucket, prefix, s3_ll_output_key_prefix, "ll_model"
)

ll_estimator = sagemaker.estimator.Estimator(
    ll_image,
    role,
    instance_count=1,
    instance_type="ml.m4.2xlarge",
    volume_size=20,
    max_run=3600,
    input_mode="File",
    output_path=s3_ll_output_location,
    sagemaker_session=sagemaker_session,
)

ll_estimator.set_hyperparameters(feature_dim=10, predictor_type="regressor", mini_batch_size=32)

ll_train_data = sagemaker.inputs.TrainingInput(
    preprocessed_train,
    distribution="FullyReplicated",
    content_type="text/csv",
    s3_data_type="S3Prefix",
)

data_channels = {"train": ll_train_data}
ll_estimator.fit(inputs=data_channels, logs=True)

先前代码从公共 Amazon ECR 注册表中检索模型的相关 Amazon ECR Docker 映像，创建估算器对象，然后使用该对象训练回归模型。

部署管道模型。

要创建管道模型对象（即预处理器对象）并部署该对象，请在 SageMaker 笔记本中输入以下代码：

from sagemaker.model import Model
from sagemaker.pipeline import PipelineModel
import boto3
from time import gmtime, strftime

timestamp_prefix = strftime("%Y-%m-%d-%H-%M-%S", gmtime())

scikit_learn_inferencee_model = sklearn_preprocessor.create_model()
linear_learner_model = ll_estimator.create_model()

model_name = "inference-pipeline-" + timestamp_prefix
endpoint_name = "inference-pipeline-ep-" + timestamp_prefix
sm_model = PipelineModel(
    name=model_name, role=role, models= [scikit_learn_inferencee_model, linear_learner_model]
)

sm_model.deploy(initial_instance_count=1, instance_type="ml.c4.xlarge", endpoint_name=endpoint_name)

请注意：可调整模型对象中使用的实例类型，以满足您的需求。

测试推理。

要确认端点是否正常工作，请在 SageMaker 笔记本中运行以下示例推理代码：

from sagemaker.predictor import Predictor
from sagemaker.serializers import CSVSerializer

payload = "M, 0.44, 0.365, 0.125, 0.516, 0.2155, 0.114, 0.155"
actual_rings = 10
predictor = Predictor(
    endpoint_name=endpoint_name, sagemaker_session=sagemaker_session, serializer=CSVSerializer()
)

print(predictor.predict(payload))