Flink CEP 在抖音电商的实践

本文整理自抖音电商实时数仓研发工程师张健，在 Flink Forward Asia 实时风控专场的分享。本篇内容主要从 Flink CEP 简介、业务场景与挑战、解决方案实践和未来展望四个方面展开介绍。

Flink CEP 是基于 Flink Runtime 构建的复杂事件处理库，擅长处理跨多个事件的复杂规则匹配场景。在电商场景下，例如检测用户下单后，是否超过一定时间仍没有发生支付行为；检测用户进入直播间后，是否有浏览商品随后加入购物车行为等。

与其他技术选型相比，Flink CEP 有以下优势：

• 支持跨多事件的规则匹配计算；
• 具备精准一次计算语义、低延迟、高吞吐等特性。

随着抖音电商业务逐渐趋于稳定和成熟，抖音电商实时数仓团队接到的实时数据规则类业务需求也逐步增多，因此我们开始尝试使用 Flink CEP 支持这些业务场景。下面列举两个典型的业务场景，并介绍 Flink CEP 在这些场景中遇到的挑战。

业务背景

1. 实时预警场景。这是非常典型的业务诉求，把用户看数据的方式从大屏“盯盘”转换为“根据规则检测结果，主动推送”，这无疑对一些关键业务问题的发现和洞察起到至关重要的作用。有如下三个具体案例：直播实时检测场景。当检测到直播间在一段时间内观看人数持续下跌时，会实时把消息推送给直播达人，方便其及时做出直播策略的调整。比如调整讲解商品的话术，发放粉丝礼物等等，进而提升转化。实时风控的场景。当检测到用户有可能存在刷单行为时，我们会将这个用户实时推送给平台治理同学，并做出相应的封禁处置，促进平台的整体生态健康。售后咨询场景。当检测到一个用户发起咨询后，超过 x 分钟都未得到回复，会立即通知相关的客服人员及时回复，提升整体的用户体验。
2. 实施营销场景。这是基于实时数据驱动，根据定义的规则策略挖掘目标群体，并根据业务目标做出精准营销投放的营销活动。有如下三个具体案例：促进购买场景。针对一些价格比较高的商品，当检测到用户下单后没有支付，那么该用户可能因价格犹豫是否支付。这个时候可以及时进行一些刺激购买的运营动作，从而提升平台的转化率。帮助商家及时发现爆款商品场景。当检测到某款商品在一定时间内成交超过 x 单时，会实时将这个商品的名称、品牌、库存等信息推送给商家，以便商家及时补货、直播间挂链接等行为，提升运营效率。在线发奖励场景。当检测到达人在完成电商大学学习后并在一定时间内进行了电商开播，或发布了电商短视频等行为，就会对这个达人发放一些仪式奖励，提升整体达人的入驻率，进而给商家提升更加多元的达人选择。

业务挑战

第一，在规则配置方面存在灵活性不足的问题。当前无论是新增还是修改规则，都需要实时数仓的研发同学通过修改代码的方式来支持，这就导致研发同学需要频繁的对接业务。在一些极端的场景，如双十一大促期间，一个研发同学往往需要同时对接多个运营同学的规则创建或者修改的诉求。业务需求也由于人力的单点阻塞问题迟迟无法上线。

第二，规则与计算任务之间存在深度耦合。当每个规则都需要强制绑定一个计算任务时，就会导致计算任务的数量会随着规则的创建逐渐增多。大量的任务会造成极高的运维成本和巨大的资源浪费，使整个系统最终变得不可维护。以前面提到的商家自定义规则检测爆款商品的这个场景为例，考虑到当前抖音电商庞大的商家群体，最终创建规则的数量可能是巨大的，进而导致整个计算任务的数量也随之爆炸。

第三，当前社区版 Flink CEP 支持的规则语义不够丰富。列举两个典型的案例：

• 第一个案例，假设需要检测用户多次下单后，没有在某一时间内完成支付行为。这种场景的特点是用户最后一次下单后，一直没有支付事件来触发规则从而完成匹配。当前社区版 Flink CEP 不支持这种场景，但在真实的业务中这又是非常普遍的规则诉求。
• 第二个案例，假设需要检测用户在过去一段时间内，是否完成一些固定行为动作。并要求这些行为不分先后顺序。这种场景当前社区版 Flink CEP 也不支持。

建设思路

整体分为四个阶段解决上述的问题。

第一阶段，对 Flink CEP 规则的核心信息进行了提炼和抽象，并设计了一套清晰易懂的规则 DSL。这样就可以让业务同学自主配置业务规则，从而解决规则配置灵活性不足的问题。那么如何让业务配置的规则运行起来就成为下一步待解决的问题。

第二阶段，对 Flink CEP 计算任务进行改造，让其支持动态提交规则或者更新规则的能力，从而实现规则与计算任务之间的彻底解耦。解耦之后，不再强制要求每一个规则必须对应一个计算任务来运行。也就是同一个计算任务可以同时接收提交的多条规则，实现收敛整体计算任务的数量，提升规则利用率的目标。

前面两个阶段解决了规则配置的灵活性以及规则与其他任务的强绑定问题，但是仍然没有解决规则本身的语义丰富性问题。因此，第三阶段，主要针对特定业务的场景的规则诉求、升级和拓展规则的语义。

经过前三阶段的升级和优化，前面提到的业务痛点已经基本得到了解决，但规则引擎在易用性和周边能力方面还有所欠缺。例如我们无法直观的查看当前系统运行的规则内容、注册事件数据；业务提交的规则与计算任务之间根据什么样的策略来进行分发；用户仍然需要订阅规则引擎的输出数据进行格式转换、写入目标存储等操作。

因此在第四阶段，整合了前面的方案，并不断丰富周边能力生态，打造了一站式实时规则平台。支持用户在平台上进行事件注册、预览、规则配置、规则调试、规则发布等全流程的自主操作，进一步提升工作效率。

规则 DSL 设计

为了实现业务自主配置规则，规则的语法必须清晰易懂。我们设计规则 DSL 整体结合了 JSON 和基础 SQL 语法，利用 JSON 的高可读性来描述规则的元数据、规则匹配属性等信息，利用 SQL 的强大表达力来描述 CEP 匹配条件以及匹配结果的处理逻辑。

这里我们发现了一个新的问题，如何通过 SQL 来表达事件是否满足匹配条件？SQL 可以查询哪些表？以一个具体的案例来回答这个问题。

假设要检测用户下单后是否发生了支付行为，那么规则编译生成的 NFA 可能是上图所示的样子。在规则运行时，我们将当前流入的事件以及当前规则的中间匹配结果，都以数据表的形式注册到上下文。当前流入的事件对应的表名称默认是 Events，规则中间匹配结果对应的表名称和它的 PatternName 保持一致。

在这个案例中，每个 SQL 可查询到的表就是三张，分别是 Events 表，表示当前流入的事件；Create_order 表，表示当前已经匹配到的下单事件；Pay_order 表，表示已经匹配到的支付事件。

在配置 SQL 时，就可以对已经注册到上下文的任意数据表进行查询。当 SQL 查询的结果非空时，就表示当前匹配条件判断通过。状态机经过 Take 边流转到下一个状态，并将事件保存到对应的表，否则就会到 Lgnore 边，丢弃掉事件。

再来看一下这个案例对应的规则配置条件的完整配置。整体是一个数组的形式，数组中每个元素表示一个 pattern，第二个 pattern 与前一个 pattern 之间的连接类型是 FOLLOWED_BY。第一个 pattern 的匹配条件是从流中检测用户下单事件，第二个 pattern 匹配条件是从流入检测用户支付事件。

注意，这个支付事件的订单是上一步我们缓存下来的下单事件对应的那个订单。经过上面的改造实现了，只要稍微有一些 SQL 基础的业务人员，都可以看懂并配置规则。

规则动态更新

前面我们提到，当前的 Flink CEP 计算任务不支持动态提交规则。主要原因是在编译阶段 Flink CEP 规则计算逻辑就确定了，并且已经通过 NFACompiler 编译完毕。在运行时计算任务只能固定执行之前已经编译好的规则。那么我们是如何改造的呢？

为了实现规则的动态发现，我们引入了一个规则流，用户提交或修改的规则都可以发到这条流中。为了实现规则的动态注入，我们将规则流设计为 Broadcast Stream。当发现新提交的规则时，广播分发到所有的 SubTask。

为了实现规则的在线加载执行，我们基于前面提到的规则 DSL，研发了一套基于规则的解析器。当 SubTask 收到分发的规则后，可以在线解析生成规则运行需要的组件。例如 NFA、规则匹配条件 SQL 对应的执行计划、匹配结果处理函数等。然后保存到 Flink State 中，持续检测和处理后续的事件。

解释一下为什么采用 Broadcast Stream 来实现规则的动态注入。由于 Flink CEP 是有状态的计算，规则的更新/删除往往需要伴随 Flink States 的操作和处理。例如：当删除规则时，连带当前规则关联的事件缓存等状态信息也需要一并删除。对比通过其他方式感知规则变更，比如启动一个异步线程定时扫描规则，通过 Broadcast Stream 的方式优势是，当检测到规则变更，能够更方便安全的操作 Flink State。

上面的方案解决了一个计算任务动态提交规则的诉求，但当一个计算任务运行多条规则时，又带来了一个新的问题。

问题一，由于规则的事件分组逻辑可能不同。(比如规则 A 需要先对事件流按照"用户的 IP 地址"路由到同一 Task 后再进行 NFA 匹配计算。而规则 B 则需要对事件流按”用户的设备 ID“进行路由）。那么当这两个规则运行在同一个计算任务时，如何兼容呢？

为了解决这个问题，我们新增了 KeyGenOperator 算子。当检测到新的事件流入时，先根据每一条规则配置生成一个与之对应分组的 Key，然后按分组 Key 再进行下游的 Task 分发，这样就实现了对多条规则的不同事件分组逻辑的兼容。

问题二，由于同一个计算任务运行多条规则，就可能会带来规则计算冗余的问题。比如，规则 A 关注用户下单、支付等支付相关事件，而规则 B 关注用户的商品浏览、评论等流量相关的事件。如果同一个计算任务同时运行这两条规则，那么这个任务就必须同时消费这两类事件。也就是说规则 A 本不关注流量类的事件，但由于整个任务整体订阅了这类事件，就导致规则 A 也必须处理这类事件。

为了解决上述问题，我们在 KeyGenOperator 算子新增了“事件筛选”组件，实现针对同一输入事件不同规则里的个性化事件筛选。也就是说，针对新流入的事件，仅当规则关注这个事件的时候，才会生成与之对应的分组 Key，并且进行后续的计算。

值得一提的是：在商家自定义预警的业务场景中，由于事件筛选的效果是比较好的（也就是说，商家自定义的每个规则仅关注当前商家所属商品的相关事件），那么经过我们测试，单个任务（在 600Core、800 并发度的情况下）可以支持的商家简单规则数量可以超过百万。

规则语义优化

当发生事件 A 后一段时间内，没有发生事件 B，其对应的伪代码可能是上面的这种形式。当前的 Flink CEP 不支持这种语义，因为可能造成没有事件触发这条规则，最终完成匹配的情况。

针对这个问题，我们在规则生成的 NFA 中引入一种 Pending 状态。当流入事件满足创建订单的条件之后，状态会随之迁移到 Pending 状态等待超时。当 Flink CEP 任务的 watermark 向前推进时，会触发 Pending 状态的 NFC 进行计算，判断是否已经超时，如果超时就会触发 NFA，迁移到下一个 Final 状态。如果在这之前系统流入了订单支付事件，就会转移到 Stop 状态。

通过这种方式，我们实现了对发生事件 A 之后一段时间内，没有发生事件 B 类的语义的支持。

整体架构

为了进一步提升规则引擎的应用性，我们整合前面的方案，拓展规则引擎的周边能力，研发了一站式规则平台。用户可以在平台上自助进行事件的注册、预览、规则配置、调试、发布等全流程的自助操作。

平台整个架构共分为四层，分别是：

事件层，例如看播事件、下单事件、物流事件、客服事件等。

计算层，负责动态的接收用户提交的 CEP 规则，并对规则进行解析，检测后续流入事件。计算层的核心是规则计算模块，也就是具体的 Flink CEP 计算任务。同时在计算层还有规则调度模块和规则解析模块，规则调度模块负责将新提交的规则分发到具体的 Flink CEP 计算任务，调度策略可以选择同事件源优先或者负载均衡优先。

• 同事件源优先是将关注相同 topic 的事件的规则，调动到同一个 Flink CEP 计算任务。例如将关注看播事件的规则调度到一个计算任务中，而将关注物流事件的规则调度到另一个计算任务中。负载均衡优先则是根据 Flink CEP 计算任务当前的负载情况，尽量将新提交的规则调度到相对空闲的计算任务执行。
• 规则解析模块负责当集团任务收到规则之后，解析并编译规则，生成规则运行时的组件。例如前面提到的 NFA、规则匹配条件对应的 SQL 执行计划等等。

触达层，负责计算层规则匹配结果的数据应用，主要包括延迟策略管理、维度字段扩充、推送目标管理等。

• 延迟策略管理主要负责当目标完成匹配后，是否立即进入下一个动作。例如，当用户完成既定的行为动作之后，可以选择立即发放优惠券，或者等待五分钟之后再发放优惠券。
• 维度字段扩充主要负责当目标完成匹配后，为数据补充相关的维度字段。例如，当用户完成浏览、下单、支付行为后，我们可以根据平台的配置，拼接补充订单关联的商品信息。例如商品的名称、价格等，供用户最终更好的决策。
• 推送目标管理主要负责当目标完成匹配后，具体需要执行的动作。例如当检测到用户有可能存在刷单行为时，给平台治理同学推送飞书消息。

平台层，负责与用户交互以及任务运维等工作。

业务成效方面：

• 业务自主配置规则，提升需求支持灵活性。目前共创建各类实时规则 2.5w+，服务平台运营同学 100+。
• 规则与计算任务解耦，无需研发介入即可支持规则创建/变更。业务规则需求支持平均耗时由 1day 缩减至 1hour。
• 提升 CEP 规则语义丰富度。规则引擎能力实现了抖音电商 70%+ 业务场景的覆盖。

技术成效方面：

• 由 Case By Case 的点状需求支持模式向面向平台的例行迭代转变，避免了单点人力阻塞问题，提升整体代码健壮性。
• 整体计算任务数量得到收敛，当前总体任务数量 ≤50，月均计算任务治理运维相关工作量降低 50%+。
• 降低计算任务整体资源浪费，单任务平均资源利用率提升 50%+。

未来计划在以下三个方面继续对规则引擎进行建设。

1. 继续打磨实时规则平台周边生态能力，实现更丰富、灵活的事件接入、触达方式。
2. 探索规则计算流批一体，打破离线、实时事件之间的壁垒，拓展平台应用范围。
3. 打通公司大数据研发环境，实现更加便捷的计算任务操作，进一步降低人工成本。