Amazon SageMakerを使用してモデル構築を効率化するためのフィーチャーストアの作成

# インポート
from sagemaker.feature_store.feature_group import FeatureGroup
from time import gmtime, strftime, sleep
from random import randint
import pandas as pd
import numpy as np
import subprocess
import sagemaker
import importlib
import logging
import time
import sys
import boto3
from datetime import datetime, timezone, date

if sagemaker.__version__ < '2.48.1':
    subprocess.check_call([sys.executable, '-m', 'pip', 'install', 'sagemaker==2.48.1'])
    importlib.reload(sagemaker)
logger = logging.getLogger('__name__')
logger.setLevel(logging.DEBUG)
logger.addHandler(logging.StreamHandler())
logger.info(f'Using SageMaker version: {sagemaker.__version__}')
logger.info(f'Using Pandas version: {pd.__version__}')

# デフォルト設定
sagemaker_session = sagemaker.Session()
default_bucket = sagemaker_session.default_bucket()
logger.info(f'Default S3 bucket = {default_bucket}')
prefix = 'sagemaker-feature-store'
region = sagemaker_session.boto_region_name

# 入力データを指し示すために以下の変数を更新します
bucket='quickstart-feature-store-demo'
data_key = 'Transaction Dataset Sample Features.csv'
data_location = 's3://{}/{}'.format(bucket, data_key)

orders_df = pd.read_csv(data_location)

# sagemakerが使用するIAMロールを取得します
role = sagemaker.get_execution_role()

# 一意のイベントタイムスタンプを生成する
def generate_event_timestamp(x):
    # ローカルタイムを表すnaive datetime
    naive_dt = datetime.now()
    # タイムゾーンを考慮に入れる
    aware_dt = naive_dt.astimezone()
    # UTC時間
    utc_dt = aware_dt.astimezone(timezone.utc)
    # ISO-8601形式に変換
    event_time = utc_dt.isoformat(timespec='milliseconds')
    event_time = event_time.replace('+00:00', 'Z')
    return event_time

orders_df['event_time'] = orders_df['Date'].apply(lambda x : generate_event_timestamp(x))

for cols in orders_df.columns:
    orders_df[cols] = orders_df[cols].astype('string')

# フィーチャーグループの作成
current_timestamp = strftime('%m-%d-%H-%M', gmtime())
orders_feature_group_name = f'orders-{current_timestamp}'
%store orders_feature_group_name

# フィーチャーグループの定義の作成
logger.info(f'Orders feature group name = {orders_feature_group_name}')
orders_feature_group = FeatureGroup(name=orders_feature_group_name, sagemaker_session=sagemaker_session)
orders_feature_group.load_feature_definitions(data_frame=orders_df)

def wait_for_feature_group_creation_complete(feature_group):
    status = feature_group.describe().get('FeatureGroupStatus')
    print(f'初期状態: {status}')
    while status == 'Creating':
        logger.info(f'フィーチャーグループ {feature_group.name} の作成を待っています...')
        time.sleep(5)
        status = feature_group.describe().get('FeatureGroupStatus')
    if status != 'Created':
        raise SystemExit(f'フィーチャーグループ {feature_group.name} の作成に失敗しました: {status}')
    logger.info(f'フィーチャーグループ {feature_group.name} が正常に作成されました。')

orders_feature_group.create(s3_uri=f's3://{default_bucket}/{prefix}', 
                            record_identifier_name='OrderId', 
                            event_time_feature_name='event_time', 
                            role_arn=role, 
                            enable_online_store=True)
wait_for_feature_group_creation_complete(orders_feature_group)

%%time

logger.info(f'フィーチャーグループ {orders_feature_group.name} にデータを取り込んでいます...')
orders_feature_group.ingest(data_frame=orders_df, max_processes=16, wait=True)
logger.info(f'{len(orders_df)} 件の注文記録がフィーチャーグループ {orders_feature_group.name} に取り込まれました。')

# すべてのフィーチャーストアのリストを取得します
boto_session = boto3.Session(region_name=region)
sagemaker_client = boto_session.client(service_name='sagemaker', region_name=region)
sagemaker_client.list_feature_groups()