第 1 步：设置第 2 步：检查您的数据第 3 步：创建功能组第 4 步：将数据提取到要素组第 5 步：清除第 6 步：后续步骤步骤 7：程序员注意

功能商店简介

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

第 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，请使用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()

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

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

创建功能组

使用功能商店检测欺诈