本文属于机器翻译版本。若本译文内容与英语原文存在差异,则一律以英文原文为准。
功能商店简介
本主题中的示例代码引用Amazon SageMaker Feature Store 简介
第 1 步:设置
要开始使用要素商店,请创建 SageMaker、boto3 和要素商店会话。然后,您要用于功能的 S3 存储桶。这是你的线下商店。以下代码使用 SageMaker 默认存储桶并向其添加自定义前缀。
您使用的角色必须附加以下托管策略:AmazonS3FullAccess
和AmazonSageMakerFeatureStoreAccess
.
# SageMaker Python SDK version 2.x is required import sagemaker import sys
import boto3 import pandas as pd import numpy as np import io from sagemaker.session import Session from sagemaker import get_execution_role prefix = 'sagemaker-featurestore-introduction' role = get_execution_role() sagemaker_session = sagemaker.Session() region = sagemaker_session.boto_region_name s3_bucket_name = sagemaker_session.default_bucket()
第 2 步:检查您的数据
在本笔记本示例中,我们从GitHub 仓库
customer_data = pd.read_csv("data/feature_store_introduction_customer.csv") orders_data = pd.read_csv("data/feature_store_introduction_orders.csv") print(customer_data.head()) print(orders_data.head())
下图说明了数据在将数据提取到要素商店之前所经历的步骤。在本笔记本中,我们展示了一个用例,即您拥有来自多个来源的数据并希望将它们独立存储在 feature store 中。我们的示例考虑来自数据仓库的数据(客户数据)和来自实时流媒体服务的数据(订单数据)。

第 3 步:创建功能组
首先,我们首先为 customer_data 和 orders_data 创建要素组名称。之后,我们创建了两个功能组,一个用于customer_data
另一个用于orders_data
.
import time from time import strftime, gmtime customers_feature_group_name = 'customers-feature-group-' + strftime('%d-%H-%M-%S', gmtime()) orders_feature_group_name = 'orders-feature-group-' + strftime('%d-%H-%M-%S', gmtime())
实例化FeatureGroup
对象customers_data
和orders_data
.
from sagemaker.feature_store.feature_group import FeatureGroup customers_feature_group = FeatureGroup( name=customers_feature_group_name, sagemaker_session=sagemaker_session ) orders_feature_group = FeatureGroup( name=orders_feature_group_name, sagemaker_session=sagemaker_session )
import time current_time_sec = int(round(time.time())) record_identifier_feature_name = "customer_id"
附加EventTime
要素到你的数据框中。此参数是必需的,并且每个数据点的时间戳。
customer_data["EventTime"] = pd.Series([current_time_sec]*len(customer_data), dtype="float64") orders_data["EventTime"] = pd.Series([current_time_sec]*len(orders_data), dtype="float64")
将要素定义加载到功能组。
customers_feature_group.load_feature_definitions(data_frame=customer_data) orders_feature_group.load_feature_definitions(data_frame=orders_data)
下面我们称之为 create 来创建两个功能组,customers_feature_group
和orders_feature_group
。
customers_feature_group.create( s3_uri=f"s3://{s3_bucket_name}/{prefix}", record_identifier_name=record_identifier_feature_name, event_time_feature_name="EventTime", role_arn=role, enable_online_store=True ) orders_feature_group.create( s3_uri=f"s3://{s3_bucket_name}/{prefix}", record_identifier_name=record_identifier_feature_name, event_time_feature_name="EventTime", role_arn=role, enable_online_store=True )
为了确认您的 FeatureGroup 已创建,我们使用DescribeFeatureGroup
和ListFeatureGroups
用于显示创建的功能组的 API。
customers_feature_group.describe()
orders_feature_group.describe()
sagemaker_session.boto_session.client('sagemaker', region_name=region).list_feature_groups() # We use the boto client to list FeatureGroups
第 4 步:将数据提取到要素组
在 FeatuRegGroups 创建之后,我们可以将数据放入 FeatuRegGroups 中。如果您使用的是 SageMaker Python SDK,请使用ingest
API 调用。如果你通过使用 boto3 使用,那么请使用PutRecord
API。摄取这两个 FageGroups 都需要不到 1 分钟的时间。本示例使用 SageMaker Python SDK,因此它使用ingest
API 调用。
def check_feature_group_status(feature_group): status = feature_group.describe().get("FeatureGroupStatus") while status == "Creating": print("Waiting for Feature Group to be Created") time.sleep(5) status = feature_group.describe().get("FeatureGroupStatus") print(f"FeatureGroup {feature_group.name} successfully created.") check_feature_group_status(customers_feature_group) check_feature_group_status(orders_feature_group)
customers_feature_group.ingest( data_frame=customer_data, max_workers=3, wait=True )
orders_feature_group.ingest( data_frame=orders_data, max_workers=3, wait=True )
使用任意客户记录 ID,我们使用 573291get_record
以检查数据是否已被摄入到功能组。
customer_id = 573291 sample_record = sagemaker_session.boto_session.client('sagemaker-featurestore-runtime', region_name=region).get_record(FeatureGroupName=customers_feature_group_name, RecordIdentifierValueAsString=str(customer_id))
print(sample_record)
下面演示如何使用batch_get_record
以获取一批记录。
all_records = sagemaker_session.boto_session.client( "sagemaker-featurestore-runtime", region_name=region ).batch_get_record( Identifiers=[ { "FeatureGroupName": customers_feature_group_name, "RecordIdentifiersValueAsString": ["573291", "109382", "828400", "124013"], }, { "FeatureGroupName": orders_feature_group_name, "RecordIdentifiersValueAsString": ["573291", "109382", "828400", "124013"], }, ] )
print(all_records)
第 5 步:清除
在这里我们删除我们创建的功能组。
customers_feature_group.delete() orders_feature_group.delete()
第 6 步:后续步骤
在本示例笔记本中,您学习了如何快速开始使用功能商店、创建要素组以及将数据提取到它们中。
有关如何将功能商店用于欺诈检测用例的高级示例,请参阅使用功能商店检测欺诈
步骤 7:程序员注意
在这本笔记本中,我们使用了各种不同的 API 调用。其中大多数都可以通过 Python SDK 访问,但是有些只存在于 boto3 中。您可以直接在功能存储对象上调用 Python SDK API 调用,而要调用 boto3 中存在的 API 调用,必须首先通过 boto 和 sagemaker 会话访问 boto 客户端:例如sagemaker_session.boto_session.client()
.
下面我们列出了本笔记本中使用的 Python SDK 中存在的 API 调用以及 boto3 中存在的 API 调用以供参考。
Python 开发工具包 API 调用
describe() ingest() delete() create() load_feature_definitions()
Boto3 API 调用
list_feature_groups() get_record()