AWS SageMaker 組み込み手法を使う際の投げ方

意外とフォーマットが共通じゃないのでメモ。
新しいのを試したら追加していく。

Estimator : LinearLiner

ファイル作成

1. 訓練・バリデーション・テストに分割（以降テストはファイル化不要）
2. dense_tensor形式に変換
3. s3に直接書き込み（下記の例ではファイル名はcsvがついているが、csvではない）

from sklearn.model_selection import train_test_split
import sagemaker.amazon.common as smac

def upload_file(data, s3_bucket, prefix, fname, key=None):
    buf = io.BytesIO()
    smac.write_numpy_to_dense_tensor(buf, np.array(data.drop(key, axis=1).as_matrix()).astype('float32'), np.array(data[key].as_matrix()).astype('float32'))
    buf.seek(0)
    s3_bucket.Object(os.path.join(prefix, fname)).upload_fileobj(buf)
    return 's3://{}/{}/{}'.format(s3_bucket.name, prefix, fname)

df_origin = pd.read_csv(fname)
train_data, validation_data, test_data = np.split(df_origin.sample(frac=1, random_state=1729), [int(0.7 * len(df_origin)), int(0.9 * len(df_origin))])

s3_bucket = boto3.resource('s3').Bucket(bucket)
s3_train_data = upload_file(data=train_data, s3_bucket=s3_bucket, prefix=os.path.join(prefix,'train'), fname='train.csv', key='Class')
s3_validation_data = upload_file(data=validation_data, s3_bucket=s3_bucket, prefix=os.path.join(prefix,'validation'), fname='validation.csv', key='Class')

モデル定義～デプロイまで

# SageMakerのセッション
sess = sagemaker.Session()
# sagemakerのestimatorへ必要項目を指定
linear = sagemaker.estimator.Estimator(get_region_path('linear-learner'),
                                       get_execution_role(), 
                                       train_instance_count=1, 
                                       train_instance_type='ml.m4.xlarge',
                                       output_path='s3://{}/{}/output'.format(bucket, prefix),
                                       sagemaker_session=sess)
# ハイパーパラメーターの指定
linear.set_hyperparameters(feature_dim=30,
                           mini_batch_size=100,
                           predictor_type='binary_classifier',
                           epochs=10,
                           num_models=32,
                           loss='absolute_loss')
# モデルフィッティングと出力先の指定（S3）
linear.fit({'train': s3_train_data, 'validation': s3_validation_data})

# デプロイ
from sagemaker.predictor import csv_serializer, json_deserializer
linear_predictor = linear.deploy(initial_instance_count=1, instance_type='ml.m4.xlarge')
linear_predictor.content_type = 'text/csv'
linear_predictor.serializer = csv_serializer
linear_predictor.deserializer = json_deserializer

予測

linear-learnerは二次元配列の形で渡せば一度に推定してくれる。
以下は参考にしたページに倣っていくつかに分割して推定してから統合している。

predictions = np.array([])
test_data_arr = test_data.drop(['Class'], axis=1).as_matrix()
split_array = np.array_split(test_data_arr, int(test_data_arr.shape[0] / float(100) + 1))
for array in split_array:
    _predictions = linear_predictor.predict(array)['predictions']
    for predict in _predictions:
        predictions = np.round(np.append(predictions, predict['score']))

結果

y_test = test_data['Class'].as_matrix()
print confusion_matrix(y_test, predictions)
print accuracy_score(y_test, predictions)
print precision_score(y_test, predictions)
print recall_score(y_test, predictions)
print f1_score(y_test, predictions)

Estimator : XG-Boost

ファイル作成

svmlight形式しか（たぶん）受け付けない。
デフォルトはsvmlightだが、content_typeを指定することでcsvも使える。
xgb = sagemaker.estimator.Estimator()で宣言したあと、xgb.content_type = 'text/csv'で変更できる。
実行の際、各ファイルが入っているディレクトリのアドレスしか渡さない（分割されたファイルに対応？）ので、関係ないファイルが入っているとエラーが出る。

from sklearn.datasets import dump_svmlight_file
dump_svmlight_file(X=train_data.drop('Class', axis=1), y=train_data['Class'], f='train.libsvm')
dump_svmlight_file(X=validation_data.drop('Class', axis=1), y=validation_data['Class'], f='validation.libsvm')

s3_bucket.Object(os.path.join(prefix, 'train_libsvm/train.libsvm')).upload_file('train.libsvm')
s3_bucket.Object(os.path.join(prefix, 'validation_libsvm/validation.libsvm')).upload_file('validation.libsvm')

s3_input_train = sagemaker.s3_input(s3_data='s3://{}/{}/train_libsvm/'.format(bucket, prefix), content_type='libsvm')
s3_input_validation = sagemaker.s3_input(s3_data='s3://{}/{}/validation_libsvm/'.format(bucket, prefix), content_type='libsvm')

モデル定義～デプロイまで

ここはほとんど変わらない。投げるパラメータが違う程度。

# SageMakerのセッション
sess = sagemaker.Session()
# sagemakerのestimatorへ必要項目を指定
xgb = sagemaker.estimator.Estimator(get_region_path('xgboost'),
                                       get_execution_role(), 
                                       train_instance_count=1, 
                                       train_instance_type='ml.m4.xlarge',
                                       output_path='s3://{}/{}/output'.format(bucket, prefix),
                                       sagemaker_session=sess)
# ハイパーパラメーターの指定
xgb.set_hyperparameters(max_depth=5,
                        eta=0.2,
                        gamma=4,
                        min_child_weight=6,
                        subsample=0.8,
                        silent=0,
                        objective='binary:logistic',
                        num_round=100)
# モデルフィッティングと出力先の指定（S3）
xgb.fit({'train': s3_input_train, 'validation': s3_input_validation})

# デプロイ
xgb_predictor = xgb.deploy(initial_instance_count=1, instance_type='ml.m4.xlarge')
xgb_predictor.content_type = 'text/csv'
xgb_predictor.serializer = csv_serializer
xgb_predictor.deserializer = json_deserializer

予測

linear-learnerと違って一度に複数のサンプルを投げられない。

test_data_arr = test_data.drop(['Class'], axis=1).as_matrix()
predictions_xgb = [xgb_predictor.predict(array) for array in test_data_arr]

結果

一緒なので割愛

TensorFlow

上記二つはEstimatorの中に含まれていたが、これは別なので投げ方もかなり違う。

importから

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import io
import time
import json
import sagemaker.amazon.common as smac
from sklearn.model_selection import train_test_split

ファイル作成

訓練とテストに分割する。（バリデーションが使えるかは未確認）
目的変数が一番最後の列になったcsv形式。

最初と二番目の列が、それぞれそのデータに含まれる【サンプル数・特徴量数】でないといけない。
（最後の列に目的変数が入るので、【特徴量数+1】列のデータになっていることに注意）
pandas DataFrameからto_csvする際は、index=Falseをすること。

df_origin = pd.read_csv('creditcard.csv')
train_df, test_df = train_test_split(df_origin)
train_df.rename(columns={train_df.columns[0]:len(train_df), train_df.columns[1]:30}, inplace=True)
test_df.rename(columns={test_df.columns[0]:len(test_df), test_df.columns[1]:30}, inplace=True)
train_df.to_csv('cc_training.csv',index=False)
test_df.to_csv('cc_test.csv',index=False)

import boto3
s3 = boto3.resource('s3')
bucket = 'creditcard.nk'
s3.Object(bucket, 'cc/cc_training.csv').upload_file('cc_training.csv')
s3.Object(bucket, 'cc/cc_test.csv').upload_file('cc_test.csv')
s3.Object(bucket, 'cc/iris_training.csv').upload_file('iris_training.csv')
s3.Object(bucket, 'cc/iris_test.csv').upload_file('iris_test.csv')
train_data_location = 's3://creditcard.nk/cc'

モデル定義～デプロイ

モデル定義ファイル

別途モデル定義ファイルをローカルの作業ディレクトリに置く必要がある。
このファイル内で、【特徴量数・クラス数・隠れ層の数】を指定する。
以下に含まれる4つの関数は、すべて必須。

define_tensorflow.py

import numpy as np
import os
import tensorflow as tf

INPUT_TENSOR_NAME = 'inputs'


def estimator_fn(run_config, params):
    feature_columns = [tf.feature_column.numeric_column(INPUT_TENSOR_NAME, shape=[30])]
    return tf.estimator.DNNClassifier(feature_columns=feature_columns,
                                      hidden_units=[30, 60, 120, 60, 30],
                                      n_classes=2,
                                      config=run_config)


def serving_input_fn(params):
    feature_spec = {INPUT_TENSOR_NAME: tf.FixedLenFeature(dtype=tf.float32, shape=[30])}
    return tf.estimator.export.build_parsing_serving_input_receiver_fn(feature_spec)()


def train_input_fn(training_dir, params):
    """Returns input function that would feed the model during training"""
    return _generate_input_fn(training_dir, 'cc_training.csv')


def eval_input_fn(training_dir, params):
    """Returns input function that would feed the model during evaluation"""
    return _generate_input_fn(training_dir, 'cc_test.csv')


def _generate_input_fn(training_dir, training_filename):
    training_set = tf.contrib.learn.datasets.base.load_csv_with_header(
        filename=os.path.join(training_dir, training_filename),
        target_dtype=np.int,
        features_dtype=np.float32)

    return tf.estimator.inputs.numpy_input_fn(
        x={INPUT_TENSOR_NAME: np.array(training_set.data)},
        y=np.array(training_set.target),
        num_epochs=None,
        shuffle=True)()

これを参照する形でモデル定義する。

jupyterインスタンス内でのモデル定義～デプロイ

from sagemaker import get_execution_role

#Bucket location to save your custom code in tar.gz format.
custom_code_upload_location = 's3://creditcard.nk/customcode/tensorflow_cc'

#Bucket location where results of model training are saved.
model_artifacts_location = 's3://creditcard.nk/artifacts'

#IAM execution role that gives SageMaker access to resources in your AWS account.
role = get_execution_role()
from sagemaker.tensorflow import TensorFlow

cc_estimator = TensorFlow(entry_point='cc_tensorflow.py',
                            role=role,
                            output_path=model_artifacts_location,
                            code_location=custom_code_upload_location,
                            train_instance_count=1,
                            train_instance_type='ml.m4.xlarge',
                            training_steps=1000,
                            evaluation_steps=100)
train_data_location = 's3://creditcard.nk/cc'
#region = boto3.Session().region_name
#train_data_location = 's3://sagemaker-sample-data-{}/tensorflow/iris'.format(region)

cc_estimator.fit(train_data_location)
tf_predictor = cc_estimator.deploy(initial_instance_count=1, instance_type='ml.m4.xlarge')
from sagemaker.tensorflow.predictor import csv_serializer
tf_predictor.content_type = 'text/csv'
tf_predictor.serializer = csv_serializer

予測

こちらも一度に複数のサンプルを投げられない。
加えて非常にややこしい形で返ってくる。
ごちゃっとしてdict型の中に、各クラスの確率がdict型で入っている。
以下は2クラスなので、クラス0の確率を取り出している。
(['classes']の後ろを[1]にするとクラス1の確率になる）

test_array = test_df.drop('Class',axis=1).as_matrix()
pred_score = []
for array in test_array:
    _res = tf_predictor.predict(array)['result']['classifications'][0]['classes'][0]['score']
    pred_score.append(_res)
predictions_tf = np.round(pred_score)

結果

割愛