4. 特征存储与版本管理

特征存储（Feature Store）是特征工程体系中的核心基础设施，它解决了特征在生产环境中的共享、复用、一致性和版本追溯问题。在特征回放对比场景中，特征存储不仅需要保存特征值，还必须记录特征随时间的演变过程，即特征快照与版本标签。

4.1 特征仓库的核心概念

特征仓库是一个专门用于管理机器学习特征的系统层组件，它统一了特征的定义、存储、服务和发现流程。与传统的特征文件存储相比，特征仓库提供了以下关键能力：

4.2 使用 Feast 管理特征快照

Feast 是目前最成熟的开源特征仓库方案，它通过特征视图（Feature View）和数据源（Data Source）的组合来管理特征快照。每个特征视图可以关联一个时间戳字段，Feast 利用该字段实现时间点查询。

4.2.1 特征视图与时间戳定义

from datetime import datetime, timedelta
from feast import FeatureView, Field, FileSource
from feast.types import Float32, Int64, String

# 定义数据源，指定时间戳列
transaction_source = FileSource(
    path="s3://feature-bucket/transactions/",
    timestamp_field="event_timestamp",       # 事件时间戳
    created_timestamp_column="created_at",   # 记录创建时间
)

# 定义特征视图，设置 TTL 和版本标签
transaction_features = FeatureView(
    name="transaction_rolling_stats",
    entities=["user_id"],
    ttl=timedelta(days=90),                  # 特征有效期
    schema=[
        Field(name="avg_7d_amount", dtype=Float32),
        Field(name="count_30d_txns", dtype=Int64),
        Field(name="last_category", dtype=String),
    ],
    source=transaction_source,
    tags={"version": "v2.1", "team": "risk"}, # 版本标签
)

4.2.2 时间点查询与快照获取

Feast 的 get_historical_features 方法接受一个包含实体 ID 和查询时间戳的 DataFrame，自动为每个时间点匹配该时刻有效的特征快照：

import pandas as pd
from feast import FeatureStore

store = FeatureStore(repo_path="./feature_repo")

# 构造回放查询：每个样本需要指定查询时间点
query_df = pd.DataFrame({
    "user_id": ["U001", "U002", "U001"],
    "event_timestamp": [
        pd.Timestamp("2024-06-01 12:00:00"),
        pd.Timestamp("2024-06-02 15:30:00"),
        pd.Timestamp("2024-06-10 09:00:00"),
    ],
})

# 获取历史特征快照
feature_df = store.get_historical_features(
    entity_df=query_df,
    features=[
        "transaction_rolling_stats:avg_7d_amount",
        "transaction_rolling_stats:count_30d_txns",
    ],
).to_df()

print(feature_df.head())

4.3 自定义特征快照存储方案

当 Feast 等开源方案无法满足特定需求（如私有云环境、自定义存储引擎、特殊版本策略）时，可以构建轻量级自定义方案。核心设计包括三个层次：

1. 存储层：使用 Delta Lake / Iceberg 等支持时间旅行的表格式，或按时间分区存储 Parquet 文件。
2. 元数据层：记录每个特征快照的版本标签、生效时间、计算时间、特征 Schema。
3. 查询层：提供按实体 + 时间戳获取快照的 API。

4.3.1 快照存储结构设计

# 特征快照目录结构示例
feature_snapshots/
  └── user_features/
      ├── version=v2.1/
      │   ├── snapshot_date=2024-06-01/
      │   │   └── data.parquet
      │   ├── snapshot_date=2024-06-02/
      │   │   └── data.parquet
      │   └── _snapshot_meta.json
      └── version=v2.2/
          ├── snapshot_date=2024-06-10/
          │   └── data.parquet
          └── _snapshot_meta.json

# _snapshot_meta.json 元数据示例
{
  "feature_view": "user_features",
  "version": "v2.1",
  "created_at": "2024-06-01T02:00:00Z",
  "valid_from": "2024-06-01T00:00:00Z",
  "valid_to": "2024-06-10T00:00:00Z",
  "features": ["avg_7d_amount", "count_30d_txns"],
  "computation_hash": "a1b2c3d4e5f6",
  "source_data_range": {
    "start": "2024-05-01",
    "end": "2024-05-31"
  }
}

4.3.2 快照查询实现

import pandas as pd
from pathlib import Path
import json

class CustomFeatureStore:
    def __init__(self, base_path: str):
        self.base_path = Path(base_path)

    def get_snapshot(self, feature_view: str, version: str,
                     entity_ids: list, timestamp: pd.Timestamp):
        """获取指定时间点的特征快照"""
        # 1. 定位版本目录
        version_path = self.base_path / feature_view / f"version={version}"

        # 2. 加载元数据，确定有效时间范围
        with open(version_path / "_snapshot_meta.json") as f:
            meta = json.load(f)

        valid_from = pd.Timestamp(meta["valid_from"])
        valid_to = pd.Timestamp(meta["valid_to"])

        if not (valid_from <= timestamp < valid_to):
            raise ValueError(f"Timestamp {timestamp} out of version range")

        # 3. 找到最近的快照日期
        snapshot_date = timestamp.strftime("%Y-%m-%d")
        snapshot_path = version_path / f"snapshot_date={snapshot_date}"

        if not snapshot_path.exists():
            # 回退到最近可用快照
            snapshot_path = self._find_nearest_snapshot(version_path, timestamp)

        # 4. 加载数据并过滤实体
        df = pd.read_parquet(snapshot_path / "data.parquet")
        return df[df["entity_id"].isin(entity_ids)]

    def _find_nearest_snapshot(self, version_path, timestamp):
        """查找不超过目标时间戳的最近快照"""
        snapshots = sorted(version_path.glob("snapshot_date=*"))
        target_str = timestamp.strftime("%Y-%m-%d")
        for snap in reversed(snapshots):
            if snap.name.split("=")[1] <= target_str:
                return snap
        return snapshots[0]  # 兜底返回最早快照

4.4 特征时间戳与版本标签设计

特征版本管理是回放对比实验准确性的基石。一个设计良好的版本标签系统需要同时表达三个时间维度：

• 事件时间（Event Time）：特征对应的业务事件发生时间，用于时间点匹配。
• 快照时间（Snapshot Time）：特征值被计算并存储的时间，反映特征的新鲜度。
• 版本时间（Version Time）：特征定义或计算逻辑变更的时间，用于版本追溯。

4.4.1 版本标签命名规范

4.4.2 版本标签在回放中的应用

在特征回放对比实验中，版本标签用于精确控制每个时间点应该使用哪个特征版本：

# 回放对比实验：比较 v2.0 和 v2.1 版本的特征效果
experiment_config = {
    "experiment_id": "exp_004",
    "feature_view": "user_features",
    "versions": {
        "baseline": {
            "version_tag": "v2.0",
            "valid_range": ("2024-01-01", "2024-06-30"),
            "description": "原始 7 天滚动窗口"
        },
        "candidate": {
            "version_tag": "v2.1",
            "valid_range": ("2024-03-01", "2024-06-30"),
            "description": "优化后的 14 天加权窗口"
        }
    },
    "overlap_period": ("2024-03-01", "2024-06-30"),  # 两个版本都有效的重叠期
    "query_timestamps": pd.date_range("2024-03-01", "2024-06-30", freq="7D")
}

# 在重叠期内分别获取两个版本的特征快照
for version_info in experiment_config["versions"].values():
    snapshot = store.get_snapshot(
        feature_view=experiment_config["feature_view"],
        version=version_info["version_tag"],
        entity_ids=["U001", "U002"],
        timestamp=pd.Timestamp("2024-04-15")
    )
    # 进行回放对比分析...

4.5 特征存储架构总览

下图展示了特征仓库在特征回放对比流程中的核心位置，以及各组件之间的数据流关系：

4.6 版本管理策略对比

根据团队规模和业务复杂度，可以选择不同的版本管理策略：