腾讯内容千亿级实时计算和规则引擎实践优化之路_环球最资讯

来源：InfoQ 　2022-11-15 05:36:48

作者 | 王冬、杨浩、李文斌、王玢、李会珠

审校 | 蔡芳芳

(资料图片)

1.系统背景

腾讯内容中台提供从内容生产、内容加工、内容分发、内容结算等全链路环节的一站式服务，在这个过程中，会产生大量的数据以及围绕这些数据衍生的实时流业务应用，如智能审核、运营决策、在线学习等，从底层去看这些内容生态场景的本质，它需要我们提供一套完善的基于规则引擎的实时流信号服务来控制信号和业务流转，且实时信号场景具有内容数据源复杂、吞吐量高、计算量大、准确度高等特点。因此，我们利用业界前沿的实时流技术，并结合一些核心内容生态实时流场景的自研关键技术，沉淀了一套数据复用度高、可用性强、需求响应快的实时流服务，高效赋能腾讯内容生态产品。

2. 问题与挑战

问题 1：多实时数据源动态感知、内容 OneID 数据

腾讯内部各个业务方生产的数据各异，且拥有各自的 ID 体系；随着业务发展，数据源还会动态添加消息 Topic，需要实时动态感知新增的数据源，并以中台统一的 ID 视角串联各个业务的内容数据。

问题 2：TB 级多流数据拼接、批数据重建流状态

内容加工时会产生较多的复杂计算需求，比如，我们需要在有限资源内保障 TB 级多条实时数据拼接工作，以及长时间运行下需要对实时流应用的计算口径进行调整而面临的批数据重建流式数据状态等问题，我们探索了一系列自研技术，解决了海量数据实时流计算问题。

问题 3：规则引擎日千亿次实时信号触发

内容生态系统很多场景依赖实时信号，并且基于规则进行控制和流转，烟囱式开发有较大成本，我们需要构建一套日千亿次匹配的规则引擎信号服务，保障资源共享，实现新增场景一键配置即可支持。

问题 4：全链路全生命周期信号服务质量保障

内容场景中实时信号通常用于审核、运营等核心在线场景，稳定性要求极高。而数据往往面对依赖组件多、链路长、吞吐高等问题，会引起反压、偶发崩溃等多种质量问题，需要我们建设一套通用的质量保障体系，包括可观测性、状态高可用等。

3. 整体架构

图 3-1 内容生态实时信号系统架构图

数据接入 ：构建准确统一的基础数据，通过动态新增数据 Topic 自适应感知、十万级 QPS 的 ID 映射等手段，解决数据源消息 Topic 动态拓展无法自动感知、数据孤岛等接入问题。

信号生产 ：提供滑动大窗口计算、多流 TB 级数据拼接、融合批数据重建流状态、单体流量适应水平扩展等通用解决方案，保障大吞吐下的信号生产的时效性、稳定性。

规则引擎 ：结合业务个性化触发逻辑，提供统一的规则引擎触发系统，支持日千亿次的实时规则匹配、信号高效去重分发，保障多样场景一键快速支持。

信号工厂 ：一些信号特征无需经过规则引擎流转，按照主题管理，直接透传给业务应用。

服务质量 ：我们构建了全链路全生命周期的服务质量保障体系。包括全链路可观测性系统，Flink 核心状态高可用设计、全生命周期质量监控和解决流程、元数据管理等。

3.1 数据接入

3.1.1 动态实时源自适应感知

腾讯内容中台，提供一站式工业化的内容加工能力，每个业务方可自定义编排加工内容的任务流拓扑。为了稳定性和隔离性，每条任务流拓扑内容加工操作流水会生成一个 Topic，随着业务发展，新的 Topic 会不断增加，同时存量 Topic 数据量可能变大。因新增 Topic 所属集群地址差异大，Flink Source 无法用正则匹配到，导致程序无法自动感知。因此，我们设计了 Topic 动态添加的自适应感知的技术方案，可以做到：

数据完整性：自动感知新添加的拓扑 Topic，保证数据不遗漏。数据时效性：存量的 Topic 数据量级变大时，能够自动扩容，保障整体时效性。

图 3-2 动态实时源自适应感知示意图

主要由以下几个模块构成：

控制器模块：监测消息队列并通过配置中心异步控制 Flink 的消费。新增 Topic 时，注册到配置中心。 Topic 数据量变大导致消费延迟时，增加该 Topic 的消费并行度。配置中心：存放所有拓扑的消息队列，如拓扑 ID、消费并行度、Kafka 配置。 Flink 自适应 Source：自适应消费 Kafka 数据，保障数据完整性和时效性。在 Task 内开启消费线程池，负责 Kafka 的消费；并有自适应 Client，负责控制线程池的消费，每分钟执行一次，保障消费的完整性和时效性。步骤 1：拉取所有消息队列配置。步骤 2：生成本 Task 消费的 Topic 消费列表，保障并行度 N 的 Topic 会被 N 个 Task 消费。总 Task 数目是 M，每个 Task 会被分到如下 Task 中：hash(pipeline_id) % M 到 (hash(pipeline_id) + N) % M。遍历 Topic 可能被消费的 Task 列表，如果其中包含本 Task，则可对其进行消费。步骤 3：调整线程池消费列表，如果步骤 2 中添加了 Topic，则添加对应 Topic 的消费。

3.1.2 十万级 QPS 高并发 ID 映射

因每个业务渠道（如腾讯新闻、QQ 浏览器等）有自己的内容 ID 体系，为此，在整合各渠道的消费流水时，我们需要将业务 ID 映射成腾讯内容中台统一的内容 One ID 体系。如果直接请求现有的 ID 映射服务，大量的网络 IO 会消耗较大的实时流计算资源。

为此，我们构建了基于二级缓存的 ID 映射解决方案，大幅降低对远程服务的访问，可节约上百倍的计算资源。

图 3-3 基于二级缓存的实时 ID 映射

如上图所示，具体步骤如下：

获取中台 ID：首先判断应用内状态中是否有该 ID 的缓存，如果有则直接返回中台 ID；如果没有，则访问 ID 映射服务，并将其更新到 State 中。一级缓存：在程序中构建 Flink 应用内状态（Flink State），缓存渠道 ID 到中台 ID 的映射。因为远程拉取中台 ID 时有缺失，缺失时无法判断是当时映射服务有遗漏但是后续请求能映射上，还是该渠道 ID 本身无法映射到中台 ID，为保障数据准确性，我们构建了 2 种 State 控制 ID 映射：可以映射的 State：存放渠道 ID 到中台 ID 的映射，为规避状态膨胀，TTL 设置成 7 天，过期时间从最近一次访问时间开始计算。不能映射的 State：存放未映射上的渠道 ID。为保障整体数据可用性，需要定期强制重新拉取中台 ID，将 TTL 设置成 1 小时，过期时间从第一次访问时间开始计算。二级缓存：远程 ID 映射服务，通过 Rest Api 访问。拼接 ID：在消费流水中，拼接上中台 ID。

3.2 信号生产

在实际应用场景中，需要提供多样的实时特征信号，信号生产过程中，我们遇到了多种挑战，本章将结合实际问题，介绍我们通用自研的解决方案。

3.2.1 千亿次滑动大窗口计算

在内容场景中，需要对内容消费数据的大时间窗口 (如 1 天、30 天等) 的每分钟滑动指标进行日千亿次的实时流计算，并基于这样的数据指标来控制业务流转，如果我们直接基于 Flink 内部的窗口函数，进行实时计算窗口指标时，因不能及时关闭窗口，状态数据会占用大量的内存，导致计算出现反压的情况，程序稳定性差，容易出现卡死现象。

基于上述挑战，我们设计了一种高性能的大窗口计算方法，主要有如下优点：

传统的方式需要每次对大窗口中的全量数据做计算，而现有方式可以复用前一次计算结果，可极大减少计算量。我们方案中大窗口是逻辑上的大窗口，相比 Flink 原生的窗口计算会保留大窗口时间内的原始数据，我们在内存中并不存放这些原始数据，只存放算法提到的聚合维度的数据，同时利用数据淘汰机制，防止内存占用过大，节约大量的内存资源。

我们针对大窗口（如 1 天）、超大窗口（如 30 天等），结合计算复杂度和精度要求，采用了不同的计算方案，保障小成本高精准计算多种窗口指标。

大窗口计算

对实时流数据根据数据自身的事件时间是否连续分为如下不同的几种情况：

情况一：分钟级别滑动，每分钟窗口连续有流量的情况

当数据自身的事件时间连续的时候，我们需要拿到上次大窗口的计算结果值，在上次计算结果的基础上，对窗口的头部和尾部进行加减操作就可以得到新的大窗口的值。

图 3-4 分钟级滑动每分钟连续的大窗口

其中，T(6, 4) 代表的是 6min 时候近 4min 的累计值大小，其中 6 代表的是当前最新时间，4 代表的是需要统计的窗口大小，是人为规定的。M(5) 代表的是第 5min 的值。

情况二：分钟级别滑动，每分钟窗口流量不连续情况

当间隔的时间小于窗口大小的情况下，计算当前窗口的值需要得到上一个窗口的值进行加减操作，由于数据自身的事件时间中断，所以要对最后一次窗口的值进行校准。

图 3-5 分钟级滑动每分钟不连续大窗口

其中，T(5, 4) 代表的是 5min 时候近 4min 的累计值大小，其中 5 代表的是当前最新时间，4 代表的是需要统计的窗口大小，是人为规定的，M(5) 代表的是第 5min 的值。

情况三：分钟级别滑动，每分钟窗口流量不连续并且当间隔的时间大于窗口的情况

当间隔的时间大于窗口大小的情况下，由于窗口时间内没有出现流量，可以直接认为大窗口的计算值为当前分钟流量值。

图 3-6 分钟级滑动每分钟不连续大窗口

其中，T(6, 4) 代表的是 6min 时候近 4min 的累计值大小，其中 6 代表的是当前最新时间，4 代表的是需要统计的窗口大小，是人为规定的，M(5) 代表的是第 5min 的值。

超大窗口（如 30 天）

针对 30 天等超大滑动窗口计算，资源开销会成数十倍的膨胀，成本难以承受。我们构建了一套解决方案，成本降低到千分之一，精度只损失了百分之一，在成本和精度间达到了高效平衡。

图 3-7 超大滑动窗口指标计算

如上图所示，计算单个内容 ID 的超大滑动窗口指标过程如下。

状态更新：读取消费流水，更新该 ID 的状态值。计算超大窗口指标：基于应用内状态进行计算。如果内容产生时间在 N 天内：取累计流量。如果内容产生时间在 N 天前：基于输入流量的时间取不同范围的数据，整体半天精度，如 30 天超大窗口的误差约 1.6%。 00:00—12:00：取过去 N 天 + 当天流量值。 12:00—23:59：取过去 N-1 天 + 当天流量值。

3.2.2 延迟流数据滚动大窗口计算

在内容生态场景中，由于历史原因和服务器时钟问题导致会出现超自然时间的数据，以及网络原因造成的延迟的数据。传统通过设置窗口水印的方式存在一定问题，对于超自然数据，会导致窗口立刻关闭；对于延迟数据，窗口关闭后，延迟到来的数据未能被统计到窗口指标中。

为了解决上述问题，我们设计了一种可以同时处理超自然数据和延迟数据的方案，优点如下：

对于大窗口的计算有绝对的优势，普通的方式大窗口计算时候由于窗口太大，窗口不能及时关闭，当内存中存在大量的窗口时性能会急速下降。此技术通过聚合 Key 设计，极大的提高了大窗口情况下计算的稳定性、时效性、准确性。提供了及时的内存清理机制，保证聚合 Key 在过期时候能够被及时的清理，保证程序不会随着时间的推移而出现性能的损耗。

图 3-8 延迟流数据滚动大窗口计算

我们窗口计算转换为 Key 的分类聚合问题，通过对要参与聚合计算的 Key 进行巧妙设计，进而实现聚合统计。

步骤 1 ：计算数据所属的窗口起始值，窗口起始时间值 = 事件时间 / 窗口大小 * 窗口大小，窗口大小是根据业务需求来指定的。对于超自然数据，需要基于业务场景进行时间矫正。

步骤 2 ：根据窗口的起始值对数据进行分配，正常数据直接放入正确的窗口中，延迟数据由于只是晚到，但是数据的生成时间是正确的，所以可以根据窗口标记找到对应的窗口，放入对应的窗口中。

步骤 3 ：对窗口中的数据生成独有的聚合 Key，聚合 Key= 计算 Key+ 日期 + 窗口起始时间值。

步骤 4 ：按照聚合 Key 的值进行 Shuffle 分组，聚合 Key 相同的数据会被发送到同一个计算任务，进行聚合或者更加复杂的计算，并且清理内存中过期的聚合 Key，避免程序随着时间推移出现性能下降问题。

3.2.3 TB 级实时流数据拼接

Flink 原生实现进行 TB 级数据拼接时，计算较慢，且状态备份时可能异常导致难以升级 APP。

因此，我们构建了可以解决大状态下多流拼接的时效性和稳定性问题的技术方案，并保证最终一致性。

图 3-9 基于 HBase 实现 TB 级实时多流拼接

主要思路如上图所示，我们借助第三方 HBase 存储完成多流关联。

阶段 1 ：特征拼接，每个源单独加工，抽取自身特征后，进行如下过程：

步骤 1：将自身特征同步到 HBase 中，每个源只更新自身属性对应的列。HBase 中会包含每个内容最新最全的属性。步骤 2：将有变更的内容推送到消息队列中。当前实现是将所有有变更的内容实时推送下游，可改造该过程，多流水位对齐后再推送，以支持多流拼接的多种语义。

在本阶段的存储设计中，HBase 的 Rowkey 为待关联的 Key，列分别为属性 Key 和属性值。同时，我们进行了大量优化设计：

批量访问：每 50 个 Key 合并访问，减少 IO。随机主键：将 Key 进行 md5 哈希，让数据均匀分布在 HBase 中，防止热点，提高随机访问性能。存储压缩：部分属性值较大，将其序列化后，使用 GZIP 压缩，减少存储。过期机制：按需设置 TTL，防止数据无限膨胀。

阶段 2 ：特征输出，通过一个程序统一加工处理，可将每个内容的全量特征输出到目标业务系统中。

步骤 3：实时感知特征有变更的内容。步骤 4：批量拉取内容的全量特征，HBase 中每一列对应一个特征，某个内容的全部列即为其全部特征。步骤 5：入库，将从 HBase 中获取的全量特征，转换成目标存储格式，输出到目标系统。

3.2.4 融合批数据重建流状态

在内容生态的实时计算场景中，我们经常会遇到累计指标的统计，比如某一条内容的实时总点击数、展现数等。传统的方式主要是用 Lambda 架构进行加工，面对口径发生变化等情形时，会有如下问题：

批处理计算和实时流计算两份代码可能由多人维护开发，因此容易造成计算结果不一致。批处理计算和实时流计算切换的时候出现数据抖动，影响用户体验。

因此，我们设计了批流状态融合架构，主要优点如下：

只需要维护一份实时流计算代码，通用性较好，适合所有实时流需要计算业务历史数据的场景。解决了实时流计算批量回溯历史数据时的算力问题，利用存量批处理计算资源回溯历史全量数据，同时，结合仅需从 T 日零点零分零秒开始的实时流数据，得到口径变化后的完整指标数据。规避了数据的抖动，提供好了良好的用户体验。

图 3-10 批量融合状态重建架构

首先计算业务历史全量累计数据存入 Key-Value 缓存中作为基准数据，把实时数据和基准数据进行融合计算得到最新累计值，并可根据下游系统的负载能力调整数据的输出间隔。

步骤 1 ：初始化时或者业务口径变更后，通过离线批处理计算历史全量数据，作为每个 Key 的基准数据，导入到 Key-Value 存储系统。

步骤 2 ：重启实时流计算应用程序后，每个 Key 根据是否初始化过基准数据，从 Key-Value 中初始化基准数据。

步骤 3 ：将基准数据和实时数据进行合并计算，通过流量控制把数据写入到下游业务存储系统中，供业务查询使用。

3.2.5 单体流量适应水平扩展

内容生态面临着内容的消费数据越来越大的情况，单个实时流计算程序在 Flink 状态不断增大的情况下，由于单个程序需要维护的状态越来越大，程序频繁出现反压问题，增加程序的并发度也提高不了稳定性。

通常我们会增加实时流应用来适应流量水平扩容的架构，但是增加应用后，如果把数据随机发往扩容后的程序，会有一些潜在的问题，例如在计算某个内容 ID 累计值的场景，需要这个内容 ID 对应的所有数据严格发送到同一个程序，才能保证最终结果的准确性。

图 3-11 单体流量适应水平扩容

为了解决以上问题，我们设计了如下可以适应流量水平扩展的架构。步骤如下：

步骤 1 ：记录数据首次进入系统的时间，为了防止数据丢失做高可用的持久化存储。

步骤 2 ：维护系统扩容前后的 buckets 的值，当数据过来之后根据数据首次进入系统时间所处的时间段找到对应的 buckets 的值。

步骤 3 ：对内容进行寻址，将内容 ID 哈希后分配到 buckets 个桶中，而下游每个 App 对应一个桶。

3.2.6 输出小文件数自适应流量

在内容加工场景中，需要将消息队列数据同步到 HDFS 中。同步时，会有 N 个同步子任务，其中 N 由流量峰值决定，N 在同步过程中不能调整，当数据时效性为分钟时，每分钟会有 N 个子文件。然而，在流量低峰期时，由于 N 不会改变，会产生大量的小文件。

图 3-12 输出文件数自适应流量

如上图所示，我们构建了一种输出小文件数自适应流量减少的解决方案。取单个文件为目标大小 S（如 64MB），以控制文件数目。我们将整个过程由原来的 1 个阶段拆分成了 2 个阶段：Map 阶段和 Reduce 阶段，其中 Map 任务数是 M，Reduce 任务数是 N。以下两个阶段，每分钟调度一次：

Map 阶段 ：读取数据进行自适应映射。缓存数据：每个任务缓存 1min 的数据。计算本批次产生的目标文件数 K：缓存的数据大小乘以 M 得到本批次所有数据输出大小 total_size，计算当前批次目标文件数 K=total_size/S。均匀映射：每条数据依次加上 1 到 K 的 Key，data 转换成 (k, data)，以方便 Shuffle 控制。 Reduce 阶段 ：Reduce 子任务 k 只拉取 Key 为 k 的数据，这样，子任务 1 到 K 之间会有数据，剩下的任务无数据。因为空任务不会产生文件，这样可以保障本批次输出的文件数为 K。

3.3 规则引擎

在内容生态中除了实时流信号的生产服务，往往我们还需要进一步基于实时流信号，结合规则引擎管理业务个性化的触发逻辑，以此来支持内容周期智能管理等多种应用场景。

图 3-13 基于规则引擎的实时信号触发

3.3.1 规则管理平台

规则类型

基于不同的业务需求场景，规则定义区分了固定规则和动态规则：

固定规则：同一规则下所有内容阈值相同。动态规则：同一规则下不同内容阈值可以精细化设置，用于满足基于内容特征属性需要不同的信号触发阈值的需求场景。

规则管理

提供规则以及内容阈值的增加、更新、查询等能力，并支持如下数据管理能力：

规则增删改查：用户可以通过管理端查询规则列表，录入和修改规则。动态阈值增删改查：提供 Rest Api 对规则下内容的阈值进行新增、更新和查询。该能力可支持预估模块训练阈值后，将相应阈值更新到规则配置中；同时供规则执行引擎查询规则配置。

规则定义

配置模块旨在对规则进行进行抽象，通过定义通用的规则抽象定义，把用户在管理配置信息进行接入存储。解耦用户规则定义和规则引擎，降低用户输入和规则引擎的依赖，这样可以便于我们无负担去升级替换规则引擎而对用户无感。

图 3-14 规则信息

规则描述包括两部分，规则条件表达式 + 规则动作：

表达式条件：上层逻辑支持且 / 或，支持多个运算算子；表达式动作：支持设置触发优先级以及携带特定信息等

3.3.2 规则执行引擎

基于上面信号产生的实时信号和规则管理提供的规则信息，我们探索了开源的 Aviator、Flink CEP 等组件。Flink CEP 构建规则执行引擎时有如下问题：

不支持规则信息的动态更新，用户使用体验较差。不支持多规则，导致难以平台化。

Aviator 支持丰富的运算符和表达式，同时具有轻量级和高性能特点，能够完全覆盖我们的场景。为此，我们选取了 Aviator 作为规则匹配引擎。

规则执行引擎主要有如下三个模块：

规则加载

负责进行执行器所需配置的加载和实时感知，主要提供如下能力：

规则变更秒级感知 ：负责加载规则列表，并保障规则变更后能及时同步，包括规则信息变更、规则添加、规则删除等； 动态规则的阈值同步 ：动态规则中，每个内容有自己的阈值。在进行动态规则匹配时，可以实时获取（内容 ID、规则 ID）对应的阈值。

规则路由

从输入信号中提取业务渠道，和规则中业务渠道进行关联匹配，依次路由到不同的规则匹配算子中。

规则匹配

为不同规则提供相应的匹配能力：

固定规则：将获取到的信号和每个规则进行轮询匹配。匹配时将规则中表达式和信号作为作为输入，通过 Aviator 进行匹配，如果信号满足规则，则将信号和规则关联，输出到下游。动态规则：整体流程和固定规则一致，但因为规则中阈值是动态的，需要进行设计以适配 Aviator：动态阈值作为实体元素放入表达式中（如 vv_1_day>=dynamic_value）中，将拉取的（内容 ID、规则 ID）的阈值拼接入输入信号中，字段为 dynamic_value，即可以实现动态规则的匹配。

3.3.3 规则匹配优化

并发匹配 ：由于单个任务计算能力有限，把数据分为若干份，在多个任务中进行规则匹配，极大的提高了规则引擎在大数据量场景下的匹配能力。

二级缓存 ：动态规则匹配时，需获取（内容 ID、规则 ID）的阈值，因输入信号峰值 QPS 数十万，拉取阈值会有较大网络 IO，造成极大资源开销。参考前文 ID 映射的解决方案，我们构建了（内容 ID、规则 ID）->阈值的二级缓存，可以极大节省匹配资源。

预编译技术 ：进行规则匹配时，首先将规则编译成机器能理解的字节码，然后将上游信号作为数据输入进行匹配运算。该过程主要耗时在将规则编译成字节码阶段，我们将规则对应的字节码进行缓存，可以节省上千倍的算力开销。

3.3.4 信号去重分发

信号去重

经过规则执行引擎后，仍然能召回大量信号，针对审核等场景，一个内容触发后，短时间内不需要再次输出，以免重复审核。为此，我们进行了个性化去重模块，支持灵活的去重周期，如永久去重、天级去重等，为不同的业务场景召回所需的信号。

信号分发

下游有多个业务系统，基于规则和业务场景的关系，将信号分发给对应的业务模块。分发方式支持消息队列投递以及接口回调，业务可以根据需要进行定制。

3.4 服务质量

3.4.1 端到端全链路服务可观测性

实时流服务对接的数据源多、加工链路长，会导致如下问题：

问题发现慢：因为很难衡量端到端时效性，导致很难感知整体延迟，往往业务反馈后才知道。定位时间长：需一步一步联系上游环节，以确认数据延迟源头。

我们构建了全链路端到端端的可观测性系统，可以监控端到端延迟并快速定位问题环节，主要分位以下 4 个模块：

数据染色：如图 3-15 所示，本模块集成到各个加工程序中，将本环节和上游各个环节的时效性信息染色到输出数据中，如事件时间、输出时间等。

图 3-15 数据染色示意图

时效性统计 ：因为每个环节的输出包含自身以及上游各个环节的时间信息，可基于某个环节的输出数据，统计从数据源到当前环节端到端分环节、分数据源的时效性信息。 延迟监控 ：基于统计模块计算的数据，监控端到端的延迟。 可观测分析工具 ：基于统计产生的数据，可以构建全链路实时拥堵分析工具，快速定位问题源头。

图 3-16 全链路实时拥堵分析

3.4.2 旁路系统保障状态高可用

内容生态中，计算内容的累计值、首次时间等场景，强依赖于 Flink 自身状态，但是因为依赖组件异常等原因，导致 Flink 有概率丢失状态，无法满足对数据一致性要求非常高的场景的需求。

我们构建了旁路系统，保障 Flink 状态异常丢失后，作业重启后核心状态的高可用。架构如图所示，主要由 2 个模块构成

图 3-17 旁路系统保障状态高可用

旁路系统：程序外起一个异步作业，将核心状态从输出中实时同步到 Redis 中。 Flink 应用内状态恢复模块：为访问 State 的前置逻辑，如果 Key 在应用内状态中丢失，则从远程 Redis 中恢复。