【AWS】SageMakerノートブックスでk-平均法 (k-Means)推論をやってみた

SageMakerノートブックスでk-平均法 (k-Means)推論をやってみた

実践！

１．環境をセットアップ
　・S3バケット作成
　・SageMaker用IAMロール作成
　　※作成したS3バケットへの読み取り権限を付与
　　　面倒であればフルアクセスをどうぞ
　・SageMakerノートブックを起動

２．SageMakerノートブックのJupyterLab(conda_python3)を開き、必要なライブラリをインポート

import pandas as pd
import numpy as np
from sklearn.datasets import make_blobs
import boto3
import sagemaker
from sagemaker import get_execution_role
from sagemaker.amazon.common import write_numpy_to_dense_tensor
from sagemaker.serializers import CSVSerializer
import matplotlib.pyplot as plt

※WARNINGが出るがいったん無視


３．サンプルデータの生成

# データ生成：3つのクラスタを持つ2次元データ
n_samples = 300
n_features = 2
n_clusters = 3

# サンプルデータ生成
X, y = make_blobs(n_samples=n_samples, centers=n_clusters, n_features=n_features, random_state=42)

# NumPy配列をfloat32に変換
features = X.astype('float32')

# データの形を確認
print("データの形:", features.shape)
print("データ型:", features.dtype)

４．データをRecordIO形式に変換してS3にアップロード

# S3バケット名とプレフィックスを設定
s3_bucket = '<作成したS3バケット名>'
prefix = 'kmeans-example'

# RecordIOに変換
with open('train_data.recordio', 'wb') as f:
    write_numpy_to_dense_tensor(f, features)

# S3にアップロード
s3 = boto3.client('s3')
s3.upload_file('train_data.recordio', s3_bucket, f'{prefix}/train_data.recordio')

# S3パスを確認
s3_train_data = f's3://{s3_bucket}/{prefix}/train_data.recordio'
print('S3にアップロード完了:', s3_train_data)

５．K-Meansアルゴリズムのトレーニング

# 実行ロールを取得
role = get_execution_role()

# K-MeansアルゴリズムのコンテナURIを取得
region = boto3.Session().region_name
container = sagemaker.image_uris.retrieve('kmeans', region)

# SageMakerセッションを作成
session = sagemaker.Session()

# K-Means Estimatorを設定
kmeans = sagemaker.estimator.Estimator(
    image_uri=container,
    role=role,
    instance_count=1,
    instance_type='ml.m5.large',
    output_path=f's3://{s3_bucket}/{prefix}/output',
    sagemaker_session=session
)

# ハイパーパラメータを設定
kmeans.set_hyperparameters(
    k=3,  # クラスタ数
    feature_dim=2  # 特徴量の次元数
)

# トレーニングを実行
kmeans.fit({'train': s3_train_data})
print("トレーニングジョブが開始されました。")

６．モデルのデプロイと予測

# モデルをデプロイ
kmeans_predictor = kmeans.deploy(
    initial_instance_count=1,
    instance_type='ml.m5.large'
)

print("エンドポイントがデプロイされました。")

７．クラスタリング結果を予測

from sagemaker.serializers import CSVSerializer

# テストデータを用意（numpy.ndarray形式）
test_data = np.array([[4.0, 5.0], [1.0, 2.0], [10.0, 8.0]]).astype('float32')

# kmeans_predictorのserializerをCSVSerializerに設定
kmeans_predictor.serializer = CSVSerializer()

# 予測を実行
result = kmeans_predictor.predict(test_data)

# 結果を表示
print("予測結果（クラスタ割り当て）:")
for record in result:
    print(record)

８．クラスタリング結果を可視化

# 散布図でクラスタリング結果を可視化
plt.scatter(test_data[:, 0], test_data[:, 1], c=result, cmap='viridis')
plt.title("クラスタリング結果の可視化")
plt.xlabel("Feature 1")
plt.ylabel("Feature 2")
plt.show()

９．リソースのクリーンアップ

# エンドポイントを削除
kmeans_predictor.delete_endpoint()
print("エンドポイントを削除しました。")

---

感想

家帰ろっと。