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 特征时效性分级不是所有特征都需要实时更新特征类型更新频率存储介质示例实时行为 1 分钟Redis最近 10 次点击的类目分布短期画像10-30 分钟Redis最近 1 小时的兴趣向量长期画像每天HBase/Cassandra用户月度兴趣标签静态属性不变MySQL注册日期、地区将特征按时效性分级后计算资源集中在需要实时更新的少数特征上。实际项目中需要实时更新的特征不超过总数的 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 实验验证每个特征实时化的增量价值。一致性的代价实时特征和离线特征在时序上存在不一致窗口离线特征还在跑 T1 更新。在推理阶段处理这种不一致——实时特征值在窗口内权重线性衰减离线特征值在窗口内保持不变——比追求两路强一致更经济。灾难恢复Flink 作业 crash 后重启时依赖 Kafka 的消息回放。Kafka 的 retention 周期通常 7 天决定了可回放的时间窗口。重要特征建议双写——Kafka 外层再加一个 S3 归档——作为极端故障的兜底。五、总结推荐系统从离线到实时架构的演进是一次选择性加速而非全面替换。先分级特征时效性只对能带来显著效果增益的少数特征做实时计算再构建稳定的实时计算管道Kafka Flink Redis最后优化在线推理的批处理能力。开始落地时以用户最近 30 分钟的行为特征为起点——这是投入产出比最高的实时特征。验证这个单一特征的增量效果后再按需扩展实时特征的范围。每个新增的实时特征都必须承受住一个问题的审视离线版本的同样特征能做到什么程度差距是否足以支撑实时化的基础设施成本