Amazon SageMaker Feature Storeに特徴量を保存する

Daisuke Asada

2022-09-11

機械学習には"Garbage in, garbage out.“という言葉があります。トレーニングデータが悪ければ、そこから作成されたモデルも悪くなる、ということですが、機械学習にとって学習データというものは重要な要素です。今回はspaCyによって出力されたテキストのベクトルデータを、Feature Storeに保存し、Amazon SageMakerのプラットフォームで管理できるようにしたいと思います。

特徴量ストアという考え方

機械学習にはデータが必要です。どんなに最先端のアルゴリズムであっても、データがなければどうしようもありません。

そして、同じアルゴリズムでもハイパーパラメータを変えて何度も学習処理を行うことも行われますし、同じデータに対して違うアルゴリズムで学習処理を行うこともあります。

また、データも変動していくので、都度違うデータで学習処理を行うこともあります。

そのような中で、ある時点での機械学習モデルがどのデータから作成されたものを管理することが必要になります。

そこで、機械学習に使用するデータ（特徴量）を管理するための仕組みが特徴量ストアという考え方になります。

Feature Storeを利用することで、機械学習におけるデータの管理をSageMakerのプラットフォームの中で統一的な形で管理することができるようになります。

特徴量の作成

まずは特徴量を用意します。

データは以前の記事で作成した、1行1パラグラフに処理したテキストデータをもとにします。

This truth would seem to hold in (省略) winner.
For this irrational, (省略) were still alive.
Thus he gained an almost unique base upon which to form his estimate.

といった形で、1行1パラグラフのテキストデータです。

この段落ごとのテキストデータをそのパラグラフを表すベクトルデータに変換します。そのためにspaCyを使用します。spaCyを使用すると、簡単に文章をベクトルデータに変換することができます。

以下でspaCyをインストールします。

pip install spacy==3.4.1

文章のベクトル変換し、データフレーム化を行うコードは以下です。

import spacy
import pandas as pd
import numpy as np


# 英語モデルの読み込み
nlp = spacy.load('en_core_web_lg')

# テキストデータをリストとして読むこむ
docs = []
with open(f"./data/scipio_africanus_paragraph_data.txt", "r") as input_f:
    while True:
        line = input_f.readline()
        if not line:
            break
        docs.append(line.rstrip())

# ベクトルに変換
docs_vector = [nlp(doc).vector for doc in docs]

# データフレーム化
df = pd.DataFrame(docs_vector)
df.columns = ["vec" + str(c) for c in df.columns ]
# 一意に特定するための列を追加
df['paragraph_no'] = np.arange(1, len(df)+1)

ポイントとしては

nlp(doc).vector

とすることで、テキストデータをベクトルとして表したデータに変換できます。

今回は300次元のベクトルを得られます。

例えば、“Hello World.“であれば、以下の様になります（使用するモデルに依存します）。

1.2472819e+00, -1.8643667e+00, 2.7026336e+00, -2.7492666e+00,(省略) -5.2936739e-01, -3.4282330e-01, -2.1853266e+00, -1.1235000e+00

このようにして得られたベクトルを分類アルゴリズムなどの特徴量として利用することができます。

Feature Storeへのデータの登録

特徴量が得られたので次にFeature Storeにデータを登録します。

Feature StoreはFeature Groupという単位で、特徴量を管理します。RDBのテーブルに相当します。

import sagemaker
import time
from time import strftime, gmtime
from sagemaker.feature_store.feature_group import FeatureGroup


prefix = 'featurestore-spacy-vector'
role = "arn:aws:iam::ACCOUNT_ID:role/SageMakerExecutionRole"

sagemaker_session = sagemaker.Session()
region = sagemaker_session.boto_region_name
s3_bucket_name = sagemaker_session.default_bucket()

feature_group_name = 'spacy-vector-feature-group'
feature_group = FeatureGroup(
    name=feature_group_name, sagemaker_session=sagemaker_session
)
current_time_sec = int(round(time.time()))
record_identifier_feature_name = "paragraph_no"
df["EventTime"] = pd.Series([current_time_sec]*len(df), dtype="float64")
feature_group.load_feature_definitions(data_frame=df)
feature_group.create(
    s3_uri=f"s3://{s3_bucket_name}/{prefix}",
    record_identifier_name=record_identifier_feature_name,
    event_time_feature_name="EventTime",
    role_arn=role,
    enable_online_store=True
)

あとはFeature Groupにデータを登録します。

feature_group.ingest(
    data_frame=df, max_workers=3, wait=True
)

Feature Storeからデータを取り出す

登録完了したので、データをFeature Groupから取り出すことができるようになります。

records = sagemaker_session.boto_session.client(
    "sagemaker-featurestore-runtime", region_name=region
).batch_get_record(
    Identifiers=[
        {
            "FeatureGroupName": feature_group_name,
            "RecordIdentifiersValueAsString": ["1"],
        }
    ]
)

これによって以下のようなデータが取得できます。

{'ResponseMetadata': {'RequestId': '8b471866-eb55-45b5-9715-51627ca48e74',
  'HTTPStatusCode': 200,
  'HTTPHeaders': {'x-amzn-requestid': '8b471866-eb55-45b5-9715-51627ca48e74',
   'content-type': 'application/json',
   'content-length': '18882',
   'date': 'Sun, 11 Sep 2022 09:52:30 GMT'},
  'RetryAttempts': 0},
 'Records': [{'FeatureGroupName': 'spacy-vector-feature-group',
   'RecordIdentifierValueAsString': '1',
   'Record': [{'FeatureName': 'vec0', 'ValueAsString': '-3.0186550617218018'},
    {'FeatureName': 'vec1', 'ValueAsString': '-0.6359930038452148'},
    {'FeatureName': 'vec2', 'ValueAsString': '-0.5122418403625488'},
    {'FeatureName': 'vec3', 'ValueAsString': '-0.016656620427966118'},
    ...(省略)
    {'FeatureName': 'vec299', 'ValueAsString': '-0.5346629619598389'},

    {'FeatureName': 'paragraph_no', 'ValueAsString': '1'},
    {'FeatureName': 'EventTime', 'ValueAsString': '1662885447.0'}]}],
 'Errors': [],
 'UnprocessedIdentifiers': []}

Athena経由でクエリ

Feature Storeを作成したときに指定したS3バケットを見てみると、spacy-vector-feature-group-xxxxxxxxxx/dataといったprefix配下に、登録したデータがParquet形式で保存されています。

また、Feature Storeを作成するとデフォルトでGlue Data Catalogにテーブルデータが登録されるので、Athenaでデータをクエリすることができます。

query = feature_group.athena_query()
table = query.table_name
query_string = f"SELECT * FROM \"{table}\" WHERE paragraph_no = 1"

# run Athena query. The output is loaded to a Pandas dataframe
query.run(query_string=query_string, output_location=f"s3://{s3_bucket_name}/{prefix}")

query.wait()
data = query.as_dataframe()