创建功能组 - Amazon SageMaker
AWS 文档中描述的 AWS 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 AWS 服务入门

本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。

创建功能组

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。 这意味着,如果数据集中的列不是 floatlong 类型,则它在功能存储String中默认为 。

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

步骤 1:设置 Feature Store

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

注意

您使用的 角色必须附加了以下 托管策略: AmazonSageMakerFullAccessAmazonSageMakerFeatureStoreAccess

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)

例如,在欺诈检测示例中,两个功能组为 identitytransaction。 在以下代码中,您可以查看如何使用时间戳自定义名称,然后通过传入名称和会话来设置每个组。

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。 其他可能的状态为 CreateFailedDeletingDeleteFailed

步骤 7:使用功能组

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

描述功能组

您可以使用 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