SageMaker クイックスタート

この記事の「学習 → デプロイ → 推論 → 片付け」セクションを、そのままコピペで動く・詰まりにくい 形に整えました。各セルは上から順に実行してください（Studio / JupyterLab 前提）。

1) セットアップ（S3 バケット準備・ライブラリ）

!pip -q install sagemaker boto3 scikit-learn

import os, json, boto3, sagemaker
from botocore.exceptions import ClientError

region = boto3.Session().region_name
sess   = sagemaker.Session()
role   = sagemaker.get_execution_role()  # StudioならOK

# 使うS3バケット名（グローバルで一意。既存バケットを使っても可）
account = boto3.client("sts").get_caller_identity()["Account"]
bucket  = f"{account}-sagemaker-{region}"
prefix  = "demo-xgb"

s3 = boto3.client("s3", region_name=region)

# バケット作成（存在しないときだけ作る）
def ensure_bucket(bkt: str, region: str):
    try:
        s3.head_bucket(Bucket=bkt)
        print("S3 bucket already exists:", bkt)
    except ClientError as e:
        if region == "us-east-1":
            s3.create_bucket(Bucket=bkt)
        else:
            s3.create_bucket(
                Bucket=bkt,
                CreateBucketConfiguration={"LocationConstraint": region},
            )
        print("S3 bucket created:", bkt)

ensure_bucket(bucket, region)

print("Region:", region)
print("Role:", role)
print("Bucket:", bucket)

メモ

• リージョンは自動検出。us-east-1 はバケット作成API仕様が異なるので分岐しています。
• 既存の専用バケットがある場合は、bucket をその名前に置き換えてOK。

2) ダミーデータ作成 → S3 へアップロード

import numpy as np, pandas as pd
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split

# 2値分類のサンプルデータ（特徴量10個）
X, y = make_classification(n_samples=2000, n_features=10, random_state=0)
Xtr, Xva, ytr, yva = train_test_split(X, y, test_size=0.2, random_state=42)

os.makedirs("data/train", exist_ok=True)
os.makedirs("data/validation", exist_ok=True)

# 先頭列=ラベル、以降=特徴量（XGBoostのCSV想定）
pd.DataFrame(np.column_stack([ytr, Xtr])).to_csv("data/train/train.csv", index=False, header=False)
pd.DataFrame(np.column_stack([yva, Xva])).to_csv("data/validation/validation.csv", index=False, header=False)

# S3にアップロード
s3_train = sess.upload_data("data/train",      bucket=bucket, key_prefix=f"{prefix}/train")
s3_val   = sess.upload_data("data/validation", bucket=bucket, key_prefix=f"{prefix}/validation")
print("Uploaded to:", s3_train, s3_val)

メモ

• ここで s3://<bucket>/<prefix>/train/ と validation/ にCSVが置かれます。

3) 学習ジョブ（SageMaker 内蔵 XGBoost）

from sagemaker import image_uris
from sagemaker.estimator import Estimator
from sagemaker.inputs import TrainingInput

# XGBoost コンテナイメージを取得
container = image_uris.retrieve(framework="xgboost", region=region, version="1.7-1")

# Estimator（学習ジョブ）
estimator = Estimator(
    image_uri=container,
    role=role,
    instance_count=1,
    instance_type="ml.m5.large",
    output_path=f"s3://{bucket}/{prefix}/output",
    use_spot_instances=True,      # コスト削減
    max_run=3600,
    max_wait=5400,
    enable_network_isolation=True,
    sagemaker_session=sess,
)

# ハイパーパラメータ（最小限）
estimator.set_hyperparameters(
    objective="binary:logistic",
    num_round=100,
    max_depth=5,
    eta=0.2,
    subsample=0.8,
    colsample_bytree=0.8,
    eval_metric="auc",
)

# 学習データ入力
train_input = TrainingInput(s3_data=s3_train, content_type="text/csv")
val_input   = TrainingInput(s3_data=s3_val,   content_type="text/csv")

# 学習実行（~数分）
estimator.fit({"train": train_input, "validation": val_input})

メモ

• ログは Studio のコンソールと CloudWatch Logs に流れます。
• 出力（学習済みモデル）は output_path（S3）に保存されます。

4) リアルタイム推論エンドポイント → 推論テスト

from sagemaker.serializers import CSVSerializer

# モデルをREST API化（エンドポイント作成）
predictor = estimator.deploy(
    initial_instance_count=1,
    instance_type="ml.m5.large",
)

# CSV 文字列で送信（XGBoostはCSVが簡単）
predictor.serializer = CSVSerializer()

test_row = ",".join(map(str, Xva[0]))  # 特徴量のみ1行
result = predictor.predict(test_row)
print("raw prediction:", result)

メモ

• 返り値は確率（例：b'0.87\n'）。用途に合わせて閾値（例：0.5）を決めます。
• エンドポイント作成 ~1–3分。課金対象なので、テスト後は必ず削除。

5) 片付け（課金止め）

# エンドポイントは使わない時に必ず削除
try:
    ep_name = predictor.endpoint_name
    predictor.delete_endpoint()
    print("Deleted endpoint:", ep_name)
except Exception as e:
    print("Endpoint cleanup error:", e)

メモ

• 片付けを忘れるとインスタンス分の料金がかかり続けます。
• 学習成果物（S3上の model.tar.gz など）は残るので、再デプロイ可能です。

よくある詰まり & チェックリスト

• AccessDenied (S3/Logs): 実行ロールに S3（対象バケット）/ CloudWatch Logs 権限があるか。
• リージョン不一致: Studio・S3・学習ジョブのリージョンを揃える。
• CSV形式: 先頭列=ラベル、以降=特徴量（学習時）。推論時は特徴量のみ。
• コスト最適化: use_spot_instances=True、テスト後は delete_endpoint()。

付録: 研究済みモデルをそのままデプロイする例

既に s3://.../model.tar.gz がある場合：

from sagemaker import image_uris, Model

container = image_uris.retrieve("xgboost", region=region, version="1.7-1")
model = Model(image_uri=container, model_data="s3://your-bucket/path/to/model.tar.gz", role=role)

predictor = model.deploy(initial_instance_count=1, instance_type="ml.m5.large")