Amazon SageMaker 入門: 機械学習のための組み込みアルゴリズム

岩佐 孝浩

2022年1月24日

19 min read

Machine Learning SageMaker

「Amazon SageMaker 入門 〜組み込みアルゴリズムで機械学習を試してみよう〜」という勉強会を開催しました。

このブログ記事では、そのプレゼンテーション内容を共有し、Amazon SageMaker の強力な機能について解説します。使用したサンプルコードは GitHub リポジトリ からアクセスできます。

この投稿は次のような読者を対象としています。

• Amazon SageMaker に興味がある。
• 機械学習や AWS の基本的な知識を持っている。

• 機械学習と SageMaker エコシステムの概要を提供する。
• SageMaker の組み込みアルゴリズムを使用した教師あり機械学習をデモする。

教師あり学習は、分類や回帰タスクによく利用されます。ラベル付きデータを用いるため、データ準備のコストが増加する可能性があります。

教師なし学習は、クラスタリングや次元削減といったタスクに焦点を当てています。ラベルなしデータを使用しますが、推論結果の解釈性に欠けることがあります。

強化学習は、統計的手法と心理学的アプローチを組み合わせ、報酬を最大化する行動に焦点を当てます。トレーニングには報酬メカニズムや適切なデータセットが必要です。

深層学習は、多層ニューラルネットワークを利用して人間を超える精度を実現します。特徴選択が自動化される一方で、結果の解釈性が制限されることがあります。また、大量の計算リソースを必要とします。

ニューラルネットワークの複雑さは、重みの合計によって測定できます。以下は複雑さを 24 と計算する例です。

• 4 (特徴量) × 3 (隠れ層 1) = 12
• 3 (隠れ層 1) × 3 (隠れ層 2) = 9
• 3 (隠れ層 2) × 1 (出力) = 3
• 合計: 12 + 9 + 3 = 24

特徴量はデータの属性を表します。行は サンプル または データポイント と呼ばれ、列は 特徴量 です。スケーリング、エンコーディング、前処理などの特徴量エンジニアリングは、精度を向上させます。

k-Fold クロスバリデーション

この方法ではデータを複数のサブセットに分割してトレーニングとテストを行い、汎化性能を高め、過学習のリスクを軽減します。

混同行列

混同行列は、精度 (accuracy)、適合率 (precision)、再現率 (recall)、および F値 (f-score) などの指標を提供し、分類モデルを包括的に評価します。

Amazon SageMaker は 機械学習のためのマネージドサービス であり、ML ライフサイクルを簡略化します。SageMaker は 2021 年 12 月現在、17 の組み込みアルゴリズム をサポートしており、BYOM (Bring Your Own Model) 機能も提供します。

以下の図は、データ準備からモデルのデプロイと監視に至るまで、機械学習ワークフローに関連する AWS サービスを示しています。

Amazon SageMaker は、特定の要件に基づいて機械学習モデルを提供するためのいくつかのエンドポイントタイプを提供します。

• SageMaker ホスティングサービス: 永続的なエンドポイントを提供し、EC2 インスタンスと同様にアクティブな状態を維持します。リアルタイム推論において最小のレイテンシを確保します。
• SageMaker サーバーレスエンドポイント (プレビュー): 使用時のみ料金が発生するコスト効率の良いオプションです。ただし、アイドル状態のときにコールドスタート遅延が発生します。
• 非同期推論: バッチ処理や頻度の低いリクエストに最適です。SageMaker はアクティブなリクエストがない場合、自動的にインスタンス数をゼロにスケールダウンし、コストを大幅に削減します。

それぞれのエンドポイントタイプは異なるユースケースに対応しており、効率的なモデル展開を可能にします。

Amazon SageMaker は、異なるユーザー要件に対応する複数の環境を提供しています。この投稿では主に、機械学習のための IDE である SageMaker Studio に焦点を当てます。以下は利用可能な環境オプションです。

• SageMaker Studio / RStudio on SageMaker:

  • 包括的な機械学習 IDE。
  • 学習や迅速な実験に適した SageMaker JumpStart などのツールを含みます。

• SageMaker Notebook Instances:

  • スタンドアロンの Jupyter Notebook インスタンス。
  • 少数のデータサイエンティストによる一時的またはアドホックな使用に最適です。

• ローカル環境 + AWS SDK + SageMaker SDK:

  • 開発者がローカル環境を使用して SageMaker と対話できるようにします。
  • 柔軟性を必要とする小規模な開発に適しています。

• SageMaker Studio Lab:

  • AWS アカウントなしで SageMaker Studio の一部機能に無料でアクセスできます。
  • 機械学習の概念を探求する初心者に最適です。

それぞれのオプションには独自の利点があり、プロジェクトのニーズに基づいて最適な環境を選択できます。

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

ステップ 1: セットアップタイプの選択

• クイックスタートには Quick Setup を選択します。
• 高度な構成を必要とする本番環境では Custom Setup を選択します。

ステップ 2: VPC の選択

オンボーディングプロセス中に、セットアップに必要な任意の VPC を選択します。

ステップ 3: SageMaker Studio の起動

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

ステップ 4: SageMaker Studio へのアクセス

起動後、SageMaker Studio のインターフェースが表示されます。

クイックセットアップ

Quick Setup は、すぐに始めるための簡単な方法です。公式ドキュメントで詳細をご覧ください: クイックセットアップのドキュメント

カスタムセットアップ

Custom Setup オプションは、本番環境向けに設計されており、VPC エンドポイントを使用してネットワークトラフィックを AWS の内部ネットワーク内に維持します。詳細はこちら: カスタムセットアップのドキュメント

この例では、SageMaker の k-NN アルゴリズム と、有名な Iris データセット を使用します。

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 モジュールをインポートするセルを作成します。Python3 Data Science インスタンスには、これらのライブラリがあらかじめインストールされています。

import os
import random
import string

import boto3
import matplotlib.pyplot as plt
import pandas as pd
import sagemaker
from IPython.display import display
from sagemaker import image_uris
from sagemaker.deserializers import JSONDeserializer
from sagemaker.estimator import Estimator, Predictor
from sagemaker.inputs import TrainingInput
from sagemaker.serializers import CSVSerializer
from sklearn.model_selection import train_test_split

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

# 定数を定義
CSV_PATH = './tmp/iris.csv'
S3_DATASET_BUCKET = os.getenv('S3_DATASET_BUCKET')
S3_DATASET_TRAIN = os.getenv('S3_DATASET_TRAIN')
S3_DATASET_TEST = os.getenv('S3_DATASET_TEST')
S3_TRAIN_OUTPUT = os.getenv('S3_TRAIN_OUTPUT')
SAGEMAKER_ROLE = os.getenv('SAGEMAKER_ROLE')
ESTIMATOR_INSTANCE_COUNT = 1
ESTIMATOR_INSTANCE_TYPE = 'ml.m5.large'
PREDICTOR_INSTANCE_TYPE = 'ml.t2.medium'
PREDICTOR_ENDPOINT_NAME = f'sagemaker-knn-{PREDICTOR_INSTANCE_TYPE}'.replace('.', '-')

# 変数を定義
bucket = boto3.resource('s3').Bucket(S3_DATASET_BUCKET)
train_df = None
test_df = None
train_object_path = None
test_object_path = None
knn = None
predictor = None

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

!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 を読み込み、前処理を行います。

• SageMaker は CSV の最初の列がターゲットラベルまたはクラスである必要があります。そのため、Species 列を最初の位置に移動する必要があります。
• ターゲットラベル (種の名前) は、互換性のために整数に変換する必要があります。

SageMaker のドキュメントで詳細を確認してください: CSV Format Requirements

def load_csv(path: str) -> pd.DataFrame:
    # CSV を Pandas DataFrame に読み込む
    df = pd.read_csv(path)
    # ラベル列 ('Species') を最初の位置に移動
    df = df[['Species', 'Sepal.Length', 'Sepal.Width', 'Petal.Length', 'Petal.Width']]
    # ターゲットラベル ('Species') を整数に変換
    df['Species'] = df['Species'].map({'setosa': 0, 'versicolor': 1, 'virginica': 2})
    return df

この関数は、Iris データセットを SageMaker の要件に準拠させ、組み込みアルゴリズムによるトレーニングの準備を行います。

散布図を作成して、特徴量と分布を確認します。

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

生成された散布図の軸マッピングは以下の通りです。

• X 軸: Species、Sepal.Length、Sepal.Width、Petal.Length、Petal.Width を左から右に表します。
• Y 軸: Petal.Width、Petal.Length、Sepal.Width、Sepal.Length、Species を下から上に表します。

プロットを観察すると、特徴量から種を予測できる可能性が高いことが明らかであり、データポイントが明確なグループに分類されていることがわかります。

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

def upload_csv_to_s3(df: pd.DataFrame, object_path: str) -> str:
    filename = ''.join([random.choice(string.digits + string.ascii_lowercase) for i in range(10)])
    path = os.path.abspath(os.path.join('./tmp', filename))
    df.to_csv(path, header=False, index=False)
    # デフォルトの content-type が binary/octet-stream であるため変更
    bucket.upload_file(path, object_path, ExtraArgs={'ContentType': 'text/csv'})
    return f's3://{bucket.name}/{object_path}'

if __name__ == '__main__':
    df = load_csv(CSV_PATH)
    display(df)
    plot(df)
    train_df, test_df = train_test_split(df, shuffle=True, random_state=0)
    train_object_path = upload_csv_to_s3(train_df, S3_DATASET_TRAIN)
    test_object_path = upload_csv_to_s3(test_df, S3_DATASET_TEST)

k-NN Predictor を設定し、トレーニングプロセスを開始します。

def get_estimator(**hyperparams) -> Estimator:
    estimator = Estimator(
        image_uri=image_uris.retrieve('knn', boto3.Session().region_name),
        role=SAGEMAKER_ROLE,
        instance_count=ESTIMATOR_INSTANCE_COUNT,
        instance_type=ESTIMATOR_INSTANCE_TYPE,
        input_mode='Pipe',
        output_path=f's3://{S3_DATASET_BUCKET}/{S3_TRAIN_OUTPUT}',
        sagemaker_session=sagemaker.Session(),
    )
    hyperparams.update({'predictor_type': 'classifier'})
    estimator.set_hyperparameters(**hyperparams)
    return estimator


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


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

ECR コンテナ URI

image_uri (3 行目) は、AWS によって提供される k-NN トレーニングアルゴリズムの ECR コンテナ URI を指定します。適切な URI は、SageMaker のユーティリティ関数を使用してリージョンに応じて取得できます。

組み込みアルゴリズムのコンテナ URI の詳細については、公式ドキュメントをご参照ください: Amazon SageMaker Built-in Algorithm ECR URIs。

チャネル名

Amazon SageMaker の組み込みアルゴリズムで使用される チャネル名 (18 行目) は、train に固定されています。トレーニングジョブ作成時に test チャネルを含める場合、ML モデルはトレーニング後にテストデータで自動的に評価されます。

Pipe モードの使用

データストリーミング効率を高めるために、TrainingInput 定義の input_mode (7 行目) パラメータを "Pipe" に設定することで Pipe モード を有効にできます。Pipe モードでは、データが S3 から SageMaker インスタンスに直接ストリームされるため、データセット全体をダウンロードする際のレイテンシやメモリ要件が削減されます。

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

k-NN アルゴリズムには、いくつかの構成可能なハイパーパラメータがあります。それらの使用法と効果の詳細については、公式ドキュメントをご参照ください: k-NN Hyperparameters。

トレーニングログ

トレーニングジョブを開始すると、次のようなログが表示されます。

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

• このログには、accuracy や macro F1 score などのメトリクスが記載されており、テストデータセットにおけるモデルのパフォーマンスに関する洞察を提供します。
• トレーニング中に train チャネルと test チャネルの両方を使用することで、組み込みアルゴリズムがモデルの汎化能力を自動的に評価します。

トレーニング済みモデルをエンドポイントにデプロイし、予測を検証します。SageMaker では、デプロイ済み推論エンドポイントと対話する際に、入力データと出力データのフォーマットを指定するために serializer と deserializer を使用します。

def deploy(estimator: Estimator) -> Predictor:
    return estimator.deploy(
        initial_instance_count=1,
        instance_type=PREDICTOR_INSTANCE_TYPE,
        serializer=CSVSerializer(),
        deserializer=JSONDeserializer(),
        endpoint_name=PREDICTOR_ENDPOINT_NAME,
    )


def validate(predictor: Predictor, test_df: pd.DataFrame) -> pd.DataFrame:
    rows = []
    for _, data in test_df.iterrows():
        predict = predictor.predict(
            pd.DataFrame([data.drop('Species')]).to_csv(header=False, index=False),
            initial_args={'ContentType': 'text/csv'},
        )
        predicted_label = predict['predictions'][0]['predicted_label']
        row = data.tolist()
        row.append(predicted_label)
        row.append(data['Species'] == predicted_label)
        rows.append(row)
    return pd.DataFrame(rows, columns=('Species', 'Sepal.Length', 'Sepal.Width', 'Petal.Length', 'Petal.Width', 'Prediction', 'Result'))


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

推論結果には Prediction と Result 列が含まれ、表形式で表示されます。データの構成は以下の通りです。

• Prediction: モデルに提供された各サンプルの予測ラベル。
• Result: 予測が実際のラベルと一致するかを示すブール値 (True または False)。

不要なコストを回避するために、リソースを解放します。

def delete_model(predictor: Predictor) -> None:
    predictor.delete_model()


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


if __name__ == '__main__':
    delete_model(predictor)
    delete_endpoint(predictor)

SageMaker の k-NN のような組み込みアルゴリズムを使用することで、トレーニングからデプロイまで ML ワークフローを簡略化できます。SageMaker は、基盤となるインフラストラクチャをシームレスに処理しながら、開発者が効果的なモデルの構築に集中できるよう支援します。

Happy Coding! 🚀

モデルをデプロイする際、serializer と deserializer を定義することで、入力データおよび出力データのフォーマットを制御できます。

from sagemaker.predictor import Predictor


predictor = Predictor(
    endpoint_name="my-endpoint",
    serializer=CSVSerializer(),  # 入力データを CSV フォーマットに変換
    deserializer=JSONDeserializer()  # 出力データを JSON フォーマットに変換
)

これらの設定により、SageMaker 推論エンドポイントとのシームレスな通信が可能になり、リクエストの送信やレスポンスの処理が容易になります。

serializer は入力データ (例: Python オブジェクト) を SageMaker エンドポイントが期待するフォーマットに変換します。

例:

• CSVSerializer: データを CSV フォーマットに変換。
• JSONSerializer: データを JSON フォーマットに変換。
• NumpySerializer: NumPy 配列をバイナリフォーマットに変換。

deserializer は SageMaker エンドポイントからの出力データを解釈し、使用可能な Python オブジェクトに変換します。

例:

• JSONDeserializer: JSON レスポンスを Python の辞書やリストに変換。
• BytesDeserializer: 生のバイトデータを返却。