步骤 1：设置步骤 2：检查数据步骤 3：创建特征组步骤 4：将数据摄取到特征组第 5 步：清理步骤 6：后续步骤步骤 7：编程人员备注

“Feature Store 简介”示例笔记本

本页上的示例代码是指 Feature Store 简介示例笔记本。建议您在 Amazon SageMaker Studio、Jupyter 和 JupyterLab 中运行该笔记本，因为本指南中的代码是概念性的，复制后不能完全发挥作用。要在 Studio 中打开笔记本，请先按照在 SageMaker Studio 中克隆 Git 存储库中的步骤将 aws/amazon-sagemaker-examples GitHub 存储库克隆到 Studio，然后导航到 amazon-sagemaker-examples/sagemaker-featurestore 目录并选择文件。

步骤 1：设置

要开始使用 Feature Store，请创建 SageMaker 会话并设置要用于特征的 Amazon S3 存储桶。Amazon S3 存储桶是您的离线存储。以下代码使用 SageMaker 默认存储桶并为其添加自定义前缀。

注意

用于运行笔记本的角色必须附加以下托管策略：AmazonS3FullAccess 和 AmazonSageMakerFeatureStoreAccess。有关向 IAM 角色添加策略的信息，请参阅向您的 IAM 角色添加策略。


# 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 之前所经历的步骤。在此笔记本中，我们说明了一个使用案例，即您拥有来自多个数据源的数据，并希望将它们独立存储在 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())

为 customers_data 和 orders_data 实例化一个 FeatureGroup 对象。


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
)

为了确认您的特征组已创建，我们使用 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：将数据摄取到特征组

创建 FeatureGroups 后，我们可以将数据放入 FeatureGroups。如果您使用的是 SageMaker Python SDK，请使用 ingest API 调用。如果您使用的是 boto3，则使用 PutRecord API。这两个选项都只需不到 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 573291，利用 get_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：后续步骤

在此示例笔记本中，您学习了如何快速开始使用 Feature Store，如何创建特征组，以及如何向其中摄取数据。

有关如何在欺诈检测使用案例中使用 Feature Store 的高级示例，请参阅使用 Feature Store 进行欺诈检测。

步骤 7：编程人员备注

在此笔记本中，我们使用了各种不同的 API 调用。其中大部分可以通过 Python SDK 访问，但有些只能在 boto3 中使用。您可以直接在 Feature Store 对象上调用 Python SDK API 调用，而要调用 boto3 中存在的 API 调用，则必须先通过 boto3 和 SageMaker 会话（例如 sagemaker_session.boto_session.client()）访问 boto3 客户端。

下面列出了此笔记本中使用且存在于 Python SDK 中的 API 调用和存在于 boto3 中的 API 调用，以供参考。

Python SDK API 调用


describe()
ingest()
delete()
create()
load_feature_definitions()

Boto3 API 调用


list_feature_groups()
get_record()

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

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

将 Feature Store 与 SDK for Python (Boto3) 结合使用

“使用 Feature Store 进行欺诈检测”示例笔记本