Amazon SageMaker Processing 処理ジョブを作成してデータ変換ワークロードを実行する

aws:
  sagemaker:
    s3bucket: "{train.csv が配置されている S3 バケット名}"

# 必要なライブラリのインポート
import pandas as pd
import yaml
import boto3
import sagemaker
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.processing import ProcessingInput, ProcessingOutput

# 設定ファイルのパスを指定
SETTING_FILE_PATH = './settings.yaml'

# YAMLファイルを読み込み、AWS情報を取得
with open(SETTING_FILE_PATH) as file:
    aws_info = yaml.safe_load(file)

# SageMakerセッションを初期化
sess = sagemaker.Session()

# S3バケット名を設定ファイルから取得
s3bucket = aws_info['aws']['sagemaker']['s3bucket']

# boto3を使用してS3クライアントを初期化
s3_client = boto3.client('s3',region_name='ap-northeast-1')

# S3バケットから'data/input/train.csv'ファイルをローカルの'train.csv'ファイルにダウンロード
s3_client.download_file(s3bucket, 'data/input/train.csv', './train/train.csv')

# ダウンロードしたCSVファイルをPandasデータフレームとして読み込み
df = pd.read_csv("./train/train.csv")

# データフレームの先頭数行を表示
df.head()

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "s3:GetObject",
                "s3:ListBucket"
            ],
            "Resource": [
                "arn:aws:s3:::{train.csv が配置されている S3 バケット名}",
                "arn:aws:s3:::{train.csv が配置されている S3 バケット名}/*"
            ]
        }
    ]
}

import os

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split

def preprocess_data(df):
    # 1. 欠損値の処理 'workclass'カラムの欠損値を'Unknown'で埋める
    df['workclass'] = df['workclass'].fillna('Unknown')

    # 2. 異常値の処理 'age'が100以上の値を100に制限する
    df['age'] = df['age'].clip(upper=100)

    # 3. カテゴリカル変数のエンコーディング
    df = pd.get_dummies(df, columns=['gender'], prefix='gender')

    # 4. `age`から年齢層カラムの作成
    df['age_group'] = pd.cut(df['age'], bins=[0, 18, 30, 45, 60, 100], 
                             labels=['Under 18', '18-30', '31-45', '46-60', 'Over 60'])

    # 5. 教育年数から教育レベルカラムを作成
    df['education_level'] = df['educational-num'].apply(education_level)

    return df

def education_level(years): # 5. 教育年数から教育レベルカラムを作成
    if years <= 8:
        return 'Elementary'
    elif years <= 12:
        return 'High School'
    elif years <= 16:
        return 'College'
    else:
        return 'Graduate'

# メイン処理
if __name__ == "__main__":
    # データの読み込み
    input_train_data_path = os.path.join("/opt/ml/processing/input","train")
    df = pd.read_csv(input_train_data_path)

    # データの前処理
    processed_df = preprocess_data(df)

    # データの分割
    train_data, test_data = train_test_split(processed_df, test_size=0.2, random_state=42)

    # 出力先パス指定
    output_train_data_path = "/opt/ml/processing/output/train"
    output_test_data_path = "/opt/ml/processing/output/test"

    # データの書き出し用フォルダを作成する
    try :
        os.makedirs(output_train_data_path, exist_ok=True)
        os.makedirs(output_test_data_path, exist_ok=True)
    except :
        pass

    # データを直接指定したフォルダに書き出す
    try :
        train_data.to_csv(os.path.join(output_train_data_path, "processed_train_data.csv"), index=False)
        test_data.to_csv(os.path.join(output_test_data_path, "processed_test_data.csv"), index=False)
        print("データの書き出しが完了しました。")
    except :
        pass

import sagemaker
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.processing import ProcessingInput, ProcessingOutput

# 設定ファイルのパスを指定
SETTING_FILE_PATH = './settings.yaml'

# YAMLファイルを読み込み、AWS情報を取得
with open(SETTING_FILE_PATH) as file:
    aws_info = yaml.safe_load(file)

# 読み込んだ内容から必要な情報を取得
s3bucket = aws_info['aws']['sagemaker']['s3bucket']

# 入力データ、出力データの S3 URI の指定
input_data_s3_uri = "s3://{}/data/input/".format(s3bucket)
output_data_s3_uri = "s3://{}/data/output/".format(s3bucket)

# `ProcessingInput`、`ProcessingOutput`を設定
res = processor.run(
    code = "preprocessing.py",
    inputs = [ProcessingInput(
            source = input_data_s3_uri,
            destination = "/opt/ml/processing/input"),
            ],
    outputs = [
        ProcessingOutput(
            source = "/opt/ml/processing/output/train",
            destination = output_data_s3_uri),
        ProcessingOutput(
            source = "/opt/ml/processing/output/test",
            destination = output_data_s3_uri),
            ],
        )

aws:
  sagemaker:
    role: "arn:aws:iam::{自分のAWSアカウントID}:role/service-role/SageMaker-AmazonSageMaker-ExecutionRole"
    s3bucket: "{train.csv が配置されている S3 バケット名}"

# 必要なライブラリのインポート
import pandas as pd
import yaml
import boto3
import sagemaker
from sagemaker.sklearn.processing import SKLearnProcessor
from sagemaker.processing import ProcessingInput, ProcessingOutput

# 設定ファイルのパスを指定
SETTING_FILE_PATH = './settings.yaml'

# YAML ファイルを読み込み、AWS情報を取得
with open(SETTING_FILE_PATH) as file:
    aws_info = yaml.safe_load(file)

# S3 バケット、IAM ロールの情報を設定ファイルから取得
role = aws_info['aws']['sagemaker']['role']
s3bucket = aws_info['aws']['sagemaker']['s3bucket']

# SageMaker セッション初期化
sm = boto3.client('sagemaker')

# インスタンスタイプと台数の指定
processing_instance_type = "ml.t3.medium"
processing_instance_count = 1

# 入力データ、出力データの S3 URI の指定
input_data_s3_uri = "s3://{}/data/input/train.csv".format(s3bucket)
output_data_s3_uri = "s3://{}/data/output/".format(s3bucket)

# SageMaker Processing ジョブ名の指定
processing_job_name = "sklearn-preprocessor-test"

# SageMaker Processing を動かすためのリソースの作成
processor = SKLearnProcessor(
    framework_version = "0.23-1",
    role=role,
    instance_type = processing_instance_type,
    instance_count = processing_instance_count,
    max_runtime_in_seconds = 7200
)

# SageMaker Processing ジョブの作成
res = processor.run(
    code = "preprocessing.py",
    inputs = [ProcessingInput(
            source = input_data_s3_uri,
            destination = "/opt/ml/processing/input/train"),
            ],
    outputs = [
        ProcessingOutput(
            source = "/opt/ml/processing/output/train",
            destination = output_data_s3_uri),
        ProcessingOutput(
            source = "/opt/ml/processing/output/test",
            destination = output_data_s3_uri),
            ],
    wait=True,
    logs=False,
    job_name=processing_job_name,
    experiment_config=None
)

sm = boto3.client('sagemaker')
jobs = sm.list_processing_jobs()
pd.DataFrame(jobs['ProcessingJobSummaries'])[:1]