4. 特征工程生产化

在真实生产环境中，特征工程往往不是“算完一次就结束”，而是要在训练、批处理评分、实时推理三条路径上持续复用同一批特征。如果每个团队各自写 SQL、各自维护一份特征逻辑，就会出现训练/推理不一致（training-serving skew）、重复计算、口径漂移等问题。Databricks Feature Store 的核心价值，就是把特征从“散落在 Notebook 里的代码”变成“可发现、可复用、可追溯、可在线服务”的一等资产。

一、Feature Store 到底解决什么问题

Feature Store 是一个集中登记特征定义、存储特征数据、并对外提供读写接口的组件。它同时承担四件事：

• 特征表（Feature Table）：以 Delta 表形式持久化特征值，带主键（entity key）和时间戳（timestamp key），支持按时间点正确回看历史特征。
• 元数据与血缘：记录特征来源、计算逻辑、所属实体、版本，便于跨团队检索与审计。
• 离线服务：通过 create_training_set 按时间点 join，生成无泄漏的训练集。
• 在线服务：把特征同步到低延迟存储（如在线表/键值库），供实时推理按实体键毫秒级读取。

关键概念是实体（Entity）与时间戳键。实体是特征的主体，例如 user_id、product_id；时间戳键保证“用某时刻之前已知的特征”去预测该时刻的目标，避免未来信息泄漏。离线与在线服务共享同一份特征定义，这是消除训练/推理不一致的根本手段。

二、创建特征表：从 DataFrame 到可服务特征

在 Databricks 中，创建特征表通常从一段特征计算逻辑开始。假设我们要为电商推荐场景构建用户级特征，主键为 user_id，时间戳为 event_ts，包含近 7 天点击数和近 30 天成交额。

from databricks.feature_store import FeatureStoreClient
from pyspark.sql.functions import col, count, sum, window

fs = FeatureStoreClient()

# 1. 计算特征（示例逻辑）
user_features = (
    events
    .groupBy("user_id", window("event_ts", "1 day").alias("event_ts"))
    .agg(
        count("click_id").alias("click_7d"),
        sum("gmv").alias("gmv_30d")
    )
    .select("user_id", col("event_ts.start").alias("event_ts"), "click_7d", "gmv_30d")
)

# 2. 创建特征表
fs.create_table(
    name="prod_ml.reco.user_features",
    primary_keys=["user_id"],
    timestamp_keys=["event_ts"],
    df=user_features,
    description="用户行为特征：近7天点击、近30天成交额"
)

这里有两个容易踩坑的点。第一，主键必须唯一且非空，否则在线写入会失败；第二，时间戳键决定历史回看语义，如果缺失，训练集 join 只能做“最新值”匹配，容易引入未来信息。创建完成后，特征表会以 Delta 表形式出现在 Unity Catalog 中，同时注册进 Feature Store 元数据。

三、离线特征服务：时间点正确的训练集

离线服务的核心是 create_training_set。它接收一个带主键和时间戳的标签 DataFrame，按时间点把特征“回看”到标签发生之前，从而避免泄漏。

# 标签表：user_id, label, label_ts
training_set = fs.create_training_set(
    df=labels,
    feature_lookups=[
        FeatureLookup(
            table_name="prod_ml.reco.user_features",
            feature_names=["click_7d", "gmv_30d"],
            lookup_key="user_id",
            timestamp_lookup_key="label_ts"
        )
    ],
    label="label",
    exclude_columns=["event_ts"]
)

training_df = training_set.load_df()

执行后，training_df 会包含 user_id、label 以及两个特征列。判断结果是否正确，可以检查三点：特征列无空值比例是否合理、时间戳是否严格小于标签时间、行数是否与标签表一致。若某用户在某时刻没有历史特征，Databricks 会返回 null，此时需要在训练前做填充或过滤。

四、在线特征服务：低延迟读取

在线服务需要把特征表发布到在线存储。Databricks 支持通过 publish_table 将 Delta 表同步到在线表（Online Table），底层通常是低延迟键值存储。

fs.publish_table(
    name="prod_ml.reco.user_features",
    online_table_name="prod_ml.reco.user_features_online",
    mode="merge"   # 增量合并，避免全量重刷
)

发布后，推理服务可以通过 Feature Store SDK 或在线表 REST 接口按 user_id 读取特征。典型延迟目标是个位数毫秒到几十毫秒。需要注意：在线表只保留每个主键的最新值，历史版本仅存在于离线 Delta 表中。因此在线服务适合“当前状态”类特征，而需要时间点回看的训练场景必须走离线通道。

五、离线与在线的协同关系

把这三件事串起来看：特征表是唯一事实来源，离线服务负责“正确地训练”，在线服务负责“快速地推理”，两者共享同一份定义，才真正实现特征工程的生产化。实际落地时，建议先从一个高价值实体（如用户）和 3～5 个稳定特征开始，跑通训练到在线的完整链路，再逐步扩展特征数量和实体类型。