<jupyter_start><jupyter_text>Amazon SageMaker Model Monitor の機能を使って PyTorch の MNIST 分類モデルの推論入出力をキャプチャする[Amazon SageMaker Model Monitor](https://docs.aws.amazon.com/sagemaker/latest/dg/model-monitor.html) は推論エンドポイントにデプロイされたモデルの性能を監視するためのサービスですが、2022年1月現在、ほとんどの機能はテーブルデータのみの対応です。しかし、推論リクエスト時の入出力データをキャプチャする機能はテーブルデータでなくても利用することができるので、PyTorch の MNIST 分類モデルをサンプルに使い方をご紹介します。キャプチャした入力データを使って Ground Truth データ（正解データ）を作り、キャプチャした出力データ（推論結果）と比較すれば、正解率などのメトリクスを算出することができます。 Table of Contents1&nbsp;&nbsp;SDK の更新2&nbsp;&nbsp;セットアップ3&nbsp;&nbsp;学習データの準備4&nbsp;&nbsp;デプロイするモデルの学習5&nbsp;&nbsp;モデルのデプロイ6&nbsp;&nbsp;データキャプチャを試す7&nbsp;&nbsp;エンドポイントの削除  SDK の更新最新の Amazon SageMaker Python SDK と AWS SDK for Python をインストールします。<jupyter_code>import sys
!{sys.executable} -m pip install --upgrade pip --quiet
!{sys.executable} -m pip install -U awscli sagemaker boto3 --quiet
import sagemaker
sagemaker.__version__<jupyter_output><empty_output><jupyter_text>セットアップ使用するモジュールのインポートやパラメタの設定などを行います。<jupyter_code># Python Built-Ins:
from datetime import datetime
import os
import json
import logging
from tempfile import TemporaryFile
import time

# External Dependencies:
import boto3
from botocore.exceptions import ClientError
import numpy as np
import sagemaker
from sagemaker.multidatamodel import MultiDataModel
from sagemaker.pytorch import PyTorch

sagemaker_session = sagemaker.Session()
role = sagemaker.get_execution_role()

boto_session = boto3.session.Session()
sagemaker_client = boto_session.client("sagemaker")
sagemaker_runtime = boto_session.client("sagemaker-runtime")
region = boto_session.region_name
account_id = boto3.client('sts').get_caller_identity().get('Account')

# Configuration:
bucket_name = sagemaker_session.default_bucket()
prefix = "mnist"
output_path = f"s3://{bucket_name}/{prefix}"
data_capture_prefix = f"{prefix}/monitoring/datacapture"
s3_capture_upload_path = f"s3://{bucket_name}/{data_capture_prefix}"
# output_path = f"s3://{bucket_name}/{prefix}"
# data_capture_prefix = f"{prefix}/monitoring/datacapture"
# s3_capture_upload_path = f"s3://{bucket_name}/{data_capture_prefix}"
import os, json
NOTEBOOK_METADATA_FILE = "/opt/ml/metadata/resource-metadata.json"
if os.path.exists(NOTEBOOK_METADATA_FILE):
    with open(NOTEBOOK_METADATA_FILE, "rb") as f:
        metadata = json.loads(f.read())
        domain_id = metadata.get("DomainId")
        on_studio = True if domain_id is not None else False
print("Is this notebook runnning on Studio?: {}".format(on_studio))<jupyter_output><empty_output><jupyter_text>学習データの準備<jupyter_code>!aws s3 cp s3://fast-ai-imageclas/mnist_png.tgz . --no-sign-request
if on_studio:
    !tar -xzf mnist_png.tgz -C /opt/ml --no-same-owner
else:
    !tar -xvzf  mnist_png.tgz
from torchvision import datasets, transforms
from torch.utils.data import DataLoader
import torch
import os

root_dir_studio = '/opt/ml'
data_dir = os.path.join(root_dir_studio,'data') if on_studio else 'data'
training_dir = os.path.join(root_dir_studio,'mnist_png/training') if on_studio else 'mnist_png/training'
test_dir = os.path.join(root_dir_studio,'mnist_png/testing') if on_studio else 'mnist_png/testing'

os.makedirs(data_dir, exist_ok=True)

training_data = datasets.ImageFolder(root=training_dir,
                            transform=transforms.Compose([
                            transforms.Grayscale(),
                            transforms.ToTensor(),
                            transforms.Normalize((0.1307,), (0.3081,))]))
test_data = datasets.ImageFolder(root=test_dir,
                            transform=transforms.Compose([
                            transforms.Grayscale(),
                            transforms.ToTensor(),
                            transforms.Normalize((0.1307,), (0.3081,))]))

training_data_loader = DataLoader(training_data, batch_size=len(training_data))
training_data_loaded = next(iter(training_data_loader))
torch.save(training_data_loaded, os.path.join(data_dir, 'training.pt'))

test_data_loader = DataLoader(test_data, batch_size=len(test_data))
test_data_loaded = next(iter(test_data_loader))
torch.save(test_data_loaded, os.path.join(data_dir, 'test.pt'))<jupyter_output><empty_output><jupyter_text>以下のセルを実行して、学習データを Amazon S3 にアップロードします。<jupyter_code>inputs = sagemaker_session.upload_data(path=data_dir, bucket=bucket_name, key_prefix=os.path.join(prefix, 'data'))
print('input spec (in this case, just an S3 path): {}'.format(inputs))<jupyter_output><empty_output><jupyter_text>以下のセルを実行して、モデルをデプロイした後の推論テストに使う画像を確認します。データセットからランダムjに5枚の画像をピックアップしています。<jupyter_code>%matplotlib inline
import random
import numpy as np
import matplotlib.pyplot as plt

raw_test_data = datasets.ImageFolder(root=test_dir,
                                        transform=transforms.Compose([
                                        transforms.Grayscale(),
                                        transforms.ToTensor()]))
num_samples = 5
indices = random.sample(range(len(raw_test_data) - 1), num_samples)
raw_images = np.array([raw_test_data[i][0].numpy() for i in indices])
raw_labels = np.array([raw_test_data[i][1] for i in indices])


for i in range(num_samples):
    plt.subplot(1,num_samples,i+1)
    plt.imshow(raw_images[i].reshape(28, 28), cmap='gray')
    plt.title(raw_labels[i])
    plt.axis('off')
    
images = np.array([test_data[i][0].numpy() for i in indices])<jupyter_output><empty_output><jupyter_text>デプロイするモデルの学習MNIST 分類モデルを学習します。<jupyter_code># from sagemaker import image_uris

# # Specify an AWS container image and region as desired
# container = image_uris.retrieve(
#     region=region, framework="pytorch", 
#     version="1.8.1", instance_type='ml.c5.xlarge', image_scope='training', py_version='py36')
# container
estimator = PyTorch(
        entry_point="train.py",
        source_dir="code",  # directory of your training script
        role=role,
        framework_version="1.8.0",
        py_version="py3",
        instance_type="ml.c4.xlarge",
        instance_count=1,
#         output_path=output_path,
        hyperparameters={
                        'batch-size':128,
                        'lr': 0.01,
                        'epochs': 10,
                        'backend': 'gloo'
                    },
    )
estimator.fit({'training': inputs})<jupyter_output><empty_output><jupyter_text>モデルのデプロイ学習したモデルを推論エンドポイントにデプロイします。推論エンドポイントが InService になるまで数分かかります。デプロイの際にデータキャプチャ設定を `data_capture_config` に設定します。このサンプルでは、トラフィックの100%をキャプチャするよう設定しています。キャプチャされたデータは `destination_s3_uri` で指定した S3 パスに保存されます。<jupyter_code>from time import gmtime, strftime
from sagemaker.model_monitor import DataCaptureConfig

endpoint_name = "DEMO-pytorch-mnist-model-monitor-" + strftime(
    "%Y-%m-%d-%H-%M-%S", gmtime()
)
print(endpoint_name)

model = estimator.create_model(role=role, source_dir="code", entry_point="inference.py")

predictor = model.deploy(
    initial_instance_count=1,
    instance_type="ml.m5.xlarge",
    endpoint_name=endpoint_name,
    data_capture_config=DataCaptureConfig(
        enable_capture=True, sampling_percentage=100, destination_s3_uri=s3_capture_upload_path
    ),
)<jupyter_output><empty_output><jupyter_text>データキャプチャを試す推論エンドポイントが起動したら、テストデータを使って推論を実行します。<jupyter_code>prediction = predictor.predict(images)
predicted_label = prediction.argmax(axis=1)

print('The GT labels are: {}'.format(raw_labels))
print('The predicted labels are: {}'.format(predicted_label))<jupyter_output><empty_output><jupyter_text>キャプチャされたデータはエンドポイント作成時に指定したパスの下の `エンドポイント名/AllTraffic/yyyy/mm/dd/hh/` に JSONL 形式で保存されます。以下のセルを実行して、保存された JSONL ファイルのリストを表示します。<jupyter_code>s3_client = boto3.Session().client("s3")
result = s3_client.list_objects(Bucket=bucket_name, Prefix=data_capture_prefix)
capture_files = [capture_file.get("Key") for capture_file in result.get("Contents")]
print("Found Capture Files:")
print("\n ".join(capture_files))<jupyter_output><empty_output><jupyter_text>以下のセルを実行して、JSONL ファイルの中を見てみましょう。`"encoding": "BASE64"` と書かれていることから、base64 形式にエンコードされていることがわかります。<jupyter_code>import json

def get_obj_body(obj_key):
    return s3_client.get_object(Bucket=bucket_name, Key=obj_key).get("Body").read().decode("utf-8")

capture_file = json.loads(get_obj_body(capture_files[-1]))

print(json.dumps(capture_file, indent=2))<jupyter_output><empty_output><jupyter_text>以下のセルを実行して、キャプチャデータをデコードして表示します。このサンプルは推論の入力形式が Numpy ですが、上記セルで確認したようにキャプチャされたデータは base64 形式にデコードされて JSONL ファイルに記載されています。入力データと出力データ（推論結果）をデコードして表示しています。<jupyter_code>import base64
import numpy as np
from io import BytesIO

input_npy = base64.b64decode(capture_file['captureData']['endpointInput']['data'].encode())
output_npy = base64.b64decode(capture_file['captureData']['endpointOutput']['data'].encode())

input_data = np.load(BytesIO(input_npy))
output_data = np.load(BytesIO(output_npy))

print('---- input data ---- size:', np.shape(input_data))
print(input_data)
print('---- output data ---- size:', np.shape(output_data))
print(output_data)
predicted_label = output_data.argmax(axis=1)

print('The GT labels are: {}'.format(raw_labels))
print('The predicted labels are: {}'.format(predicted_label))<jupyter_output><empty_output><jupyter_text>エンドポイントの削除以下のセルを実行して、作成したエンドポイントを削除します。エンドポイントは明示的に削除しない限り課金が発生するので、不要になったら必ず削除してください。<jupyter_code>predictor.delete_endpoint(delete_endpoint_config=True)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11873860345382388, 'fraction_numerical': 0.012442346884050198, 'mean_word_length': 5.4170681348933245, 'pattern_counts': {'":': 1, '<': 49, '<?xml version=': 0, '>': 49, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 3, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '477497', 'license': 'MIT', 'n_tokens_mistral': 3952, 'n_tokens_neox': 3784, 'n_words': 485, 'path': 'sagemaker-model-monitor/data-capture-with-non-table-data/model-monitor-pytorch-capture-inference.ipynb', 'repo_name': 'ohbuchim/sample-codes-for-aiml', 'size': '9323'}