功能商店简介 - Amazon SageMaker
Amazon Web Services 文档中描述的 Amazon Web Services 服务或功能可能因区域而异。要查看适用于中国区域的差异,请参阅中国的 Amazon Web Services 服务入门

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

功能商店简介

本主题中的示例代码引用Amazon SageMaker Feature Store 简介笔记本示例。建议您在 Amazon SageMaker Studio 中运行此笔记本,因为本指南中的代码是概念性的,如果复制,则无法完全正常工作。

第 1 步:设置

要开始使用要素商店,请创建 SageMaker、boto3 和要素商店会话。然后,您要用于功能的 S3 存储桶。这是你的线下商店。以下代码使用 SageMaker 默认存储桶并向其添加自定义前缀。

注意

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

# 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_dataorders_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_grouporders_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 已创建,我们使用DescribeFeatureGroupListFeatureGroups用于显示创建的功能组的 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,请使用ingestAPI 调用。如果你通过使用 boto3 使用,那么请使用PutRecordAPI。摄取这两个 FageGroups 都需要不到 1 分钟的时间。本示例使用 SageMaker Python SDK,因此它使用ingestAPI 调用。

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()