추천 시스템 맛보기 - AWS SageMaker를 중심으로

import sagemaker
import sagemaker.amazon.common as smac
import numpy as np
import pandas as pd
import boto3
import io
import os

from sagemaker import get_execution_role
from sagemaker.predictor import json_deserializer
from sagemaker.amazon.amazon_estimator import get_image_uri
from scipy.sparse import lil_matrix

# 트레이닝 데이터를 Pandas DataFrame 에 담습니다.
user_movie_ratings_train = pd.read_csv(
  'ua.base.shuffled',
  sep='\t',
  index_col=False,
  names=['user_id' , 'movie_id' , 'rating'],
)

# 상위 5건 데이터를 출력시킵니다.
user_movie_ratings_train.head(5)

# 테스트 데이터를 Pandas DataFrame 에 담습니다.
user_movie_ratings_test = pd.read_csv(
  'ua.test',
  sep='\t',
  index_col=False,
  names=['user_id' , 'movie_id' , 'rating'],
)

# 상위 5건 데이터를 출력시킵니다.
user_movie_ratings_test.head(5)

def loadDataset(dataframe, lines, columns):
    # (lines, columns) 모양의 빈 희소 행렬을 만들고 float32 타입으로 변환합니다.
    X = lil_matrix((lines, columns)).astype('float32')
    # 레이블의 여기에 저장됩니다.
    Y = []

    line=0
    for index, row in dataframe.iterrows():
        X[line,row['user_id']-1] = 1
        X[line, nb_users+(row['product_id']-1)] = 1

        # 4점 이상의 데이터만 고려합니다.
        if int(row['rating']) >= 4:
            Y.append(1)
        else:
            Y.append(0)

            line=line+1

    Y=np.array(Y).astype('float32')
    return X,Y

nb_users = train_set['user_id'].max()
nb_products = train_set['product_id'].max()
nb_features = nb_users + nb_products

nb_ratings_test = len(user_product_ratings_test.index)
nb_ratings_train = len(user_product_ratings_train.index)

print ('# of users: ', nb_users)
print ('# of products: ', nb_products)
print ('Features (# of users + # of movies): ', nb_features)

print ('\nLoadDataset...')

X_train, Y_train = loadDataset(user_product_ratings_train, nb_ratings_train, nb_features)
X_test, Y_test = loadDataset(user_product_ratings_test, nb_ratings_test, nb_features)

print ('Success')

assert X_train.shape == (nb_ratings_train, nb_features)
assert Y_train.shape == (nb_ratings_train, )

zero_labels = np.count_nonzero(Y_train)

print("Training labels: %d zeros, %d ones" % (zero_labels, nb_ratings_train-zero_labels))

assert X_test.shape == (nb_ratings_test, nb_features)
assert Y_test.shape == (nb_ratings_test, )

zero_labels = np.count_nonzero(Y_test)

print("Test labels: %d zeros, %d ones" % (zero_labels, nb_ratings_test-zero_labels))

def writeDatasetToProtobuf(X, bucket, prefix, key, d_type, Y=None):
    buf = io.BytesIO()

    if d_type == 'sparse':
        smac.write_spmatrix_to_sparse_tensor(buf, X, labels=Y)
    else:
        smac.write_numpy_to_dense_tensor(buf, X, labels=Y)

    buf.seek(0)
    obj = '{}/{}'.format(prefix, key)
    boto3.resource('s3').Bucket(bucket).Object(obj).upload_fileobj(buf)

    return 's3://{}/{}'.format(bucket,obj)

# 업로드 할 경로를 지정합니다.
prefix = 'labs-fm'

train_key = 'train.protobuf'
train_prefix = '{}/{}'.format(prefix, 'train')

test_key = 'test.protobuf'
test_prefix = '{}/{}'.format(prefix, 'test')

output_prefix = 's3://{}/{}/output'.format(bucket, prefix)

fm_train_data_path = writeDatasetToProtobuf(X_train, bucket, train_prefix, train_key, 'sparse', Y_train)
fm_test_data_path  = writeDatasetToProtobuf(X_test, bucket, test_prefix, test_key, 'sparse', Y_test)

print ('Training data S3 path: ',fm_train_data_path)
print ('Test data S3 path: ',fm_test_data_path)
print ('FM model output S3 path: {}'.format(output_prefix))

# 트레이닝에 사용할 수 있는 인스턴스 타입은 정해져 있으며, 그 외 타입을 지정할 경우 ValidationException 가 발생합니다.
# 인스턴스 타입이 너무 작을 경우에는 OutOfMemoryError 가 발생하므로 준비된 모델의 크기를 고려하여 지정합니다.
instance_type='ml.m5.large'

fm = sagemaker.estimator.Estimator(
  get_image_uri(boto3.Session().region_name, 'factorization-machines'),
  get_execution_role(),
  train_instance_count=1,
  train_instance_type=instance_type,
  output_path=output_prefix,
  sagemaker_session=sagemaker.Session(),
)

# 필수로 적용해주어야 하는 하이퍼파라미터를 지정합니다.
fm.set_hyperparameters(
  feature_dim=nb_features,
  predictor_type='binary_classifier',
  num_factors=64,
)

# 훈련을 시작합니다!
fm.fit(
  {
    'train': fm_train_data_path,
    'test': fm_test_data_path
  }
)

# 모듈이 없을 경우에는 pip install 과정을 추가합니다.
import mxnet as mx

model_file_name = 'model.tar.gz'
model_full_path = fm.output_path +'/'+ fm.latest_training_job.job_name +'/output/'+model_file_name

print 'Model Path: ', model_full_path

# S3 버킷으로부터 FM 모델을 다운로드 합니다.
os.system('aws s3 cp '+model_full_path+ ' .')

# 모델을 추출하여 MXNet 으로 불러옵니다.
os.system('tar xzvf '+model_file_name)
os.system('unzip -o model_algo-1')
os.system('mv symbol.json model-symbol.json')
os.system('mv params model-0000.params')


m = mx.module.Module.load('./model', 0, False, label_names=['out_label'])
V = m._arg_params['v'].asnumpy()
w = m._arg_params['w1_weight'].asnumpy()
b = m._arg_params['w0_weight'].asnumpy()

# 아이템 잠재 매트릭스를 생성합니다.
knn_item_matrix = np.concatenate((V[nb_users:], w[nb_users:]), axis=1)
knn_train_label = np.arange(1,nb_movies+1)

# 사용자 잠재 매트릭스를 생성합니다.
ones = np.ones(nb_users).reshape((nb_users, 1))
knn_user_matrix = np.concatenate((V[:nb_users], ones), axis=1)

ResourceLimitExceeded: An error occurred (ResourceLimitExceeded) when calling the CreateEndpoint operation: The account-level service limit 'ml.m5.2xlarge for endpoint usage' is 0 Instances, with current utilization of 0 Instances and a request delta of 1 Instances. Please contact AWS support to request an increase for this limit.