翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

特徴量グループからデータセットを作成する

Feature Store の特徴量グループがオフラインストアに作成されたら、次の方法を使用してデータを取得できます。

オフラインストアの機能グループを作成し、データを入力したら、クエリを実行するか、 を使用して、さまざまな機能グループからオフラインストアに保存されているデータを結合SDKすることで、データセットを作成できます。特徴量グループを 1 つの Pandas データフレームと連携させることもできます。Amazon Athena を使用して、SQLクエリの書き込みと実行を行うことができます。

Amazon SageMaker Python SDKを使用して特徴量グループからデータを取得する

Feature Store APIs を使用して、特徴量グループからデータセットを作成できます。データサイエンティストは、オフラインストアの 1 つ以上の特徴量グループから ML 特徴量データを取得して、トレーニング用の ML データセットを作成します。create_dataset() 関数を使用してデータセットを作成します。を使用して、以下SDKを実行できます。

デフォルトでは、Feature Store にはデータセットから削除したレコードは含まれません。また、重複したレコードは含まれません。重複レコードでは、イベント時間列にレコード ID とタイムスタンプ値が表示されます。

を使用してデータセットSDKを作成する前に、 SageMaker セッションを開始する必要があります。次のコードを使用してセッションを開始します。

import boto3
from sagemaker.session import Session
from sagemaker.feature_store.feature_store import FeatureStore

region = boto3.Session().region_name
boto_session = boto3.Session(region_name=region)

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,
)

feature_store = FeatureStore(feature_store_session) 

次のコードは、複数の特徴量グループからデータセットを作成する例を示しています。次のコードスニペットでは、特徴量グループの例を使用します「base_fg_name", "first_fg_name「」、「」second_fg_name「」。これは Feature Store 内に存在しないか、同じスキーマを持つ可能性があります。これらの特徴量グループを、Feature Store 内に存在する特徴量グループに置き換えることをお勧めします。特徴量グループの作成方法については、「ステップ 3: 特徴グループを作成する」を参照してください。

from sagemaker.feature_store.feature_group import FeatureGroup

s3_bucket_name = "offline-store-sdk-test" 

base_fg_name = "base_fg_name"
base_fg = FeatureGroup(name=base_fg_name, sagemaker_session=feature_store_session)

first_fg_name = "first_fg_name"
first_fg = FeatureGroup(name=first_fg_name, sagemaker_session=feature_store_session)

second_fg_name = "second_fg_name"
second_fg = FeatureGroup(name=second_fg_name, sagemaker_session=feature_store_session)

feature_store = FeatureStore(feature_store_session)
builder = feature_store.create_dataset(
    base=base_fg,
    output_path=f"s3://{amzn-s3-demo-bucket1}",
).with_feature_group(first_fg
).with_feature_group(second_fg, "base_id", ["base_feature_1"])         
        

次のコードは、複数の特徴量グループと Pandas データフレームからデータセットを作成する例を示しています。

base_data = [[1, 187512346.0, 123, 128],
             [2, 187512347.0, 168, 258],
             [3, 187512348.0, 125, 184],
             [1, 187512349.0, 195, 206]]
base_data_df = pd.DataFrame(
    base_data, 
    columns=["base_id", "base_time", "base_feature_1", "base_feature_2"]
)

builder = feature_store.create_dataset(
    base=base_data_df, 
    event_time_identifier_feature_name='base_time', 
    record_identifier_feature_name='base_id',
    output_path=f"s3://{s3_bucket_name}"
).with_feature_group(first_fg
).with_feature_group(second_fg, "base_id", ["base_feature_1"])            
        

Feature Store APIs は、create_dataset関数のヘルパーメソッドを提供します。これを使用すると以下のことを実行できます。

基本特徴量グループは結合の重要な概念です。基本特徴量グループとは、他の特徴量グループまたは Pandas データフレームが結合されている特徴量グループのことです。各データセットについて

以下のオプションメソッドを create_dataset 関数に追加して、データセットの作成方法を設定できます。

データセットを設定したら、次のいずれかの方法を使用して出力を指定できます。

特定の期間以降のデータを取得できます。次のコードは、タイムスタンプ以降のデータを取得します。

fg1 = FeatureGroup("example-feature-group-1")
feature_store.create_dataset(
    base=fg1, 
    output_path="s3://example-S3-path"
).with_number_of_records_from_query_results(5).to_csv_file()  

特定の期間のデータを取得することもできます。次のコードを使用して、特定の時間範囲のデータを取得できます。

fg1 = FeatureGroup("fg1")
feature_store.create_dataset(
    base=fg1, 
    output_path="example-S3-path"
).with_event_time_range(
    datetime(2021, 11, 28, 23, 55, 59, 342380), 
    datetime(2020, 11, 28, 23, 55, 59, 342380)
).to_csv_file() #example time range specified in datetime functions

複数の特徴量グループを Pandas データフレームに結合して、その特徴量グループのイベント時間値が遅くともデータフレームのイベント時間になるようにするとよいでしょう。次のコードをテンプレートとして使用すると、結合を実行しやすくなります。

fg1 = FeatureGroup("fg1")
fg2 = FeatureGroup("fg2")
events = [['2020-02-01T08:30:00Z', 6, 1],
          ['2020-02-02T10:15:30Z', 5, 2],
          ['2020-02-03T13:20:59Z', 1, 3],
          ['2021-01-01T00:00:00Z', 1, 4]]
df = pd.DataFrame(events, columns=['event_time', 'customer-id', 'title-id']) 
feature_store.create_dataset(
    base=df, 
    event_time_identifier_feature_name='event_time', 
    record_identifier_feature_name='customer_id',
    output_path="s3://example-S3-path"
).with_feature_group(fg1, "customer-id"
).with_feature_group(fg2, "title-id"
).point_in_time_accurate_join(
).to_csv_file()

特定の期間以降のデータを取得できます。次のコードは、as_of メソッドのタイムスタンプで指定された時間以降のデータを取得します。

fg1 = FeatureGroup("fg1")
feature_store.create_dataset(
    base=fg1, 
    output_path="s3://example-s3-file-path"
).as_of(datetime(2021, 11, 28, 23, 55, 59, 342380)
).to_csv_file() # example datetime values

Amazon Athena クエリの例

Amazon Athena にクエリを記述して、特徴量グループからデータセットを作成できます。また、特徴量グループと 1 つの Pandas データフレームからデータセットを作成するクエリを記述することもできます。

対話型探索

このクエリは、最初の 1000 レコードを選択します。 

SELECT *
FROM <FeatureGroup.DataCatalogConfig.DatabaseName>.<FeatureGroup.DataCatalogConfig.TableName>
LIMIT 1000

最新のスナップショット (重複なし)

このクエリは、重複しない最新のレコードを選択します。

SELECT *
FROM
    (SELECT *,
         row_number()
        OVER (PARTITION BY <RecordIdentiferFeatureName>
    ORDER BY  <EventTimeFeatureName> desc, Api_Invocation_Time DESC, write_time DESC) AS row_num
    FROM <FeatureGroup.DataCatalogConfig.DatabaseName>.<FeatureGroup.DataCatalogConfig.TableName>)
WHERE row_num = 1;

オフラインストア内の最新のスナップショット (重複および削除済みレコードなし)

このクエリは、削除済みレコードをすべて除外し、重複しないレコードをオフラインストアから選択します。 

SELECT *
FROM
    (SELECT *,
         row_number()
        OVER (PARTITION BY <RecordIdentiferFeatureName>
    ORDER BY  <EventTimeFeatureName> desc, Api_Invocation_Time DESC, write_time DESC) AS row_num
    FROM <FeatureGroup.DataCatalogConfig.DatabaseName>.<FeatureGroup.DataCatalogConfig.TableName>)
WHERE row_num = 1 and 
NOT is_deleted;

オフラインストア内のタイムトラベル (重複および削除済みレコードなし)

このクエリは、削除済みレコードをすべて除外し、重複しない特定時点のレコードを選択します。

SELECT *
FROM
    (SELECT *,
         row_number()
        OVER (PARTITION BY <RecordIdentiferFeatureName>
    ORDER BY  <EventTimeFeatureName> desc, Api_Invocation_Time DESC, write_time DESC) AS row_num
    FROM <FeatureGroup.DataCatalogConfig.DatabaseName>.<FeatureGroup.DataCatalogConfig.TableName>
    where <EventTimeFeatureName> <= timestamp '<timestamp>')
    -- replace timestamp '<timestamp>' with just <timestamp>  if EventTimeFeature is of type fractional
WHERE row_num = 1 and
NOT is_deleted