AI 推荐系统的架构进阶：从离线批处理到实时特征的工程演进

AI 推荐系统的架构进阶：从离线批处理到实时特征的工程演进

一、离线时代的天花板：用户行为滞后 24 小时

一个内容平台的推荐系统在 2024 年经历了一次架构升级。旧的架构是经典的离线批处理模式：每天凌晨 2 点，Spark 任务计算全量用户的推荐向量，写入 Redis，白天 API 直接读取。问题是——用户上午看了一篇 Golang 并发文章，到当天下午推荐流里仍然是 Python 基础教程。行为反馈滞后了 24 小时。

这个延迟在用户活跃度低时影响不大。当 DAU 突破 50 万后，行为稀疏性问题被放大——用户的需求在实时变化，但推荐系统活在昨天。从离线到实时的演进不是把 Spark 换成 Flink 那么简单，涉及特征工程、模型服务、在线存储三个层面的重构。

二、实时特征管道的架构设计

flowchart LR
subgraph 实时链路
A[用户行为事件] –> B[Kafka 消息队列]
B –> C[Flink 实时特征计算]
C –> D[(Redis 在线特征存储)]
D –> E[在线推理服务]
end

subgraph 离线链路
A –> F[(数据湖/HDFS)]
F –> G[Spark 离线特征宽表]
G –> H[(离线模型训练)]
H –> |模型更新| E
end

I[用户请求] –> E
E –> J[推荐结果]

style A fill:#f9f,stroke:#333
style E fill:#9cf,stroke:#333

实时链路处理用户最近的 15-30 分钟行为，离线链路处理全量历史数据。在线推理服务合并两路特征，做最终的排序。这个架构的关键不是"用上 Kafka + Flink"，而是定义哪些特征需要实时计算、哪些可以等待离线窗口。

2.1 特征时效性分级

不是所有特征都需要实时更新：

将特征按时效性分级后，计算资源集中在需要实时更新的少数特征上。实际项目中，需要实时更新的特征不超过总数的 10%。

2.2 实时特征的计算实现

// 实时特征更新器：消费 Kafka 行为事件，更新 Redis 特征
type RealtimeFeatureUpdater struct {
consumer *kafka.Reader
redis *redis.Client
batchSize int
window time.Duration // 特征聚合窗口
}

func (u *RealtimeFeatureUpdater) Run(ctx context.Context) error {
buffer := make([]UserEvent, 0, u.batchSize)
ticker := time.NewTicker(u.window)
defer ticker.Stop()

for {
select {
case <-ctx.Done():
return u.flush(buffer)

case <-ticker.C:
if len(buffer) > 0 {
if err := u.flush(buffer); err != nil {
return fmt.Errorf("刷新缓冲区失败: %w", err)
}
buffer = buffer[:0]
}

default:
msg, err := u.consumer.FetchMessage(ctx)
if err != nil {
if errors.Is(err, context.Canceled) {
return nil
}
continue // 单条消费失败不中断整体流程
}

var event UserEvent
if err := json.Unmarshal(msg.Value, &event); err != nil {
u.consumer.CommitMessages(ctx, msg) // 跳过坏消息
continue
}
buffer = append(buffer, event)

if len(buffer) >= u.batchSize {
if err := u.flush(buffer); err != nil {
return fmt.Errorf("批量刷新失败: %w", err)
}
buffer = buffer[:0]
}
}
}
}

func (u *RealtimeFeatureUpdater) flush(events []UserEvent) error {
pipe := u.redis.Pipeline()

for _, event := range events {
key := fmt.Sprintf("feature:user:%d:recent", event.UserID)
// 使用 Sorted Set 维护时间窗口内的行为
pipe.ZAdd(context.Background(), key, &redis.Z{
Score: float64(event.Timestamp),
Member: event.CategoryID,
})
// 自动清理过期行为（保留最近 1 小时）
cutoff := time.Now().Add(-1 * time.Hour).Unix()
pipe.ZRemRangeByScore(context.Background(), key, "0", fmt.Sprint(cutoff))
}

_, err := pipe.Exec(context.Background())
return err
}

缓冲 + 批量写入的设计将 Redis 的写 QPS 从每事件一次降到了每窗口一次。窗口大小在 100ms-500ms 之间做权衡：越小特征越实时，越大 Redis 写入效率越高。建议以端到端延迟的 5% 作为窗口大小上限——如果 tolerated 端到端延迟是 5 秒，窗口不超过 250ms。

三、在线推理的性能保障

从离线到实时的最大隐性成本变化在推理层。离线批处理可以预计算全量推荐向量，在线推理需要实时做特征拼接和模型计算。

关键优化策略：

特征服务的专用缓存：不要复用业务的 Redis 集群。推荐系统的特征读写 QPS 远高于其他业务场景，混用会导致热 Key 竞争。专用集群的 maxmemory-policy 设置为 allkeys-lru，让 Redis 自行淘汰低频特征。

模型推理的批量合并：当 QPS > 1000 时，将 100ms 内的请求合并为一个 batch 调用模型推理，利用 GPU 的并行能力。

// 推理请求批处理器
type InferenceBatcher struct {
queue chan InferenceRequest
batchSize int
maxWait time.Duration
}

func (b *InferenceBatcher) Run(ctx context.Context) {
batch := make([]InferenceRequest, 0, b.batchSize)
timer := time.NewTimer(b.maxWait)
defer timer.Stop()

for {
timer.Reset(b.maxWait)
select {
case req := <-b.queue:
batch = append(batch, req)
if len(batch) >= b.batchSize {
b.execute(batch)
batch = batch[:0]
}
case <-timer.C:
if len(batch) > 0 {
b.execute(batch)
batch = batch[:0]
}
case <-ctx.Done():
if len(batch) > 0 {
b.execute(batch)
}
return
}
}
}

四、边界权衡

实时不等于全量实时：将全部特征都切换到实时计算，基础设施成本增加 3-5 倍。只对"能显著提升推荐效果的少数特征"做实时化，用 A/B 实验验证每个特征实时化的增量价值。

一致性的代价：实时特征和离线特征在时序上存在不一致窗口（离线特征还在跑 T+1 更新）。在推理阶段处理这种不一致——实时特征值在窗口内权重线性衰减，离线特征值在窗口内保持不变——比追求两路强一致更经济。

灾难恢复：Flink 作业 crash 后重启时，依赖 Kafka 的消息回放。Kafka 的 retention 周期（通常 7 天）决定了可回放的时间窗口。重要特征建议双写——Kafka 外层再加一个 S3 归档——作为极端故障的兜底。

五、总结

推荐系统从离线到实时架构的演进是一次"选择性加速"而非"全面替换"。先分级特征时效性，只对能带来显著效果增益的少数特征做实时计算；再构建稳定的实时计算管道（Kafka + Flink + Redis）；最后优化在线推理的批处理能力。

开始落地时以"用户最近 30 分钟的行为特征"为起点——这是投入产出比最高的实时特征。验证这个单一特征的增量效果后，再按需扩展实时特征的范围。每个新增的实时特征都必须承受住一个问题的审视：离线版本的同样特征能做到什么程度？差距是否足以支撑实时化的基础设施成本？