ENGINEER BLOG

こんにちは！イノベーション本部の野村です。

今回はAmazon SageMakerのサーバーレス推論（Serverless Inference）を使ってみました。

はじめに

SageMakerには学習したモデルを使って推論する方法がいくつかあります。
（リアルタイム推論、バッチ変換、非同期推論、サーバーレス推論）

その中でもサーバーレス推論は、推論リクエストがあったときのみ自動的に起動して処理されるため
ユースケースによってはかなりコストを抑えることができる推論手法となっています。

そんなSageMakerのサーバーレス推論の使い方には主に以下の3つの方法があります。

• AWS SDK for Python (Boto3)
• SageMaker Python SDK
• SageMaker コンソールから

今回はこちらのサンプルコードを元に、それぞれの使い方について見ていきたいと思います。

やってみる

サンプルコードの内容としては、XGBoostの回帰モデルを使って、アワビの身体的な特徴から
年齢を推定するタスクとなっています。

• 環境
  • ml.m5.xlarge Notebook Instance
  • conda_python3 Kernel
  • ap-northeast-1

1. セットアップ

まずは各種セットアップをしていきます。

# ライブラリのアップデート
! pip install sagemaker botocore boto3 awscli --upgrade

# 必要なライブラリのインポート

import sagemaker
from sagemaker.estimator import Estimator
from time import gmtime, strftime

# clientのセットアップ
import boto3
client = boto3.client(service_name="sagemaker")
runtime = boto3.client(service_name="sagemaker-runtime")

# IAMロールやS3環境のセットアップ
boto_session = boto3.session.Session()
region = boto_session.region_name
print(region)

sagemaker_session = sagemaker.Session()
base_job_prefix = "xgboost-example"
role = sagemaker.get_execution_role()
print(role)

default_bucket = sagemaker_session.default_bucket()
s3_prefix = base_job_prefix
print(default_bucket)

# トレーニング用のインスタンスタイプの指定
training_instance_type = "ml.m5.xlarge"

# 学習用データの取得
! curl https://sagemaker-sample-files.s3.amazonaws.com/datasets/tabular/uci_abalone/train_csv/abalone_dataset1_train.csv > abalone_dataset1_train.csv

# 取得したデータをS3へアップロード
!aws s3 cp abalone_dataset1_train.csv s3://{default_bucket}/xgboost-regression/train.csv

2. トレーニング

つぎに、学習用のデータを使ってモデルをトレーニングしていきます。

from sagemaker.inputs import TrainingInput

training_path = f"s3://{default_bucket}/xgboost-regression/train.csv"
train_input = TrainingInput(training_path, content_type="text/csv")

model_path = f"s3://{default_bucket}/{s3_prefix}/xgb_model"

# xgboostのイメージをセット
image_uri = sagemaker.image_uris.retrieve(
    framework="xgboost",
    region=region,
    version="1.0-1",
    py_version="py3",
    instance_type=training_instance_type,
)

# Estimatorの設定
xgb_train = Estimator(
    image_uri=image_uri,
    instance_type=training_instance_type,
    instance_count=1,
    output_path=model_path,
    sagemaker_session=sagemaker_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,
)

# モデルのトレーニング
xgb_train.fit({"train": train_input})

3. デプロイ

やっと、ここからが本題です。

3-1. AWS SDK for Python (Boto3)

まずは、AWS SDK for Python (Boto3) を使ったやり方から見ていきたいと思います。

モデルのデプロイ

先程トレーニングしてできたモデルの場所を取得します。

model_artifacts = xgb_train.model_data
model_artifacts

モデルの名前を設定し、

model_name = "xgboost-serverless" + strftime("%Y-%m-%d-%H-%M-%S", gmtime())
print("Model name: " + model_name)

モデルをデプロイします。

byo_container_env_vars = {"SAGEMAKER_CONTAINER_LOG_LEVEL": "20", "SOME_ENV_VAR": "myEnvVar"}

create_model_response = client.create_model(
    ModelName=model_name,
    Containers=[
        {
            "Image": image_uri,
            "Mode": "SingleModel",
            "ModelDataUrl": model_artifacts,
            "Environment": byo_container_env_vars,
        }
    ],
    ExecutionRoleArn=role,
)

print("Model Arn: " + create_model_response["ModelArn"])

エンドポイント設定の作成

次に、エンドポイント設定の作成をするのですが
ここでサーバーレス推論に関する設定（ServerlessConfig）を行います。

「MemorySizeInMB」は1024 MB, 2048 MB, 3072 MB, 4096 MB, 5120 MB, 6144 MBの中から選び、
「MaxConcurrency」は1 ~ 200の間で設定します。

今回は「MemorySizeInMB」を4096、「MaxConcurrency」は1で設定しました。

xgboost_epc_name = "xgboost-serverless-epc" + strftime("%Y-%m-%d-%H-%M-%S", gmtime())

endpoint_config_response = client.create_endpoint_config(
    EndpointConfigName=xgboost_epc_name,
    ProductionVariants=[
        {
            "VariantName": "byoVariant",
            "ModelName": model_name,
            "ServerlessConfig": {
                "MemorySizeInMB": 4096,
                "MaxConcurrency": 1,
            },
        },
    ],
)

print("Endpoint Configuration Arn: " + endpoint_config_response["EndpointConfigArn"])

サーバーレスエンドポイントの作成

エンドポイント設定をもとに、サーバーレスのエンドポイントを作成します。

endpoint_name = "xgboost-serverless-ep" + strftime("%Y-%m-%d-%H-%M-%S", gmtime())

create_endpoint_response = client.create_endpoint(
    EndpointName=endpoint_name,
    EndpointConfigName=xgboost_epc_name,
)

print("Endpoint Arn: " + create_endpoint_response["EndpointArn"])

↓を実行し、しばらく待ってステータスがCreatingからInServiceに変われば作成完了です。

import time

describe_endpoint_response = client.describe_endpoint(EndpointName=endpoint_name)

while describe_endpoint_response["EndpointStatus"] == "Creating":
    describe_endpoint_response = client.describe_endpoint(EndpointName=endpoint_name)
    print(describe_endpoint_response["EndpointStatus"])
    time.sleep(15)

describe_endpoint_response

SageMakerコンソールから作成されたエンドポイントを選択して見てみると
タイプがサーバーレスになっていることがわかるかと思います。

推論リクエスト

できあがったサーバーレス推論のエンドポイントに推論リクエストを投げてみます。

response = runtime.invoke_endpoint(
    EndpointName=endpoint_name,
    Body=b".345,0.224414,.131102,0.042329,.279923,-0.110329,-0.099358,0.0",
    ContentType="text/csv",
)

無事、結果が返ってきました。

print(response["Body"].read())
# b'4.566554546356201'

3-2. SageMaker Python SDK

つぎに、SageMaker Python SDKを使ってやってみたいと思います。

いったん、先程デプロイしたモデルとエンドポイント設定、エンドポイントは削除しておきます。

client.delete_model(ModelName=model_name)
client.delete_endpoint_config(EndpointConfigName=xgboost_epc_name)
client.delete_endpoint(EndpointName=endpoint_name)

Python SDK用のライブラリをインポートしてサーバーレス推論のコンフィグを設定します。

from sagemaker.serverless import ServerlessInferenceConfig

serverless_config = ServerlessInferenceConfig(
    memory_size_in_mb=4096,
    max_concurrency=1
)

エンドポイントの名前は新たにタイムスタンプを取得しなおしておきます。

endpoint_name = "xgboost-serverless-ep" + strftime("%Y-%m-%d-%H-%M-%S", gmtime())

deploy()メソッドを使って、モデルのデプロイ、エンドポイント設定とエンドポイントの作成を一気に行います。

xgb_train.deploy(
    endpoint_name = endpoint_name,
    serverless_inference_config=serverless_config
)

コンソールからエンドポイントが作成されたことを確認し、推論リクエストを投げてみると
先ほどと同じ結果が返ってくることが確認できます。

response = runtime.invoke_endpoint(
    EndpointName=endpoint_name,
    Body=b".345,0.224414,.131102,0.042329,.279923,-0.110329,-0.099358,0.0",
    ContentType="text/csv",
)

print(response["Body"].read())
# b'4.566554546356201'

Python SDKを使うことでBoto3の時よりもかなり簡単にデプロイすることができました。

3-3. SageMaker コンソール

最後に、SageMakerコンソールからやってみたいと思います。

SageMakerコンソールの推論->モデルから対象のモデル（今回は先程デプロイしたモデル）を選択し、「エンドポイント設定の作成」をクリックします。

エンドポイント設定名を入力し、エンドポイントのタイプにサーバーレスを選択します。

バリアントは「本番稼働用バリアントを作成」をクリックして対象のモデルを選択し保存することで追加できます。

「エンドポイント設定の作成」ボタンを押して作成します。

次に、SageMakerコンソールの画面から、推論->エンドポイント->エンドポイントの作成をクリックします。

エンドポイント名を入力し、「既存のエンドポイント設定の使用」にチェックを入れたまま
先程作ったエンドポイント設定を選択します。

最後に「エンドポイントの作成」ボタンをクリックすると、サーバーレスのエンドポイントが作成されます。

出来上がったエンドポイントに推論リクエストを投げて同じ結果が返ってくればOKです。

おまけ

せっかくなので、Lambdaからも推論リクエストを投げてみたいと思います。

Lambdaの作成手順は割愛させていただきますが、タイムアウトの設定がデフォルトの3秒のままだと
エラーになる可能性があるので延ばしておきましょう。

lambda_function.pyに以下のコードを記述し、Testボタンで実行します。

import boto3

ENDPOINT_NAME = "作成したエンドポイント名"
values = [".345","0.224414",".131102","0.042329",".279923","-0.110329","-0.099358,0.0",]

def lambda_handler(event, context):

    client = boto3.client("sagemaker-runtime")

    # サーバーレスのエンドポイントに推論リクエストし、結果を受け取る
    response = client.invoke_endpoint(
        EndpointName=ENDPOINT_NAME,
        Body='{0}, {1}, {2}, {3}, {4}, {5}, {6}'.format(values[0], values[1], values[2], values[3], values[4], values[5], values[6]),
        ContentType='text/csv',
        Accept='application/json'
    )

    result = response['Body'].read().decode()

    return {
        'statusCode': 200,
        'body': result
    }

無事、Lambdaからサーバーレス推論用のエンドポイントにリクエストを投げ、結果が返ってきました。

（終わったら不要なモデル、エンドポイント設定、エンドポイントは削除しておきましょう。）

まとめ

今回はAmazon SageMakerのサーバーレス推論（Serverless Inference）を3つの方法で使ってみました。

推論ワークフローにおけるコストインパクトは大きいと思うので
ユースケースを見極めながら使っていきたいと思います。

長くなってしまいましたが、最後までお読みいただきありがとうございました。

少しでもお役に立てれば幸いです！