Feature Store サンプルノートブックを使用した不正検出

重要

Amazon SageMaker Studio または Amazon SageMaker Studio Classic が Amazon SageMaker リソースを作成できるようにするカスタムIAMポリシーでは、それらのリソースにタグを追加するアクセス許可も付与する必要があります。Studio と Studio Classic は、作成したリソースに自動的にタグ付けするため、リソースにタグを追加するアクセス許可が必要です。IAM ポリシーで Studio と Studio Classic がリソースの作成を許可されていてもタグ付けが許可されていない場合、リソースの作成時にAccessDenied「」エラーが発生する可能性があります。詳細については、「リソースにタグ付けするための SageMakerアクセス許可を提供する」を参照してください。

AWS Amazon の マネージドポリシー SageMaker SageMaker リソースを作成するアクセス許可を付与する には、それらのリソースの作成中にタグを追加するアクセス許可が既に含まれています。

このページのサンプルコードは、サンプルノートブック「Amazon SageMaker Feature Store による不正検出」を参照しています。このガイドのコードは概念的であり、コピーしても完全には機能しないため、Studio Classic、ノートブックインスタンス、または JupyterLab でこのノートブックを実行することをお勧めします。

以下を使用して、サンプルノートブックを含む aws/amazon-sagemaker-examples GitHub repository のクローンを作成します。

• Studio Classic の場合

  Studio Classic を初めて起動します。Studio または Studio Classic がデフォルトのエクスペリエンスとして有効になっている場合は、Studio Classic を開くことができます。Studio Classic を開くには、「」を参照してくださいAmazon SageMaker コンソールを使用して Studio Classic を起動する。

  「」の手順に従って、aws/amazon-sagemaker-examples GitHub repository を Studio Classic にクローンします SageMaker Studio Classic で Git リポジトリのクローンを作成する。

• Amazon SageMaker ノートブックインスタンスの場合

  「」の手順に従って SageMaker 、ノートブックインスタンスを起動しますノートブックインスタンスへのアクセス。

  「」の手順に従って、サンプルがノートブックにすでにあるかどうかを確認しますサンプルノートブック。そうでない場合は、「」の指示に従ってくださいGit リポジトリを Amazon SageMaker アカウントに追加する。

サンプル SageMaker ノートブックを作成したら、 amazon-sagemaker-examples/sagemaker-featurestore ディレクトリに移動し、Amazon SageMaker Feature Store による不正検出のサンプルノートブックを開きます。

ステップ 1: Feature Store セッションを設定する

Feature Store の使用を開始するには、 SageMaker セッション、Boto3 セッション、Feature Store セッションを作成します。また、特徴量用の Amazon S3 バケットを設定します。これがオフラインストアとなります。次のコードでは、 SageMaker デフォルトのバケットを使用してカスタムプレフィックスを追加します。

注記

ノートブックの実行に使用するロールには、AmazonSageMakerFullAccess と AmazonSageMakerFeatureStoreAccess のマネージドポリシーががアタッチされている必要があります。IAM ロールにポリシーを追加する方法については、「」を参照してくださいIAM ロールへのポリシーの追加。

import boto3
import sagemaker
from sagemaker.session import Session

sagemaker_session = sagemaker.Session()
region = sagemaker_session.boto_region_name
boto_session = boto3.Session(region_name=region)
role = sagemaker.get_execution_role()
default_bucket = sagemaker_session.default_bucket()
prefix = 'sagemaker-featurestore'
offline_feature_store_bucket = 's3://{}/{}'.format(default_bucket, prefix)

sagemaker_client = boto_session.client(service_name='sagemaker', region_name=region)
featurestore_runtime = boto_session.client(service_name='sagemaker-featurestore-runtime', region_name=region)

feature_store_session = Session(
    boto_session=boto_session,
    sagemaker_client=sagemaker_client,
    sagemaker_featurestore_runtime_client=featurestore_runtime
)

ステップ 2: データセットをロードし、データを特徴量グループに分割する

各特徴のデータフレームにデータをロードします。これらのデータフレームは、特徴グループをセットアップした後に使います。この不正検出の例では、これらのステップが次のコード中に見えます。

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import io

s3_client = boto3.client(service_name='s3', region_name=region)

fraud_detection_bucket_name = 'sagemaker-featurestore-fraud-detection'
identity_file_key = 'sampled_identity.csv'
transaction_file_key = 'sampled_transactions.csv'

identity_data_object = s3_client.get_object(Bucket=fraud_detection_bucket_name, Key=identity_file_key)
transaction_data_object = s3_client.get_object(Bucket=fraud_detection_bucket_name, Key=transaction_file_key)

identity_data = pd.read_csv(io.BytesIO(identity_data_object['Body'].read()))
transaction_data = pd.read_csv(io.BytesIO(transaction_data_object['Body'].read()))

identity_data = identity_data.round(5)
transaction_data = transaction_data.round(5)

identity_data = identity_data.fillna(0)
transaction_data = transaction_data.fillna(0)

# Feature transformations for this dataset are applied before ingestion into FeatureStore.
# One hot encode card4, card6
encoded_card_bank = pd.get_dummies(transaction_data['card4'], prefix = 'card_bank')
encoded_card_type = pd.get_dummies(transaction_data['card6'], prefix = 'card_type')

transformed_transaction_data = pd.concat([transaction_data, encoded_card_type, encoded_card_bank], axis=1)
transformed_transaction_data = transformed_transaction_data.rename(columns={"card_bank_american express": "card_bank_american_express"})

ステップ 3: 特徴量グループを設定する

特徴グループをセットアップするときは、特徴名を一意の名前でカスタマイズし、FeatureGroup クラスを使って各特徴グループをセットアップする必要があります。

from sagemaker.feature_store.feature_group import FeatureGroup
feature_group_name = "some string for a name"
feature_group = FeatureGroup(name=feature_group_name, sagemaker_session=feature_store_session)

例えば、この不正検出の例では、2 つの特徴グループは identity と transaction です。次のコードでは、タイムスタンプを使って名前をカスタマイズする方法と、その後名前とセッションを渡すことで各グループをセットアップする方法を見ることができます。

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

identity_feature_group_name = 'identity-feature-group-' + strftime('%d-%H-%M-%S', gmtime())
transaction_feature_group_name = 'transaction-feature-group-' + strftime('%d-%H-%M-%S', gmtime())

identity_feature_group = FeatureGroup(name=identity_feature_group_name, sagemaker_session=feature_store_session)
transaction_feature_group = FeatureGroup(name=transaction_feature_group_name, sagemaker_session=feature_store_session)

ステップ 4: レコード識別子とイベント時刻の特徴量を設定する

このステップでは、レコード識別子名とイベント時間特徴名を指定します。この名前は、データ内の対応する特徴の列にマッピングされます。この不正検出の例では、関心のある列は TransactionID です。タイムスタンプが使用できない場合は、データに EventTime を追加できます。次のコードでは、どのようにこれらの変数が設定され、その後どのように両方の特徴のデータに EventTime が追加されるかを確認できます。

record_identifier_name = "TransactionID"
event_time_feature_name = "EventTime"
current_time_sec = int(round(time.time()))
identity_data[event_time_feature_name] = pd.Series([current_time_sec]*len(identity_data), dtype="float64")
transformed_transaction_data[event_time_feature_name] = pd.Series([current_time_sec]*len(transaction_data), dtype="float64")

ステップ 5: 特徴量定義をロードする

これで、特徴データを含むデータフレームを渡すことで、特徴定義をロードできるようになりました。不正検出の例の次のコードでは、load_feature_definitions を使ってアイデンティティの特徴とトランザクションの特徴がそれぞれロードされます。この関数はデータの各列のデータ型を自動的に検出します。自動検出ではなくスキーマを使うデベロッパーは、「Data Wrangler から特徴グループをエクスポートする」のコード例を参照してください。この例では、FeatureGroup の作成に使用できる FeatureDefinition として、スキーマをロードし、マッピングし、追加する方法を示しています。この例では、 SageMaker Python の代わりに使用できる AWS SDK for Python (Boto3) 実装についても説明しますSDK。

identity_feature_group.load_feature_definitions(data_frame=identity_data); # output is suppressed
transaction_feature_group.load_feature_definitions(data_frame=transformed_transaction_data); # output is suppressed

ステップ 6: 特徴量グループを作成する

このステップでは、create 関数を使って特徴グループを作成します。次のコードは、利用可能なすべてのパラメータを示しています。オンラインストアはデフォルトで作成されないため、有効にする場合は、これを True に設定する必要があります。s3_uri は、オフラインストアの S3 バケットの場所です。

# create a FeatureGroup
feature_group.create(
    description = "Some info about the feature group",
    feature_group_name = feature_group_name,
    record_identifier_name = record_identifier_name,
    event_time_feature_name = event_time_feature_name,
    feature_definitions = feature_definitions,
    role_arn = role,
    s3_uri = offline_feature_store_bucket,
    enable_online_store = True,
    online_store_kms_key_id = None,
    offline_store_kms_key_id = None,
    disable_glue_table_creation = False,
    data_catalog_config = None,
    tags = ["tag1","tag2"])

次のコードは不正検出の例で、作成する 2 つの特徴グループそれぞれについて、最小限の create 呼び出しを示しています。

identity_feature_group.create(
    s3_uri=offline_feature_store_bucket,
    record_identifier_name=record_identifier_name,
    event_time_feature_name=event_time_feature_name,
    role_arn=role,
    enable_online_store=True
)

transaction_feature_group.create(
    s3_uri=offline_feature_store_bucket,
    record_identifier_name=record_identifier_name,
    event_time_feature_name=event_time_feature_name,
    role_arn=role,
    enable_online_store=True
)

特徴グループを作成すると、データのロードに時間がかかり、特徴グループが作成されて使えるようになるまで待つ必要があります。次の方法を使うと、状態をチェックできます。

status = feature_group.describe().get("FeatureGroupStatus")

特徴グループが作成中である場合は、レスポンスとして Creating を受け取ります。このステップが正常に完了した場合、レスポンスは Created になります。その他のあり得るステータスは CreateFailed、Deleting、または DeleteFailed です。

ステップ 7: 特徴量グループを操作する

特徴量グループを設定できたので、次のどのタスクも実行できます。

特徴量グループを記述する

describe 関数を使って、特徴グループに関する情報を取得できます。

feature_group.describe()

特徴量グループを一覧表示する

list_feature_groups 関数を使って、すべての特徴量グループを一覧表示できます。

sagemaker_client.list_feature_groups()

特徴量グループにレコードを入れる

ingest 関数を使って特徴データをロードできます。特徴データのデータフレームを渡し、ワーカー数を設定し、それが戻るのを待つかどうかを選択します。次の例は ingest 関数を使う方法を示しています。

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

各特徴グループに対して、ロードする特徴データに ingest 関数を実行します。

特徴量グループからレコードを取得する

get_record 関数を使い、レコード識別子によって特定の特徴のデータを取得できます。次の例では、サンプルの識別子を使ってレコードを取得しています。

record_identifier_value = str(2990130)
featurestore_runtime.get_record(FeatureGroupName=transaction_feature_group_name, RecordIdentifierValueAsString=record_identifier_value)

不正検出の例からのレスポンスの例を次に示します。

...
'Record': [{'FeatureName': 'TransactionID', 'ValueAsString': '2990130'},
  {'FeatureName': 'isFraud', 'ValueAsString': '0'},
  {'FeatureName': 'TransactionDT', 'ValueAsString': '152647'},
  {'FeatureName': 'TransactionAmt', 'ValueAsString': '75.0'},
  {'FeatureName': 'ProductCD', 'ValueAsString': 'H'},
  {'FeatureName': 'card1', 'ValueAsString': '4577'},
...

Hive DDL コマンドを生成する

SageMaker Python SDKの FeatureStore クラスは、Hive DDL コマンドを生成する機能も提供します。テーブルのスキーマは、特徴定義に基づいて生成されます。列の名前は特徴名に基づいており、データ型は特徴のタイプに基づいて推論されます。

print(feature_group.as_hive_ddl())

出力例:

CREATE EXTERNAL TABLE IF NOT EXISTS sagemaker_featurestore.identity-feature-group-27-19-33-00 (
  TransactionID INT
  id_01 FLOAT
  id_02 FLOAT
  id_03 FLOAT
  id_04 FLOAT
 ...

トレーニングデータセットを構築する

Feature Store は、特徴量グループを作成するときに AWS Glue データカタログを自動的に構築し、必要に応じてこれをオフにすることができます。以下で、このトピックの前の方で作成したアイデンティティ特徴グループとトランザクション特徴グループの両方の特徴値を使って、1 つのトレーニングデータセットを作成する方法を説明します。また、Amazon Athena クエリを実行して、オフラインストアに保存されたアイデンティティ特徴グループとトランザクション特徴グループの両方のデータを結合する方法を説明します。

まず、athena_query() を使って、アイデンティティとトランザクションの両方の特徴グループに対して Athena クエリを作成します。`table_name` は、Feature Store によって自動生成される AWS Glue テーブルです。

identity_query = identity_feature_group.athena_query()
transaction_query = transaction_feature_group.athena_query()

identity_table = identity_query.table_name
transaction_table = transaction_query.table_name

Athena クエリを作成して実行する

これらの特徴量グループSQLで を使用してクエリを記述し、 .run() コマンドを使用してクエリを実行し、そこでデータセットを保存する Amazon S3 バケットの場所を指定します。

# Athena query
query_string = 'SELECT * FROM "'+transaction_table+'" LEFT JOIN "'+identity_table+'" ON "'+transaction_table+'".transactionid = "'+identity_table+'".transactionid'

# run Athena query. The output is loaded to a Pandas dataframe.
dataset = pd.DataFrame()
identity_query.run(query_string=query_string, output_location='s3://'+default_s3_bucket_name+'/query_results/')
identity_query.wait()
dataset = identity_query.as_dataframe()

ここから、このデータセットを使ってモデルをトレーニングし、次いで推論を実行できます。

特徴量グループの削除

特徴グループは、delete 関数を使って削除できます。

feature_group.delete()

次のコード例は、不正検出の例のものです。

identity_feature_group.delete()
transaction_feature_group.delete()

詳細については、「特徴量グループの削除API」を参照してください。