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

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

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

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

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)

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)

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)

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)