本文属于机器翻译版本。若本译文内容与英语原文存在差异，则一律以英文原文为准。
使用功能商店检测欺诈
第 1 步：设置功能商店
要开始使用要素商店，请创建 SageMaker 会话、boto3 会话和要素商店会话。此外，您要用于功能的 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，此函数会自动检测每列数据的数据类型。对于使用模式而不是自动检测的开发人员，请参阅从 Data Wrangler 中导出要素组代码示例，该代码显示如何加载架构、映射它并将其添加为FeatureDefinition您可以使用创建FeatureGroup. 此示例还介绍了 boto3 实现，您可以使用它代替 SageMaker Python SDK。
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：使用功能组
现在您已经设置了功能组，可以执行以下任意任何任务：
描述功能组
您可以使用describefunction.
feature_group.describe()
列出功能组
您可以使用列出所有功能组list_feature_groupsfunction.
sagemaker_client.list_feature_groups()
将记录放入要素组
您可以使用ingest函数来加载要素数据。您可以传入要素数据的数据框，设置工作人员的数量，然后选择等待它返回或不返回。以下示例演示如何使用ingestfunction.
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 ...
构建训练数据集
功能商店会自动构建Amazon Glue创建要素组时的数据目录，如果需要，可以将其关闭。下面介绍了如何使用本主题前面创建的身份和事务要素组的要素值创建单个训练数据集。此外，以下内容介绍了如何运行 Amazon Athena 查询来加入存储在离线商店中的身份和交易功能组中的数据。
首先，请使用创建 Athena 查询athena_query()适用于身份和交易功能组。`table_name` 是Amazon Glue由要素商店自动生成的表。
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()
从这里你可以使用此数据集训练模型，然后执行推理。
删除功能组
您可以使用删除功能组deletefunction.
feature_group.delete()
以下代码示例来自于欺诈检测示例。
identity_feature_group.delete() transaction_feature_group.delete()
有关更多信息，请参阅 。删除功能组 API