📚 课程目录第4章特征存储实现

4. 特征存储实现

任务目标

使用 Feast 搭建一个可运行的特征存储（Feature Store），完成以下闭环：定义特征视图（Feature View）、将特征数据注册到在线存储（Online Store），并通过 Python SDK 在线获取特征。本案例聚焦 Feast 的核心用法，不涉及生产级部署与权限配置。

环境与依赖

需要 Python 3.8+ 环境，安装以下依赖（建议使用虚拟环境）：

pip install "feast==0.39.1" pandas redis

Feast 默认使用 SQLite 作为在线存储（适用于本地开发），无需额外启动服务。

实现步骤与代码

本案例包含两个文件：feature_repo.py（定义特征存储配置与特征视图）和 online_retrieval.py（注册特征并在线获取）。

文件 1：feature_repo.py — 定义项目配置、实体、特征视图与数据源。

from datetime import timedelta
import pandas as pd
from feast import Entity, FeatureView, Field, FileSource
from feast.types import Float32, Int64

# 项目配置（使用本地 SQLite 在线存储）
project = "fraud_detection_demo"
registry = "data/registry.db"
provider = "local"
online_store = {"type": "sqlite", "path": "data/online_store.db"}

# 实体：用户
user = Entity(
    name="user_id",
    join_keys=["user_id"],
    description="用户ID",
)

# 模拟历史数据（离线存储）
df = pd.DataFrame({
    "user_id": [1001, 1002, 1003],
    "avg_transaction_amount": [250.5, 89.2, 1500.0],
    "transaction_count_7d": [12, 5, 3],
    "event_timestamp": pd.to_datetime(["2024-01-01", "2024-01-01", "2024-01-01"]),
})

# 将 DataFrame 写入本地 parquet 文件作为离线数据源
df.to_parquet("data/transactions.parquet")

# 离线数据源
source = FileSource(
    path="data/transactions.parquet",
    timestamp_field="event_timestamp",
)

# 特征视图：用户交易统计特征
user_transaction_stats = FeatureView(
    name="user_transaction_stats",
    entities=[user],
    schema=[
        Field(name="avg_transaction_amount", dtype=Float32),
        Field(name="transaction_count_7d", dtype=Int64),
    ],
    source=source,
    ttl=timedelta(days=7),  # 特征有效期
)

文件 2：online_retrieval.py — 注册特征仓库、将特征写入在线存储并执行在线获取。

from feast import FeatureStore

# 初始化特征存储（指向当前目录下的 feature_repo.py）
store = FeatureStore(repo_path=".")

# 将离线特征应用到在线存储（SQLite）
store.apply()

# 模拟新到的用户请求，需要获取 user_id=1001 的实时特征
features = store.get_online_features(
    features=[
        "user_transaction_stats:avg_transaction_amount",
        "user_transaction_stats:transaction_count_7d",
    ],
    entity_rows=[{"user_id": 1001}],
).to_dict()

print("在线获取的特征结果：")
print(features)

运行与验证

在包含上述两个文件的目录下依次执行：

# 1. 确保 data/ 目录存在
mkdir -p data

# 2. 运行特征注册与在线获取脚本
python online_retrieval.py

验证通过条件：程序无报错退出，且输出结果中包含 user_id=1001 对应的特征值，即 avg_transaction_amount 为 250.5，transaction_count_7d 为 12。该结果由代码中给定的 DataFrame 直接推出，属于确定性预期输出。

关键说明

Feast 的 apply() 会读取当前目录下的 feature_repo.py 并自动注册实体、特征视图与数据源，同时将特征物化到在线存储。
本案例使用 SQLite 作为在线存储，仅用于本地验证；生产环境可替换为 Redis 或 DynamoDB，只需修改 online_store 配置。
离线数据源使用本地 parquet 文件，实际生产可替换为 BigQuery、Redshift 等 Feast 支持的离线存储。