企业官网建设流程全解析

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