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

审校 | 蔡芳芳

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%。

—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 构建规则执行引擎时有如下问题：

不支持规则信息的动态更新，用户使用体验较差。

不支持多规则，导致难以平台化。