“Feature Store 简介”示例笔记本

本页上的示例代码是指 Feature Store 简介示例笔记本。我们建议您在 Studio Classic、笔记本实例或 JupyterLab 中运行本笔记本，因为本指南中的代码是概念性的，复制后无法完全发挥作用。

使用下面的方法克隆 aws/amazon-sagemaker-examples GitHub 存储库，其中包含示例笔记本：

现在您已经有了 SageMaker 示例笔记本，请导航至 amazon-sagemaker-examples/sagemaker-featurestore 目录并打开特征存放区介绍示例笔记本。

步骤 1：设置 SageMaker 会话

要开始使用特征存放区，请创建一个 SageMaker 会话。然后，设置要用于功能的 Amazon Simple Storage Service (Amazon S3) 存储桶。Amazon S3 存储桶是您的离线存储。以下代码使用 SageMaker 默认存储桶并为其添加自定义前缀。

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

下图说明了数据在特征存放区摄取之前要经过的步骤。在本笔记本中，我们将举例说明从多个来源获取数据并希望将其独立存储在特征存放区中的使用场景。我们的示例考虑了来自数据仓库的数据（客户数据）和来自实时流式传输服务的数据（订单数据）。

步骤 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：将数据摄取到特征组

创建特征组后，我们就可以将数据放入特征组中。如果您使用 SageMaker AWS SDK for Python (Boto3)，请使用 ingest API 调用。如果您使用的是 Python SDK (Boto3)，那么请使用 PutRecord API。应用程序接口。这两个选项都只需不到 1 分钟就能摄取数据。本示例使用 Python 版 SageMaker SDK (Boto3)，因此使用了 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：后续步骤

在本示例笔记本中，您将学习如何开始使用特征存放区、创建特征组并将数据导入其中。

有关如何将特征存放区用于欺诈检测使用场景的高级示例，请参阅使用特征存放区进行欺诈检测。

步骤 7：为程序员提供代码示例

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

以下是本笔记本的 API 调用列表。这些调用存在于 SDK for Python 中，也存在于 Boto3 中，供您参考：

SDK for Python (Boto3) API Calls

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

Boto3 API 调用

list_feature_groups()
get_record()