3. 特征回放系统架构：离线回放 vs 在线回放、回放管道设计

特征回放系统的架构设计直接决定了回放实验的准确性、时效性与资源成本。在 A/B 测试场景中，我们需要根据实验目标（离线验证 vs 线上实时干预）选择不同的回放模式，并构建标准化的回放管道。

3.1 离线回放 vs 在线回放

两种模式的核心差异在于“回放时机”与“数据来源”。离线回放使用历史日志批量计算特征，适合模型迭代与策略验证；在线回放则基于实时事件流，在线上服务中即时计算特征，适合低延迟干预实验。

3.2 回放管道设计

标准化的特征回放管道由四个核心阶段组成：数据源 → 特征计算 → 存储 → 服务。每个阶段在离线与在线模式下有不同的技术选型与处理逻辑。

3.2.1 管道四阶段详解

1. 数据源（Data Source）
   • 离线：从 Hive 表或 HDFS 文件中读取用户行为日志、物品属性快照、上下文日志。
   • 在线：从 Kafka Topic 中消费实时行为事件，并关联 Redis 中的最新状态数据。
   • 关键要求：数据源必须包含精确的时间戳（event_time），用于对齐回放时间窗口。
2. 特征计算（Feature Computation）
   • 离线：使用 Spark SQL 或 PySpark 执行特征 SQL 逻辑，支持窗口聚合、跨表 Join、UDF 扩展。
   • 在线：使用 Flink SQL 或自定义 ProcessFunction 实现增量特征计算，维护状态后端（RocksDB）。
   • 关键要求：离线与在线必须使用同一份特征配置（如 YAML 或 JSON 定义的特征元数据），确保计算逻辑一致。
3. 存储（Storage）
   • 离线：计算结果写入 Hive 分区表或 Parquet 格式的列存文件，按日期 + 实验标识分区。
   • 在线：计算结果写入在线特征存储（如 Redis Cluster / HBase），设置 TTL 以控制内存占用。
   • 关键要求：存储层需支持时间旅行查询，即能够根据回放时间戳获取当时计算的特征值。
4. 服务（Serving）
   • 离线：通过 Presto / Trino 或 Spark Thrift Server 提供 SQL 查询接口，供分析师与模型训练脚本调用。
   • 在线：通过高并发 RPC 服务（gRPC / Thrift）暴露特征获取接口，响应时间 < 10ms。
   • 关键要求：服务层需支持特征版本管理，确保实验组与对照组使用同一版本的特征逻辑。

3.3 系统组件图

下图展示了特征回放系统的完整组件架构，涵盖离线与在线两条管道，以及它们共享的特征元数据层与存储层。

3.4 管道设计关键考量

• 特征一致性保障：离线与在线管道必须使用同一份特征计算代码或配置。推荐将特征逻辑封装为独立的计算库（如 Java/Python SDK），在批处理和流处理中统一调用。
• 时间旅行能力：存储层需要支持按历史时间戳查询特征值。离线场景下通过 Hive 分区字段实现；在线场景下通过 Redis 的带时间戳的 Sorted Set 或 HBase 的版本号实现。
• 实验隔离：服务层需根据实验标识（experiment_id）路由到对应的特征版本，确保实验组与对照组使用正确的特征逻辑，避免相互干扰。
• 监控与告警：在管道各阶段埋点监控数据延迟、计算耗时、特征缺失率等指标。离线管道关注数据产出时间（SLA），在线管道关注 p99 延迟与吞吐量。

3.5 回放管道代码示例

以下是一个简化的特征回放管道配置示例，展示如何通过 YAML 统一声明特征计算逻辑，并分别用于离线 Spark 任务与在线 Flink 任务。

# feature_replay_config.yaml
features:
  - name: user_click_count_7d
    type: window_agg
    window: 7d
    source: user_behavior_log
    aggregation: count
    filter: action_type = 'click'
    offline_sql: |
      SELECT user_id, COUNT(*) AS click_count_7d
      FROM user_behavior_log
      WHERE action_type = 'click'
        AND event_time >= DATE_SUB('${replay_date}', 7)
        AND event_time < '${replay_date}'
      GROUP BY user_id
    online_flink: |
      SELECT user_id, COUNT(*) AS click_count_7d
      FROM user_behavior_stream
      WHERE action_type = 'click'
      GROUP BY user_id, TUMBLE(event_time, INTERVAL '7' DAY)

  - name: item_price
    type: lookup
    source: item_catalog
    key: item_id
    offline_sql: |
      SELECT item_id, price AS item_price
      FROM item_catalog_snapshot
      WHERE dt = '${replay_date}'
    online_flink: |
      -- 从 Redis 中 lookup 最新价格
      CREATE TEMPORARY TABLE item_catalog_redis (
        item_id STRING,
        price DOUBLE,
        PRIMARY KEY (item_id) NOT ENFORCED
      ) WITH (
        'connector' = 'redis',
        'redis-mode' = 'cluster',
        'lookup.cache.max-rows' = '10000'
      );
      SELECT item_id, price AS item_price FROM item_catalog_redis