Amazon SageMaker 入門: 組み込みアルゴリズムを使う

SageMaker Studioと組み込みアルゴリズムを使って、Irisデータセットでk-NN分類器を学習・デプロイします。

Takahiro Iwasa

2022-01-24

10 min read

機械学習向けの統合開発環境であるSageMaker Studioを使って、SageMakerの組み込みK近傍法（k-NN）アルゴリズムを、広く使われているIrisデータセットに適用する方法を説明します。

SageMakerのオンボーディング

まず、SageMaker Studioのオンボーディングプロセスを完了する必要があります。

Quick SetupとCustom Setupからセットアップ方法を選択します。

VPCを選択します。

自動的に作成されたユーザーのメニューから、SageMaker Studioを起動します。

起動すると、SageMaker Studioが表示されます。

SageMaker Studio中央のNotebookボタンをクリックします。

前処理

データセットの準備

まず環境変数を設定します。

以下を実際の値に置き換えてください。

<YOUR_S3_BUCKET>
<YOUR_SAGEMAKER_ROLE>

%env S3_DATASET_BUCKET=<YOUR_S3_BUCKET>
%env S3_DATASET_TRAIN=knn/input/iris_train.csv
%env S3_DATASET_TEST=knn/input/iris_test.csv
%env S3_TRAIN_OUTPUT=knn/output
%env SAGEMAKER_ROLE=<YOUR_SAGEMAKER_ROLE>

次に、以下のPythonのimport文を含むセルを作成します。Python3 Data Scienceインスタンスには、これらのライブラリがあらかじめインストールされています。

1
import os
2
import random
3
import string
4

5
import boto3
6
import matplotlib.pyplot as plt
7
import pandas as pd
8
import sagemaker
9
from IPython.display import display
10
from sagemaker import image_uris
11
from sagemaker.deserializers import JSONDeserializer
12
from sagemaker.estimator import Estimator, Predictor
13
from sagemaker.inputs import TrainingInput
14
from sagemaker.serializers import CSVSerializer
15
from sklearn.model_selection import train_test_split

定数と変数を定義します。

1
# Define constants
2
CSV_PATH = './tmp/iris.csv'
3
S3_DATASET_BUCKET = os.getenv('S3_DATASET_BUCKET')
4
S3_DATASET_TRAIN = os.getenv('S3_DATASET_TRAIN')
5
S3_DATASET_TEST = os.getenv('S3_DATASET_TEST')
6
S3_TRAIN_OUTPUT = os.getenv('S3_TRAIN_OUTPUT')
7
SAGEMAKER_ROLE = os.getenv('SAGEMAKER_ROLE')
8
ESTIMATOR_INSTANCE_COUNT = 1
9
ESTIMATOR_INSTANCE_TYPE = 'ml.m5.large'
10
PREDICTOR_INSTANCE_TYPE = 'ml.t2.medium'
11
PREDICTOR_ENDPOINT_NAME = f'sagemaker-knn-{PREDICTOR_INSTANCE_TYPE}'.replace('.', '-')
12

13
# Define variables
14
bucket = boto3.resource('s3').Bucket(S3_DATASET_BUCKET)
15
train_df = None
16
test_df = None
17
train_object_path = None
18
test_object_path = None
19
knn = None
20
predictor = None

AWSのSageMaker ExamplesリポジトリからIrisデータセットをダウンロードします。このデータセットは、そのシンプルさと明確に定義された構造から、分類モデルのデモンストレーションに広く使われています。Irisデータセットの属性や用途など詳細については、公式ページを参照してください。

!mkdir -p tmp
!curl -o "$(pwd)/tmp/iris.csv" -L https://raw.githubusercontent.com/aws/amazon-sagemaker-examples/master/hyperparameter_tuning/r_bring_your_own/iris.csv

ダウンロード後、以下のコードでCSVを読み込み、前処理を行います。この関数は、Irisデータセットを組み込みアルゴリズムでの学習に使えるよう、SageMakerの要件に合わせて処理します。

詳細はSageMakerのドキュメントを参照してください。

1
def load_csv(path: str) -> pd.DataFrame:
2
    # Load the CSV into a Pandas DataFrame
3
    df = pd.read_csv(path)
4
    # Move the label column ('Species') to the first position
5
    df = df[['Species', 'Sepal.Length', 'Sepal.Width', 'Petal.Length', 'Petal.Width']]
6
    # Convert target labels ('Species') to integers
7
    df['Species'] = df['Species'].map({'setosa': 0, 'versicolor': 1, 'virginica': 2})
8
    return df

🔥 Caution

SageMakerではCSVの最初の列がターゲットラベル（クラス）である必要があるため、Species列を先頭に移動する必要があります。
ターゲットラベル（品種名）も、互換性のために整数へ変換する必要があります。

データセットの可視化

データセットの特徴量と分布を把握するため、散布図を作成します。

1
def plot(df: pd.DataFrame) -> None:
2
    pd.plotting.scatter_matrix(df, figsize=(15, 15), c=df['Species'])
3
    plt.show()

生成された散布図の軸は次のように対応しています。

X軸
- 左から右に向かってSpecies、Sepal.Length、Sepal.Width、Petal.Length、Petal.Widthを表します。
Y軸
- 下から上に向かってPetal.Width、Petal.Length、Sepal.Width、Sepal.Length、Speciesを表します。

プロットを見ると、データポイントがはっきりとしたグループに分かれて分類されていることから、特徴量から品種を予測できることが読み取れます。

データセットのS3へのアップロード

前処理済みのデータセットをS3にアップロードします。

1
def upload_csv_to_s3(df: pd.DataFrame, object_path: str) -> str:
2
    filename = ''.join([random.choice(string.digits + string.ascii_lowercase) for i in range(10)])
3
    path = os.path.abspath(os.path.join('./tmp', filename))
4
    df.to_csv(path, header=False, index=False)
5
    # Change content-type because the default is binary/octet-stream
6
    bucket.upload_file(path, object_path, ExtraArgs={'ContentType': 'text/csv'})
7
    return f's3://{bucket.name}/{object_path}'

前処理の実行

上記の前処理を完了するには、以下を実行します。

1
if __name__ == '__main__':
2
    df = load_csv(CSV_PATH)
3
    display(df)
4
    plot(df)
5
    train_df, test_df = train_test_split(df, shuffle=True, random_state=0)
6
    train_object_path = upload_csv_to_s3(train_df, S3_DATASET_TRAIN)
7
    test_object_path = upload_csv_to_s3(test_df, S3_DATASET_TEST)

学習

k-NN推定器を設定し、学習プロセスを開始します。

1
def get_estimator(**hyperparams) -> Estimator:
2
    estimator = Estimator(
3
        image_uri=image_uris.retrieve('knn', boto3.Session().region_name),
4
        role=SAGEMAKER_ROLE,
5
        instance_count=ESTIMATOR_INSTANCE_COUNT,
6
        instance_type=ESTIMATOR_INSTANCE_TYPE,
7
        input_mode='Pipe',
8
        output_path=f's3://{S3_DATASET_BUCKET}/{S3_TRAIN_OUTPUT}',
9
        sagemaker_session=sagemaker.Session(),
10
    )
11
    hyperparams.update({'predictor_type': 'classifier'})
12
    estimator.set_hyperparameters(**hyperparams)
13
    return estimator
14

15

16
def train(estimator: Estimator, train_object_path: str, test_object_path: str) -> None:
17
    train_input = TrainingInput(train_object_path, content_type='text/csv', input_mode='Pipe')
18
    test_input = TrainingInput(test_object_path, content_type='text/csv', input_mode='Pipe')
19
    estimator.fit({'train': train_input, 'test': test_input})
20

21

22
if __name__ == '__main__':
23
    knn = get_estimator(k=1, sample_size=1000)
24
    train(knn, train_object_path, test_object_path)

学習ジョブを開始すると、以下のようなログが表示されます。

1
2022-01-08 13:38:34 Starting - Starting the training job...
2
2022-01-08 13:38:57 Starting - Launching requested ML instancesProfilerReport-1641649113: InProgress
3
......
4
[01/08/2022 13:43:00 INFO 140667182901056] #test_score (algo-1) : ('accuracy', 0.9736842105263158)
5
[01/08/2022 13:43:00 INFO 140667182901056] #test_score (algo-1) : ('macro_f_1.000', 0.97170347)

ログにはaccuracy（精度）やmacro F1 scoreといった指標が含まれており、テストデータセットに対するモデルの性能を把握できます。
学習時にtrainとtestの両方のチャンネルを活用することで、組み込みアルゴリズムが自動的にモデルの汎化性能を評価します。

ECR URI

image_uri（3行目）は、AWSが提供するk-NN学習アルゴリズムのECRコンテナURIを指定しています。

組み込みアルゴリズムのコンテナURIに関する詳細情報は、公式ドキュメントを参照してください。

チャンネル名

SageMakerの組み込みアルゴリズムでは、チャンネル名（18行目）はtrainに固定されています。学習ジョブの作成時にtestチャンネルを含めると、学習後に自動的にテストデータでMLモデルが評価されます。

Pipeモードの使用

データストリーミングの効率を高めるには、TrainingInput定義のinput_mode（7行目）パラメータを"Pipe"に設定することで、Pipeモードを有効化できます。Pipeモードは、S3からSageMakerインスタンスへデータを直接ストリーミングするため、データセット全体をダウンロードする場合と比べてレイテンシとメモリ要件を削減できます。

k-NNのハイパーパラメータ

k-NNアルゴリズムには、設定可能なハイパーパラメータがいくつか用意されています。使い方や効果の詳細については、公式ドキュメントを参照してください。

推論

学習済みモデルをエンドポイントにデプロイし、予測結果を検証します。

SageMakerのserializerとdeserializerは、デプロイされた推論エンドポイントとやり取りする際の、入出力データの形式を指定するために使われます。

Serializer
- CSVSerializer: データをCSV形式に変換します。
- JSONSerializer: データをJSON形式に変換します。
- NumpySerializer: NumPy配列をバイナリ形式に変換します。
Deserializer
- JSONDeserializer: JSONレスポンスをPythonの辞書やリストに変換します。
- BytesDeserializer: 生のバイト列をそのまま返します。

1
def deploy(estimator: Estimator) -> Predictor:
2
    return estimator.deploy(
3
        initial_instance_count=1,
4
        instance_type=PREDICTOR_INSTANCE_TYPE,
5
        serializer=CSVSerializer(),
6
        deserializer=JSONDeserializer(),
7
        endpoint_name=PREDICTOR_ENDPOINT_NAME,
8
    )
9

10

11
def validate(predictor: Predictor, test_df: pd.DataFrame) -> pd.DataFrame:
12
    rows = []
13
    for _, data in test_df.iterrows():
14
        predict = predictor.predict(
15
            pd.DataFrame([data.drop('Species')]).to_csv(header=False, index=False),
16
            initial_args={'ContentType': 'text/csv'},
17
        )
18
        predicted_label = predict['predictions'][0]['predicted_label']
19
        row = data.tolist()
20
        row.append(predicted_label)
21
        row.append(data['Species'] == predicted_label)
22
        rows.append(row)
23
    return pd.DataFrame(rows, columns=('Species', 'Sepal.Length', 'Sepal.Width', 'Petal.Length', 'Petal.Width', 'Prediction', 'Result'))
24

25

26
if __name__ == '__main__':
27
    predictor = deploy(knn)
28
    predictions = validate(predictor, test_df)
29
    display(predictions)

推論結果には、Prediction列とResult列が含まれます。

Prediction: モデルに与えられた各サンプルに対する予測ラベルです。
Result: 予測結果が実際のラベルと一致するかどうかを示すブール値（TrueまたはFalse）です。

後片付け

学習済みモデルとそのエンドポイントは、以下で削除できます。

1
def delete_model(predictor: Predictor) -> None:
2
    predictor.delete_model()
3

4

5
def delete_endpoint(predictor: Predictor) -> None:
6
    predictor.delete_endpoint(delete_endpoint_config=True)
7

8

9
if __name__ == '__main__':
10
    delete_model(predictor)
11
    delete_endpoint(predictor)

まとめ

SageMaker StudioでIrisデータセットを使ってk-NN分類器を学習・デプロイしたところ、学習スクリプトやコンテナを自作することなく、動作する推論エンドポイントを構築できました。ここで役立っているのは、image_uris.retrieve('knn', ...)による学習用イメージの取得と、input_mode='Pipe'の設定です。Pipeモードにより、データセット全体を先にダウンロードするのではなく、S3のCSVデータをインスタンスへ直接ストリーミングできます。estimator.fit()にtrainとtestの両方のチャンネルを渡すことで、SageMakerによる自動評価がトリガーされ、追加の検証コードなしに学習ログに現れるaccuracyやmacro_f_1.000といった指標が得られます。ここで得られた97%という精度は、Irisの各品種がすでに特徴量空間の中できれいに分離されていることを反映したものです。データセットの難易度に関わらず、デプロイしたエンドポイントは稼働している間ずっと時間単位で課金され続けるため、使い終わったら忘れずに削除しておく価値があります。