Amazon の推論パイプラインを使用して、1 つのエンドポイントの 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 機械学習のリポジトリのアバロンデータセットを使用しています。

データセットを 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 推定器と呼ばれる) を作成するには、 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))