腾讯看点视频推荐索引构建方案-InfoQ

一、背景

在视频推荐场景中，一方面我们需要让新启用的视频尽可能快的触达用户，这一点对于新闻类的内容尤为关键；另一方面我们需要快速识别新物品的好坏，通过分发的流量，以及对应的后验数据，来判断新物品是否值得继续分发流量。

而这两点对于索引先验数据和后验数据的延迟都有很高的要求。下文将为大家介绍看点视频推荐的索引构建方案，希望和大家一同交流。文章作者：纪文忠，腾讯 QQ 端推荐研发工程师。

注：这里我们把视频创建时就带有的数据称为先验数据，如 tag，作者账号 id 等，而把用户行为反馈的数据称为后验数据，如曝光、点击、播放等。

二、看点视频推荐整体架构

从数据链路来看此架构图，从下往上来看，首先视频内容由内容中心通过消息队列给到我们，经过一定的处理入库、建索引、生成正排/倒排数据，这时候在存储层可召回的内容约有 1 千万条。

然后经过召回层，通过用户画像、点击历史等特征召回出数千条视频，给到粗排层；粗排将这数千条视频打分，取数百条给到精排层；精排再一次打分，给到重排；重排根据一定规则和策略进行打散和干预，最终取 10+条给到用户；

视频在用户侧曝光后，从上之下，是另一条数据链路：用户对视频的行为，如曝光、点击、播放、点赞、评论等经过上报至日志服务，然后通过实时/离线处理产生特征回到存储层，由此形成一个循环。

基于此架构，我们需要设计一套召回/倒排索引，能够以实时/近实时的延迟来处理所有数据。

三、方案设计

在旧方案中，索引是每半小时定时构建的，无法满足近实时的要求。在分析这个索引构建的方案时，我们遇到的主要挑战有：

数据虽不要求强一致性，但需要保证最终一致性；
后验数据写入量极大，看点用户行为每日达到百亿+；
召回系统要求高并发、低延迟、高可用。

1. 业界主流方案调研

通过对比业界主流方案，我们可以看到，基于 Redis 的方案灵活性较差，直接使用比较困难，需要进行较多定制化开发，可以首先排除。

因此我们可选择的方案主要在自研或者选择开源成熟方案。经过研究，我们发现如果自研索引开发成本较高，而简单的自研方案可能无法满足业务需求，完善的自研索引方案所需要的开发成本往往较高，往往需要多人的团队来开发维护，最终我们选择了基于 ES 的索引服务。

至于为什么选择基于 ES，而不是选择基于 Solr，主要是因为 ES 有更成熟的社区，以及有腾讯云 PaaS 服务支持，使用起来更加灵活方便。

2. 数据链路图

（1）方案介绍

如下图所示：

这个方案从数据链路上分为两大块。

第一块，先验数据链路，就是上半部分，我们的数据源主要来自内容中心，通过解析服务写入到 CDB 中。其中这个链路又分为全量链路和增量链路。

全量链路主要是在重建索引时才需要的，触发次数少但也重要。它从 DB 这里 dump 数据，写入 kafka，然后通过写入服务写入 ES。

增量链路是确保其实时性的链路，通过监听 binlog，发送消息至 kafka，写入服务消费 kafka 然后写入 ES。

第二块，是后验数据链路。看点的用户行为流水每天有上百亿，这个量级直接打入 ES 是绝对扛不住的。所以我们需要对此进行聚合计算。

这里使用 Flink 做了 1 分钟滚动窗口的聚合，然后把结果输出到写模块，得到 1 分钟增量的后验数据。在这里，Redis 存储近 7 天的后验数据，写模块消费到增量数据后，需要读出当天的数据，并于增量数据累加后写回 Redis，并发送对应的 rowkey 和后验数据消息给到 Kafka，再经由 ES 写入服务消费、写入 ES 索引。

（2）一致性问题分析

这个数据链路存在 3 个一致性问题需要小心处理：

第一，Redis 写模块这里，需要先读出数据，累加之后再写入。先读后写，需要保证原子性，而这里可能存在同时有其他线程在同一时间写入，造成数据不一致。

解决方案 1 是通过 redis 加锁来完成；解决方案 2 如下图所示，在 kafka 队列中，使用 rowkey 作为分区 key，确保同一 rowkey 分配至同一分区，而同一只能由同一消费者消费，也就是同一 rowkey 由一个进程处理，再接着以 rowkey 作为分线程 key，使用 hash 算法分线程，这样同一 rowkey 就在同一线程内处理，因此解决了此处的一致性问题。另外，通过这种方案，同一流内的一致性问题都可以解决。

第二，还是 Redis 写模块这里，我们知道 Redis 写入是需要先消费 kafka 的消息的，那么这里就要求 kafka 消息 commit 和 redis 写入需要在一个事务内完成，或者说需要保证原子性。

如果这里先 commit 再进行 redis 写入，那么如果系统在 commit 完且写入 redis 前宕机了，那么这条消息将丢失掉；如果先写入，在 commit，那么这里就可能会重复消费。

如何解决这个一致性问题呢？我们通过先写入 redis，并且写入的信息里带上时间戳作为版本号，然后再 commit 消息；写入前会比较消息版本号和 redis 的版本号，若小于，则直接丢弃；这样这个问题也解决了。

第三，我们观察到写入 ES 有 3 个独立的进程写入，不同流写入同一个索引也会引入一致性问题。这里我们可以分析出，主要是先验数据的写入可能会存在一致性问题，因为后验数据写入的是不同字段，而且只有 update 操作，不会删除或者插入。

举一个例子，上游的 MySQL 这里删除一条数据，全量链路和增量链路同时执行，而刚好全量 Dump 时刚好取到这条数据，随后 binlog 写入 delete 记录，那么 ES 写入模块分别会消费到插入和写入两条消息，而他自己无法区分先后顺序，最终可能导致先删除后插入，而 DB 里这条消息是已删除的，这就造成了不一致。

那么这里如何解决该问题呢？其实分析到问题之后就比较好办，常用的办法就是利用 Kfaka 的回溯能力：在 Dump 全量数据前记录下当前时间戳 t1，Dump 完成之后，将增量链路回溯至 t1 即可。而这段可能不一致的时间窗口，也就是 1 分钟左右，业务上是完全可以忍受的。

线上 0 停机高可用的在线索引升级流程如下图所示：