從特徵群組建立資料集 - Amazon SageMaker AI
使用 Amazon SageMaker Python SDK 從特徵群組取得資料Amazon Athena 查詢範例

本文為英文版的機器翻譯版本，如內容有任何歧義或不一致之處，概以英文版為準。

從特徵群組建立資料集

在離線儲存中建立特徵商店特徵群組後，您可以選擇使用以下方法取得資料：

  • 使用 Amazon SageMaker Python SDK

  • 在 Amazon Athena 中執行 SQL 查詢

重要

Feature Store 需要在資料目錄中註冊 AWS Glue 資料。在預設情況下，當您建立功能群組時，特徵存放區會自動建置 AWS Glue 資料目錄。

為離線儲存建立特徵群組並填入資料後，您可以透過執行查詢建立資料集，或使用 SDK 聯結儲存在不同特徵群組的離線儲存中的資料。您還可以將特徵群組加入到單個 Pandas 資料框架。您可以使用 Amazon Athena 來寫入和執行 SQL 查詢。

注意

為了確保您的資料是最新的，您可以設定 AWS Glue 爬蟲程式以排程執行。

若要設定 AWS Glue 爬蟲程式，請指定爬蟲程式用來存取離線存放區 Amazon S3 儲存貯體的 IAM 角色。如需詳細資訊，請參閱建立 IAM 角色。

如需如何使用 AWS Glue 和 Athena 建置模型訓練和推論的訓練資料集的詳細資訊，請參閱搭配適用 SDK for Python (Boto3) 使用功能存放區。

使用 Amazon SageMaker Python SDK 從特徵群組取得資料

您可以透過 特徵商店 API 從特徵群組建立資料集。資料科學家透過從離線儲存中的一個或多個特徵群組擷取機器學習 (ML) 特徵資料，來建立用於訓練的 ML 資料集。使用 create_dataset() 函式來建立資料集。您可以透過 SDK 執行下列動作：

  • 從特徵群組建立多個資料集。

  • 從特徵群組和 Pandas 資料框架建立資料集。

依預設，特徵商店不包含您從資料集中刪除的記錄。它也不包括重複的記錄。重複的記錄在事件時間欄中具有記錄 ID 和時間戳記值。

使用 SDK 建立資料集之前，您必須先啟動 SageMaker AI 工作階段。使用下方程式碼來啟動工作階段。

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 中可能不存在或具有相同的結構。建議將這些特徵群組取代為特徵商店中存在的特徵群組。如需如何建立特徵群組的資訊，請參閱步驟 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"])

特徵商店 API 為您提供 create_dataset 函式的協助城市方法。您可以使用此函式執行下列動作：

  • 從特徵群組建立多個資料集。

  • 從多個特徵群組和 Pandas 資料框架建立資料集。

  • 從單個特徵群組和 Pandas 資料框架建立資料集。

  • 使用時間點精確關連建立資料集，其中聯結的特徵群組中的記錄會依序進行。

  • 使用重複的記錄建立資料集，而不是遵循函式的預設行為。

  • 使用刪除的記錄建立資料集，而不是遵循函式的預設行為。

  • 為指定的時段建立資料集。

  • 將資料集另存為 CSV 檔案。

  • 將資料集另存為 Pandas 資料框架。

基準特徵群組是聯結的重要概念。基準特徵群組是具有其他特徵群組或 Pandas 資料框架與其聯結的特徵群組。針對每個資料集

您可以將下列選用方法新增至 create_dataset 函式，以設定建立資料集的方式：

  • with_feature_group – 使用記錄識別符和基準特徵群組中的目標特徵名稱，在基準特徵群組與其他特徵群組之間執行內部聯結。以下提供您指定之參數的相關資訊：

    • feature_group – 您要加入的特徵群組。

    • target_feature_name_in_base – 您在聯結中用作關鍵字的基本特徵群組中的特徵名稱。其他特徵群組中的記錄識別符是特徵商店在關連中使用的其他鍵。

    • included_feature_names – 表示基準特徵群組之特徵名稱的字串清單。您可以透過使用欄位來指定您要包含在資料集中的特徵。

    • feature_name_in_target – 代表目標特徵群組中將與基準特徵群組中的目標特徵進行比較的可選字串。

    • join_comparator – 表示將基本特徵群組中的目標特徵與目標特徵群組中特徵聯結時使用的比較器的可選 JoinComparatorEnum。依預設，JoinComparatorEnum 值可以是 GREATER_THAN、GREATER_THAN_OR_EQUAL_TO、LESS_THAN、LESS_THAN_OR_EQUAL_TO、NOT_EQUAL_TO 或 EQUALS。

    • join_type – 表示基準特徵群組和目標特徵群組之間的聯結類型的可選 JoinTypeEnum。依預設，JoinTypeEnum 值可以是 LEFT_JOIN、RIGHT_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 撰寫查詢，以便從特徵群組建立資料集。您還可以撰寫查詢，從特徵群組和單個 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