步骤 1：设置 Feature Store 步骤 2：将数据集和分区数据加载到功能组中步骤 3：设置功能组步骤 4：设置记录标识符和事件时间功能步骤 5：加载功能定义步骤 6：创建功能组步骤 7：使用功能组

创建功能组

FeatureGroup 是主要Feature Store资源，其中包含存储在中的所有数据的元数据Amazon SageMaker Feature Store。功能组是功能存储中定义的用于描述记录的逻辑特征分组。功能组的定义由用于其在线和离线存储的功能定义、记录标识符名称和配置的列表组成。本主题中的示例代码使用 SageMaker Python 开发工具包。基础适用于使用其他语言APIs的开发人员。

在使用功能存储之前，您通常会加载数据集、运行转换并设置您的功能以进行提取。此过程有很多变化，高度依赖于您的数据。本主题中的示例代码引用了 Amazon 欺诈检测SageMakerFeatureStore示例笔记本。我们建议您在中运行此笔记本Amazon SageMaker Studio，因为本指南中的代码是概念性的，如果复制，则不能完全正常运行。

Feature Store 支持以下数据类型String：、 Fractional （IEE 64-bit浮点值）和 Integral （Int64 - 64 位有符号整数值）。默认类型设置为 String。这意味着，如果数据集中的列不是 float 或 long 类型，则它在功能存储String中默认为。

您可以使用架构来描述数据的列和数据类型。将此架构传递到（ FeatureDefinitions的必需参数FeatureGroup）。您可以使用 SageMaker Python 开发工具包，该开发工具包在使用 load_feature_definitions函数时具有自动数据类型检测功能。

主题

步骤 1：设置 Feature Store
步骤 2：将数据集和分区数据加载到功能组中
步骤 3：设置功能组
步骤 4：设置记录标识符和事件时间功能
步骤 5：加载功能定义
步骤 6：创建功能组
步骤 7：使用功能组

步骤 1：设置 Feature Store

要开始使用 Feature Store，请创建 SageMaker 会话boto3 会话和 Feature Store 会话。此外，设置要用于您的功能的 S3 存储桶。这是您的脱机存储。以下代码使用SageMaker默认存储桶并向其添加自定义前缀。

注意

您使用的角色必须附加了以下托管策略： AmazonSageMakerFullAccess 和 AmazonSageMakerFeatureStoreAccess。


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

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)

例如，在欺诈检测示例中，两个功能组为 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，并且此函数会自动检测每列数据的数据类型。对于使用架构而不是自动检测的开发人员，请参阅从数据 Wrangler https://docs.amazonaws.cn/sagemaker/latest/dg/data-wrangler-data-export.html#data-wrangler-data-export-feature-store 导出功能组示例以了解展示如何加载架构、映射架构并添加为可用于创建 FeatureDefinition 的的代码FeatureGroup。此示例还介绍了 boto3 实施，您可以使用它而不是 SageMaker Python 开发工具包。


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"])

欺诈检测示例中的以下代码显示了正在创建的两个功能组中的每个功能组的最小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：使用功能组

现在，您已设置功能组，您可以执行以下任一任务：

主题

描述功能组
列出功能组
将记录放入功能组中
从功能组中获取记录
生成 Hive DDL 命令
构建训练数据集
编写和执行 Athena 查询
删除功能组

描述功能组

您可以使用 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 开发工具包的 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 数据目录，并且您可以根据需要将其关闭。下面介绍如何使用来自本主题前面部分创建的身份和事务功能组的功能值创建单个训练数据集。此外，下面介绍了如何运行 Amazon Athena 查询以联接存储在脱机存储中来自身份和事务功能组的数据。

要开始，请使用 Athena 为身份和事务功能组创建 athena_query() 查询。“table_name”是自动生成的AWS Glue表Feature Store。


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() 命令执行查询，并指定要将数据集保存到其中的 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

Javascript 在您的浏览器中被禁用或不可用。

要使用 AWS 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

开始使用 Amazon SageMaker Feature Store

将所需的策略添加到您的 IAM 角色