特徴量グループからデータセットを作成する - Amazon SageMaker
Amazon SageMaker Python SDKを使用して特徴量グループからデータを取得するAmazon Athena クエリの例

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

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

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

  • Amazon SageMaker Python の使用 SDK

  • Amazon Athena でのSQLクエリの実行

重要

Feature Store では、データを AWS Glue データカタログに登録する必要があります。デフォルトでは、特徴量グループを作成すると、Feature Store は自動的に AWS Glue データカタログを構築します。

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

注記

データが最新であることを確認するために、スケジュールに従って実行する AWS Glue クローラーを設定できます。

AWS Glue クローラーを設定するには、クローラーがオフラインストアの Amazon S3 バケットにアクセスするために使用する IAMロールを指定します。詳細については、「 IAMロールの作成」を参照してください。

AWS Glue と Athena を使用してモデルトレーニングと推論用のトレーニングデータセットを構築する方法の詳細については、「」を参照してくださいSDK for Python (Boto3) で Feature Store を使用する。

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

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

  • 複数の特徴量グループからデータセットを作成します。

  • 特徴量グループと Pandas データフレームからデータセットを作成します。

デフォルトでは、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 データフレームからデータセットを作成します。

  • 1 つの特徴量グループと Pandas データフレームからデータセットを作成します。

  • 結合された特徴量グループのレコードが順番に続く、特定の時点での正確な結合を使用してデータセットを作成します。

  • 関数のデフォルト動作に従う代わりに、重複したレコードを含むデータセットを作成します。

  • 関数のデフォルト動作に従う代わりに、削除したレコードを含むデータセットを作成します。

  • 指定した期間のデータセットを作成します。

  • データセットをCSVファイルとして保存します。

  • データセットを Pandas データフレームとして保存します。

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

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

  • with_feature_group — 基本特徴量グループのレコード識別子 とターゲット特徴量名を使用して、基本特徴量グループと別の特徴量グループとの内部結合を実行します。以下には、指定したパラメータに関する情報が記載されています。

    • feature_group — 結語している特徴量グループ

    • target_feature_name_in_base — 結合のキーとして使用している基本特徴量グループの特徴量の名前 他の特徴量グループのレコード識別子は、Feature Store が結合に使用するその他のキーです。

    • included_feature_names — 基本特徴量グループの特徴量名を表す文字列のリスト このフィールドを使用して、データセットに含める特徴量を指定することができます。

    • feature_name_in_target — 基本特徴量グループのターゲット特徴量と比較される、ターゲット特徴量グループの特徴量を表すオプションの文字列

    • join_comparator — 基本特徴量グループのターゲット特徴量とターゲット特徴量グループの特徴量を結合するときに使用されるコンパレータを表すオプションの JoinComparatorEnum。これらの JoinComparatorEnum 値はデフォルトで GREATER_THAN、LESS_THAN、LESS_THAN_OR_EQUAL_TO、NOT_EQUAL_TO、または EQUALS になります。

    • join_type — 基本特徴量グループとターゲット特徴量グループの結合タイプを表すオプションの JoinTypeEnum。これらの JoinTypeEnum 値はデフォルトで LEFT_JOIN、FULL_JOIN、CROSS_JOIN、または INNER_JOIN になります。

  • with_event_time_range — 指定したイベント時間範囲を使用してデータセットを作成します。

  • as_of — 指定したタイムスタンプまでのデータセットを作成します。例えば、値として datetime(2021, 11, 28, 23, 55, 59, 342380) を指定すると、2021 年 11 月 28 日までのデータセットが作成されます。

  • point_time_accurate_join — 基本特徴量グループのすべてのイベント時間値が、結合している特徴量グループまたは Pandas データフレームのすべてのイベント時間値よりも小さいデータセットを作成します。

  • include_duplicated_records — 重複した値を特徴量グループ内に保持します。

  • include_deleted_records — 削除された値を特徴量グループに保持します。

  • with_number_of_recent_records_by_record_identifier — データセット内に表示される最新のレコードの数を決定するために指定する整数。

  • with_number_of_records_by_record_identifier — データセットに表示されるレコード数を表す整数。

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

  • to_csv_file – データセットをCSVファイルとして保存します。

  • to_dataframe — データセットを Pandas データフレームとして保存します。

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

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