訓練済みモデルをSageMakerエンドポイントにデプロイするの変更点

追加された行はこの色です。
削除された行はこの色です。
訓練済みモデルをSageMakerエンドポイントにデプロイするへ行く。
#mynavi(AWSメモ,Amazon SageMakerを使ってみる)
#setlinebreak(on);

* 概要 [#h894648f]
#html(<div class="pl10">)
Amazon SageMaker のエンドポイントに訓練済みモデルをデプロイして推論を行ってみた。
尚、今回は PyTorch を利用したが、 他にも tensorflow や chainer など他のフレームワークも利用できる。

Amazon SageMaker で PyTorch を使用
https://docs.aws.amazon.com/ja_jp/sagemaker/latest/dg/pytorch.html

#html(</div>)

* 目次 [#e3986f6f]
#contents
- 関連
-- [[AWSメモ]]
-- [[Amazon SageMakerを使ってみる]]
-- [[PyTorchで重回帰分析]]
- 参考
-- https://docs.aws.amazon.com/ja_jp/sagemaker/latest/dg/pytorch.html
-- https://sagemaker.readthedocs.io/en/stable/using_pytorch.html#deploy-endpoints-from-model-data
-- https://aws.amazon.com/jp/blogs/news/building-training-and-deploying-fastai-models-with-amazon-sagemaker/
-- https://github.com/aws/sagemaker-python-sdk/blob/master/src/sagemaker/pytorch/README.rst
-- https://github.com/awslabs/amazon-sagemaker-examples/blob/master/sagemaker-python-sdk/chainer_sentiment_analysis/src/sentiment_analysis.py


* モデルデータを作成しS3バケットに上げる [#a47349e0]
#html(<div class="pl10">)

モデルは [[PyTorchで重回帰分析]] で作成したものをそのまま利用する。

** アップロードするフォルダの構成 [#fdc2f419]
#html(<div class="pl10">)

以下の構成のフォルダを作成する。
#html(){{
<div style="padding: 10px; border: 1px solid #333; display: inline-block;">
sample_torch_model.tgz<br />
　└ sample_torch_model.pth    ...   エクスポートした訓練済みモデル<br />
　└ sample_torch_model.json   ...   今回はエンドポイント側で入力データの標準化をしたいので必要な情報をJSON化して一緒にアップしておく(後述)<br />
　└ sample_torch_model_scalar.json   ...   今回はエンドポイント側で入力データの標準化をしたいので必要な情報をJSON化して一緒にアップしておく(後述)<br />
</div>
}}

#html(</div>)

** モデルの作成 及び エクスポート [#o29e046c]
#html(<div class="pl10">)

モデルは [[PyTorchで重回帰分析]] で作成したものをそのまま使用。

あとは以下の通り、エクスポートするだけ。
#mycode2(){{
import json
import os

model_name = "sample_torch_model"
if not os.path.exists(model_name):
    os.mkdir(model_name)

# 訓練済みモデルを保存
model_path = f"{model_name}/{model_name}.pth"
#model_state = model.state_dict()
#model_state["my_scaler_params"] = scaler.get_params()
#model_state["my_scaler_mean"] = scaler.mean_
#model_state["my_scaler_var"] = scaler.var_
#model_state["my_scaler_scale"] = scaler.scale_
#torch.save(model_state, model_path)
torch.save(model.state_dict(), model_path)

#
# 標準化に必要な値をJSONに保存
#
scaler_dict = {}
scaler_dict["my_scaler_params"] = scaler.get_params()
scaler_dict["my_scaler_mean"] = scaler.mean_.tolist()
scaler_dict["my_scaler_var"] = scaler.var_.tolist()
scaler_dict["my_scaler_scale"] = scaler.scale_.tolist()
with open(f"{model_name}/{model_name}_scalar.json", "w") as f:
    f.write(json.dumps(scaler_dict))
}}


#html(</div>)

** tar.gz にする [#sf75e15d]
#html(<div class="pl10">)

階層を作りたくなかったので、いったん対象フォルダに移動して同じフォルダのものをアーカイブした。
#myterm2(){{
cd sample_torch_model
tar czfv ../sample_torch_model.tar.gz .
cd ../
}}

#html(</div>)

** S3にアップロード [#ka1b62de]
#html(<div class="pl10">)

バケット作成
#myterm2(){{
aws s3 mb s3://バケット名
}}

s3にアップロード
#myterm2(){{
aws s3api put-object --bucket 作成したバケット名 --key sample_torch_model.tar.gz --body ./sample_torch_model.tar.gz
}}
#html(</div>)

#html(</div>)

* ノートブックインスタンスの作成 [#p7d249a7]
#html(<div class="pl10">)

Amazon SageMaker コンソールから [[ノートブックインスタンス:https://ap-northeast-1.console.aws.amazon.com/sagemaker/home?region=ap-northeast-1#/notebook-instances]] に移動後、[ノートブックインスタンスの作成] を押下して作成する。

今回は以下の通り作成した。
※以下に記載のないものは、全て未選択 または デフォルト値で作成した。

| 設定名 | 設定内容 | 補足 |h
| ノートブックインスタンス名 | sample-notebook1 | |
| ノートブックインスタンスのタイプ | ml.t2.medium | |
| Elastic Inference | 今回は使用しない | 推論をガンガン行う場合は使用する(GPUが使える) |
| ライフサイクル設定 | なし | |
| ボリュームサイズ | 5GB | |
| IAM ロール | [新しいロールの作成]  から 作成 | 「任意の S3 バケット」へのアクセスを許可。 |
| ルートアクセス | 有効化 | |

しばらくするとステータスが [Pending] から [InService] に変わるので、[Jupyter を開く] から jupyter notebook を開く。

[補足]
恐らく IAM ロール ( AmazonSageMakerFullAccess ) が設定されているIAMユーザを使用していれば、ローカルの jupyter でも問題ないと思われる。
※今回は試していない。
#html(</div>)

* デプロイ [#hb0614f9]
#html(<div class="pl10">)

** エントリポイントとなるファイルの作成 [#cf6141ad]
#html(<div class="pl10">)

まずエントリポイントとなるファイルをノートブックインスタンス上に作成する。
解説は後述する事としてまずはコード。

entry_point.py
#mycode2(){{
import argparse
import logging
import sagemaker_containers
import requests

import torch
import torch.nn as nn
import numpy as np
from six import BytesIO
from sklearn.preprocessing import StandardScaler

import os
import io
import json
import glob
import time
import re

logger = logging.getLogger(__name__)
logger.setLevel(logging.DEBUG)

JSON_CONTENT_TYPE = 'application/json'
XNPY_CONTENT_TYPE = 'application/x-npy'
CSV_CONTENT_TYPE  = 'text/csv'

INPUT_SIZE = 2
OUTPUT_SIZE = 1

class LinearRegression(nn.Module):
    """モデル定義"""
    def __init__(self, input_size, output_size):
        super(LinearRegression, self).__init__()
        self.linear = nn.Linear(input_size, output_size)
    def forward(self, x): 
        out = self.linear(x)
        return out

def model_fn(model_dir):
    """モデルのロード."""
    logger.info('START model_fn')
    model = LinearRegression(INPUT_SIZE, OUTPUT_SIZE)
    # モデルのパラメータ設定
    with open(os.path.join(model_dir, 'sample_torch_model.pth'), 'rb') as f:
        model.load_state_dict(torch.load(f))
    # 独自パラメータを設定
    with open(os.path.join(model_dir, 'sample_torch_model_scalar.json')) as f:
        my_state = json.load(f)
        for k,v in my_state.items():
            model.__dict__[k] = v
    logger.info('END   model_fn')
    return model

def input_fn(request_body, content_type=JSON_CONTENT_TYPE):
    """入力データの形式変換."""
    logger.info('START input_fn')
    logger.info(f'content_type: {content_type}')
    logger.info(f'request_body: {request_body}')
    logger.info(f'type: {type(request_body)}')
    if content_type == XNPY_CONTENT_TYPE:
        stream = BytesIO(request_body)
        input_data = np.load(stream)
    elif content_type == CSV_CONTENT_TYPE:
        request_body = request_body.encode("utf-8") if isinstance(request_body, str) else request_body
        input_data = np.loadtxt(BytesIO(request_body), delimiter=",")
    elif content_type == JSON_CONTENT_TYPE:
        input_data = np.array(json.loads(request_body))
    else:
        # TODO: content_typeに応じてデータ型変換
        logger.error(f"content_type invalid: {content_type}")
        input_data = {"errors": [f"content_type invalid: {content_type}"]}
    logger.info('END   input_fn')
    return input_data

def predict_fn(input_data, model):
    """推論."""
    logger.info('START predict_fn')

    if isinstance(input_data, dict) and 'errors' in input_data:
        logger.info('SKIP  predict_fn')
        logger.info('END   predict_fn')
        return input_data
        
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model.to(device)
    model.eval()

    # 説明変数の標準化
    scaler = StandardScaler()
    scaler.set_params(**model.my_scaler_params)
    scaler.mean_ = model.my_scaler_mean
    scaler.var_ = model.my_scaler_var
    scaler.scale_ = model.my_scaler_scale
    scaled_input_data = scaler.transform(input_data)
    converted_input_data = torch.Tensor(scaled_input_data)

    # 推論
    with torch.no_grad():
        logger.info('END   predict_fn')
        return model(converted_input_data.to(device))

def output_fn(prediction, accept=JSON_CONTENT_TYPE):
    """出力データの形式変換."""
    logger.info('START output_fn')
    logger.info(f"accept: {accept}")

    if isinstance(prediction, dict) and 'errors' in prediction:
        logger.info('SKIP  output_fn')
        response = json.dumps(prediction)
        content_type = JSON_CONTENT_TYPE
    elif accept == XNPY_CONTENT_TYPE:
        buffer = BytesIO()
        np.save(buffer, prediction)
        response = buffer.getvalue()
        content_type = XNPY_CONTENT_TYPE
    elif accept == JSON_CONTENT_TYPE:
        response = json.dumps({"results": [prediction.data[i].item() for i in range(len(prediction.data))]})
        content_type = JSON_CONTENT_TYPE
    else:
        # TODO: コンテンツタイプに応じて変換
        response = json.dumps({"results": [prediction.data[i].item() for i in range(len(prediction.data))]})
        content_type = JSON_CONTENT_TYPE

    logger.info('END   output_fn')
    return response, content_type


if __name__ == '__main__':
    # 訓練してからデプロイする場合はここで行う
    logger.info("process main!")
    pass
}}
#html(</div>)


** エントリポイントの解説 [#b8216f4a]
#html(<div class="pl10">)

参考
https://sagemaker.readthedocs.io/en/stable/using_pytorch.html#load-a-model
https://sagemaker.readthedocs.io/en/stable/using_pytorch.html#serve-a-pytorch-model

*** 実装する必要がある関数 [#j3b99a93]
#html(<div class="pl10">)

上記でエントリポイントに指定したファイルには、以下の関数を含める必要がある。

| 関数名 | 説明 |h
| model_fn | モデルのロードを行う関数 |
| input_fn | 入力データの形式変換を行う関数 |
| predict_fn | 推論を行う関数 |
| output_fn | 結果を呼び出し元に返却する時に形式変換を行う関数 |
#html(</div>)

*** モデルのロード ( model_fn ) [#zf621a38]
#html(<div class="pl10">)
model_fn ではモデルのロードを行う。
やる事は、引数で渡ってくるモデルディレクトリ配下のモデルを利用する形式に合わせてロードするだけ。
モデルの他に別のデータ (今回でいうと標準化用のJSON) を含めてアップロードした場合は、ここで利用する事ができる。
#html(</div>)

*** 入力データの形式変換 ( input_fn ) [#v87bb0fa]
#html(<div class="pl10">)
input_fn では入力データの形式変換を行う。

引数のコンテンツタイプから入力データの形式を判別して、推論用のデータ型に変換する作業を行う事になる。
尚、今回試した３つの形式( numpy.ndarray、csv、json ) の場合、引数のデータは全て bytearray で渡ってくる為、
bytearray  を numpy.ndarray に変換する作業を行った。

今回は PyTorch を使用しているので、ここで torch.Tensor まで変換しておいてもよかったのだが、
標準化用の独自パラメータをモデルに引っ付けており、ここでは標準化まで行う事ができなかったので、いったん numpy.ndarray までの変換とした。
※引数にモデルがない為。

#html(</div>)

*** 推論の実行 ( predict_fn ) [#pdbd8342]
#html(<div class="pl10">)
input_fn で形式変換したデータを入力として実際の推論を行う。
エンドポイントのインスタンスタイプによっては GPU が利用できない為、GPU/CPU どちらでも推論ができるように実装しておく。

今回は、都合上ここで標準化 及び torch.Tensor  への変換まで行った。
#html(</div>)

*** 出力データの形式変換 ( output_fn ) [#h1ca6bc6]
#html(<div class="pl10">)

呼び出し元で指定された accept の値に応じて、出力形式の変換を行う。
今回は、サンプルとして numpy.ndarray、csv、json の３つだけ実装してみた。

#html(</div>)

*** 推論時の処理の流れ [#z7e3c7d1]
#html(<div class="pl10">)

推論時の処理は以下の通り実行される。

#mycode2(){{
input_object = input_fn(request_body, request_content_type)

prediction = predict_fn(input_object, model)

output = output_fn(prediction, response_content_type)
}}

#html(</div>)


*** その他 [#k53069a2]
#html(<div class="pl10">)

ここらへんを見る限り、他の依存ライブラリ等がある場合にはモデルデータに含めてアップロードする事ができる模様。
https://sagemaker.readthedocs.io/en/stable/using_pytorch.html#optional-arguments
#html(</div>)


#html(</div>)


** エンドポイントの作成、デプロイ [#f8f82725]
#html(<div class="pl10">)

ノートブックインスタンス上から以下を実行する。
#mycode2(){{
# エンドポイントの作成、デプロイ
sagemaker_session = sagemaker.Session()
role = sagemaker.get_execution_role()

# モデルの作成
pytorch_model = PyTorchModel(model_data="s3://バケット名/sample_torch_model.tar.gz",
                             role=role,
                             framework_version='1.3.1',
                             entry_point="entry_point.py")
# デプロイパラメータ
deploy_params = {
    'instance_type'          : 'ml.t2.medium'  # お試し用 (https://aws.amazon.com/jp/sagemaker/pricing/instance-types/ )
    ,'initial_instance_count' : 1              # お試し用
    #,'endpoint_name'          : 'sample-torch-model4'  # エンドポイント名を指定してのデプロイが何故かできない
}

# デプロイ
predictor = pytorch_model.deploy(**deploy_params)
}}
https://sagemaker.readthedocs.io/en/stable/sagemaker.pytorch.html#sagemaker.pytorch.model.PyTorchModel

#html(</div>)

#html(</div>)

* デプロイしたエンドポイントを使って推論してみる [#cab537f1]
#html(<div class="pl10">)

#mycode2(){{
import pandas as pd

# 入力データ ([部屋の広さ, 築年数])
input_data = [[60.0, 10.0], [50.0, 10.0], [40.0, 10.0]]

# 推論
predict_data = np.array(input_data)
results = predictor.predict(predict_data)

# 結果表示
result_df = pd.DataFrame(results, columns=["家賃(万円)"])
result_df["広さ(㎡)"] = predict_data[:,0]
result_df["築年数"] = predict_data[:,1]
result_df
}}

結果
#html(){{
<style scoped="">
.dataframe {
    border: none;
    border-collapse: collapse;
    border-spacing: 0;
    color: black;
    font-size: 14px;
    table-layout: fixed;
}
.dataframe tbody tr th:only-of-type {
    vertical-align: middle;
}
.dataframe tbody tr th {
    vertical-align: top;
    padding: 4px;
}
.dataframe thead th {
    text-align: right;
    padding: 4px;
}
</style>
<table border="1" class="dataframe">
  <thead>
    <tr style="text-align: right;">
      <th></th>
      <th>家賃(万円)</th>
      <th>広さ(㎡)</th>
      <th>築年数</th>
    </tr>
  </thead>
  <tbody>
    <tr>
      <th>0</th>
      <td>8.117216</td>
      <td>60.0</td>
      <td>10.0</td>
    </tr>
    <tr>
      <th>1</th>
      <td>7.191902</td>
      <td>50.0</td>
      <td>10.0</td>
    </tr>
    <tr>
      <th>2</th>
      <td>6.266588</td>
      <td>40.0</td>
      <td>10.0</td>
    </tr>
  </tbody>
</table>
}}


#html(</div>)

* Lambdaなどからエンドポイントを利用する [#i0483a0d]
#html(<div class="pl10">)

#mycode2(){{
#
# sage maker以外からエンドポイントを利用して推論
#
import boto3
import json

# 入力データ ([部屋の広さ, 築年数])
input_data = [[60.0, 10.0], [50.0, 10.0], [40.0, 10.0]]

# エンドポイント名
endpoint_name = "pytorch-inference-2020-02-28-12-35-37-541"

# JSONを送信する場合
request_body = json.dumps(input_data)
content_type = "application/json"
accept_type  = "application/json"

# CSVを送信する場合
#request_body = '\n'.join([','.join([str(x) for x in rec]) for rec in input_data])
#content_type = "text/csv"
#accept_type  = "application/json"

# 推論
client = boto3.client('sagemaker-runtime')
response = client.invoke_endpoint(
    EndpointName=endpoint_name,
    Body=request_body,
    ContentType=content_type,
    Accept=accept_type
)

# 結果表示
print("### response (Body以外)###")
print(json.dumps({k:v for k,v in response.items() if k != 'Body'}, indent=4))
print("### response (Body) ###")
response_dict = json.loads(response['Body'].read().decode("utf-8"))
print(json.dumps(response_dict, indent=4))
}}

結果
#mycode3(){{
### response (Body以外)###
{
    "ResponseMetadata": {
        "RequestId": "f5cca038......",
        "HTTPStatusCode": 200,
        "HTTPHeaders": {
            "x-amzn-requestid": "f5cca038......",
            "x-amzn-invoked-production-variant": "AllTraffic",
            "date": "Sat, 29 Feb 2020 XX:XX:XX GMT",
            "content-type": "application/json",
            "content-length": "69"
        },
        "RetryAttempts": 0
    },
    "ContentType": "application/json",
    "InvokedProductionVariant": "AllTraffic"
}
### response (Body) ###
{
    "results": [
        8.117216110229492,
        7.191902160644531,
        6.26658821105957
    ]
}
}}
#html(</div>)


* 後片付け [#j5efc1d5]
#html(<div class="pl10">)

ノートブックインスタンスから以下を実行する事でエンドポイントの削除が可能。
#mycode2(python){{
sagemaker.Session().delete_endpoint(predictor.endpoint)
}}

#html(</div>)
訓練済みモデルをSageMakerエンドポイントにデプロイする の変更点 - 闘うITエンジニアの覚え書き

訓練済みモデルをSageMakerエンドポイントにデプロイするの変更点 - 闘うITエンジニアの覚え書き
