风控要回答的问题往往很朴素：同一个账户 5 分钟内给同一个人转了好几笔，加起来超过 1000 块，要不要拦一下？ 难的不是算这个数，而是这样的规则会越来越多，而且天天在改：今天阈值是 1000，明天要改成 2000；这条按账户统计，下一条要按收款人统计；有的看 5 分钟，有的看一整天。

如果每改一次规则都要改代码、重新打包、重启 Flink 作业，谁都受不了。本文要做的，就是把规则从代码里拿出来，变成可以随时下发、立刻生效的数据，让一个一直在跑的 Flink 作业始终按最新的规则工作——这就是规则引擎。

风控按时机可以分成事前（提前识别）、事中（实时识别）和事后（复盘分析），按方法可以分成基于规则和基于算法。规则又有两类：统计规则（如「5 分钟内访问超过 100 次」）和序列规则（如「点击 → 加入购物车 → 删除」）。本文只讨论最常见的一种：用统计规则做事中风控，数据一到，就按规则把异常挑出来。

先动手：一条规则是怎么触发告警的

实验分成交易流、动态规则和告警记录几块，预置了三条规则：

建议按这个顺序玩：

1. 连点 6 次「单步」。第 6 笔交易是 A101 → B201 的 $310，R1 命中：这对账户 5 分钟内已经有 4 笔，合计 $1,170，超过了 $1,000。在「告警记录」里点开这条告警，「这条告警为何触发」会列出窗口里的这 4 笔交易。

2. 再点 2 次。第 8 笔是 A105 → B201 的 $680，R1 没有反应（这对账户第一次转账），R2 却命中了：B201 在 5 分钟内一共收到 6 笔、$1,970。同一笔交易，换一种分组方式，看到的就是完全不同的东西。

3. 点 R1 的「编辑规则」，把阈值改成 2000，再点「应用规则」。下一笔交易立刻按新阈值判断，之前那条告警不受影响——规则改了，作业没停。

4. 启用 R3，或者把「交易场景」换成「集中汇入」，看看告警怎么变。

这几步正好对应规则引擎要解决的三件事：

• 规则随时会变：改阈值、加规则、停规则，都不能重启作业。→ 规则广播

• 每条规则的分组方式不同：R1 按账户对，R2 按收款人。→ 动态分组

• 每条规则的窗口和算法不同：5 分钟还是 1 天，求和还是计数。→ 动态窗口

下面一件一件来看。

为什么不能直接写一个 Flink 作业

如果只有 R1 这一条规则，用 Flink 写起来非常直接：

// 示意：只有 R1 一条规则时，直接用 DataStream API 写
transactions
    .keyBy(tx -> tx.payeeId + "#" + tx.beneficiaryId)        // 分组：付款人 + 收款人
    .window(SlidingEventTimeWindows.of(
        Duration.ofMinutes(5), Duration.ofMinutes(1)))       // 窗口：最近 5 分钟
    .sum("amount")                                           // 聚合：金额求和
    .filter(total -> total.amount > 1000);                   // 阈值：超过 1000

问题是，分组字段、窗口、聚合方式、阈值，这四样东西全都写死在代码里。Flink 在作业启动时就把它们编排成了固定的执行图，想改其中任何一样，都只能停作业、改代码、重新上线。规则一多，还会变成「一条规则一个作业」，资源和运维都扛不住。

规则引擎的思路，就是把这些写死的东西挪进规则：

规则本身也变成一条流，和数据一起流进作业。作业的代码只写一次：拿到一条数据，看看现在有哪些规则，按规则处理。

一条规则长什么样

QFlink 里一条规则是一段 JSON。下面这条规则的意思是：从原始日志里挑出含 stdout 的行，拆出时间、关键字和 JSON 里的两个字段，按 key1 的奇偶分组，统计每天 key2 有几个不同的值，超过 1 个就告警。

{
    "rule_id": 100,
    "rule_state": "ACTIVE",
    "clear_state": false,
    "job_id": "test",
    "filter": {
        "expr": "string.contains(raw, 'stdout')",
        "params": ["raw"]
    },
    "flat_map": {
        "pattern": "(\\d{4}-\\d{2}-\\d{2} \\d{2}:\\d{2}:\\d{2}) (.*?) (.*)",
        "normal_fields": [
            {"name": "time", "type": "STRING", "default": null},
            {"name": "key_word", "type": "STRING", "default": null},
            {
                "name": "",
                "type": "JSON",
                "json_paths": [
                    {"name": "key1", "expr": "$.key1", "type": "INT", "default": 0},
                    {"name": "key2", "expr": "$.key2", "type": "STRING", "default": null}
                ],
                "default": null
            }
        ],
        "mapping_fields": [
            {"name": "is_odd", "type": "BOOLEAN", "expr": "key1 % 2 == 1", "default": null}
        ]
    },
    "keys": ["is_odd"],
    "window": {
        "type": "TUMBLE",
        "trigger": "SINGLE",
        "offset": 0,
        "size": 86400000,
        "step": 0
    },
    "aggregates": [
        {"name": "val_cnt", "inputs": ["key2"], "method": "COUNT_DISTINCT"}
    ],
    "threshold": {
        "expr": "val_cnt > 1",
        "params": ["val_cnt"]
    },
    "extras": {"etl.table": "test_topic"}
}

从上往下读，就是一条数据要走的六步：

1. 筛（filter）：表达式 string.contains(raw, 'stdout') 为真才往下走。

2. 拆（flat_map）：用正则把一行日志拆成三段，第三段是 JSON，再用 JSONPath 取出 key1、key2；最后用表达式算出新字段 is_odd。

3. 分组（keys）：按 is_odd 分组，奇数一组，偶数一组。

4. 开窗（window）：滚动窗口，一天一个（size 的单位是毫秒，86400000 就是 24 小时）。

5. 算（aggregates）：窗口里 key2 去重计数，结果叫 val_cnt。

6. 判（threshold）：val_cnt > 1 就输出一条告警。

剩下几个字段是规则的「身份证」：rule_id 是主键；rule_state 控制上线（ACTIVE / UPDATE）和下线（REMOVE）；job_id 决定这条规则由哪个 Flink 作业执行；clear_state 表示下发时顺便清掉这条规则攒下的统计，后面会讲为什么需要它。

实验里也能看到规则的 JSON（判断依据面板下方的「查看当前规则 JSON」）。那是 Flink 官方欺诈检测示例的格式，字段名不一样，但分组字段、聚合方式、比较符、阈值、窗口这几样一个不少。

整体架构：两条流，两个算子

作业有两条输入：

• 数据流：日志或交易，量大，并行度高。

• 规则流：规则的新增、修改和删除，量很小，并行度为 1。

规则流通过 Flink 的广播状态发给下游的每一个并行实例。可以把它想成「贴公告」：规则一更新，每个工位都收到一份，各自存进自己的广播状态——一个 rule_id → 规则 的 Map。ACTIVE / UPDATE 就是覆盖写入，REMOVE 就是删除，job_id 不属于本作业的规则直接忽略。

数据流先后经过两个算子：

• ① ETL 算子：把一条数据按每条规则处理一遍，算出它在这条规则下属于哪个分组。

• ② 聚合算子：按 (rule_id, key) 分好组，在各自的窗口里累计、判断阈值、输出告警。

解析不了的日志、迟到太久的数据、不合法的规则都不会被悄悄丢掉，而是写进单独的错误输出，方便排查。

这个思路最早来自 Flink 官方博客的欺诈检测系列，上面的实验也参考了它的示例项目。QFlink 在此基础上补上了日志解析、多种窗口、状态清理和性能优化。

第一步 ETL：一条数据，按每条规则各走一遍

ETL 算子收到一条数据后，会遍历广播状态里的所有规则：先过滤、再解析，然后按规则的 keys 取出对应字段的值，用 # 连起来作为 key，最后给数据加上两列：rule_id 和 key。

这里有一个关键的小技巧。Flink 的 keyBy 必须在作业启动时就确定按什么分组，没法「每条规则按不同字段分组」。换个角度就简单了：按哪些字段分组由规则决定，key 的值由 ETL 算子提前算好。下游只要永远按 (rule_id, key) 这两列分组——这个 keyBy 是固定的，动态的部分已经在上游消化掉了。

实验里的「03 · 分组与窗口」显示的就是当前有多少个这样的 key。R1 和 R2 的 key 不同，所以同一笔交易会同时出现在两个分组里。

如果暂时不需要聚合（作业参数 agg.enable = false），这个作业就是一个纯 ETL 作业：按规则过滤、解析大流量的原始日志，再按规则里的 extras.etl.table 分发到不同的 Kafka Topic，给下游各自使用。数据源量级很大时，先统一读一遍、拆出关心的数据，本身就能减轻公共层的压力。

{
    "type": "record",
    "name": "default",
    "fields": [
        {"name": "rule_id", "type": "int"},
        {"name": "key", "type": "string"},
        {"name": "time", "type": "string"},
        {"name": "key_word", "type": "string"},
        {"name": "key1", "type": "int"},
        {"name": "key2", "type": "string"},
        {"name": "is_odd", "type": "boolean"}
    ]
}

第二步聚合：自己写一个「能变」的窗口

分好组以后，要在窗口里累计。Flink 自带的窗口算子同样在启动时就定死了窗口的类型和大小，所以这里得自己写一个窗口算子。好在 Flink 窗口的原理并不复杂（可以参考这篇源码分析），无非五步：

1. 分配：判断这条数据属于哪些窗口；

2. 累计：把数据聚合进对应窗口的状态；

3. 触发：决定什么时候算出窗口的结果；

4. 定时：注册事件时间定时器，到点触发计算，窗口结束后清理状态；

5. 迟到：来得太晚的数据放进侧输出。

我们的算子把这五步照做一遍，唯一的区别是：每来一条数据，先用它的 rule_id 到广播状态里查出规则，再按规则里写的窗口去分配、累计和触发。规则一改，下一条数据就按新窗口算。

三种窗口

滑动和累积窗口有一个共同点：一条数据同时属于好多个窗口。「最近 1 小时、每分钟滑一次」，一条数据会落进 60 个窗口；「今天累计、每分钟更新」，零点刚过的一条数据要算进当天 1440 个窗口。

最早的实现和 DataStream API 的窗口一样，把数据往它所属的每个窗口里都累计一遍。窗口越长、步长越短，状态读写就成倍增加，性能会非常差。

切片：每条数据只记一次

解决办法借鉴自 Flink SQL 的窗口表值函数（原理可以看这篇介绍）：把时间切成最小的、互不重叠的小段，叫切片（slice）。滚动窗口的切片就是窗口本身，滑动和累积窗口的切片长度等于步长。

打个比方：记账本每分钟一页，每笔账只记在它发生的那一页。有人问「最近一小时花了多少」，就把最近 60 页的小计加起来，而不是把每笔账抄 60 遍。

还有两个细节：

• 累积窗口每走一步，会把刚结束的切片并进窗口的第一个切片。下一步只需要再并一片，不用把整个窗口重新读一遍。

• 一个切片不再被任何窗口需要时（用到它的最后一个窗口已经触发，并且过了允许迟到的时间），定时器会把它删掉，状态不会无限增长。

什么时候出告警

规则里的 trigger 决定告警的时机：

实验里的规则都是 SINGLE 式的：每笔交易进来立即判断。不过为了好懂，实验的窗口是「以当前这笔交易为终点，往回数 5 分钟」；QFlink 的窗口则对齐在切片的边界上（比如整分钟）。这是模拟和真实实现的一个区别。

动态规则的三个坑

坑一：Watermark 被规则流拖住

两条流汇进同一个算子时，算子的 Watermark（事件时间走到了哪里）取的是两边的最小值。规则流可能几个小时都没有一条新规则，它的 Watermark 一直不动，整个算子的事件时间就被卡住，窗口永远不会触发。

解决方法是让规则流定期发出一个「无穷大」的 Watermark（Long.MAX_VALUE），相当于告诉 Flink：按时间算，规则流永远不会拖后腿。这样算子的 Watermark 就完全由数据流决定。

坑二：规则删了，状态还在

每条规则在每个 key、每个切片上都攒着统计结果。规则从广播状态里删掉以后，这些状态并不会自己消失；规则多、key 多，时间一长，内存就会被撑爆。

所以规则多了一个开关 clear_state。下发时带上它，聚合算子会找出这条规则在所有 key、所有切片上的状态，逐个清掉。遍历时要先把这些 key 复制一份再删，避免一边遍历一边修改。规则口径变了、之前的累计不再有意义时，也可以用它从零开始。

实验里也能感受到：R3 是「命中后清空」的计数规则，提醒一次后它的分组会归零重新数；「清空窗口状态」按钮则会清掉所有分组里攒下的交易。

坑三：规则不会同时到达每个地方

广播只保证每个并行实例最终都会收到规则，不保证同一时刻收到。规则更新的那一瞬间，有的实例已经在用新规则，有的还在用旧规则；ETL 算子可能已经按新规则输出了数据，聚合算子却还没收到这条规则。代码要容忍这种短暂的不一致：比如找不到规则的数据直接跳过，key 里的字段个数和规则对不上时也不报错。

另外，规则只影响之后的数据，改了规则，历史告警不会重算。实验里改完阈值，旧告警的依据保持不变，就是这个效果。

跑得更快：攒一批，先合并

规则引擎往往放在数据量最大的地方，性能值得单独说说。QFlink 借鉴了 Flink SQL 的两个优化，都可以用作业参数打开。

微批（mini batch）：数据先在内存里攒一会儿、先聚合好，再一次性写进状态。一个切片从「每条数据读写一次」变成「每批读写一次」，在 RocksDB 这种每次访问都要序列化的状态后端上，差别非常明显。攒够条数、超过允许的延迟、Watermark 到来之前、做 checkpoint 之前，都会把这一批刷进状态。

两阶段聚合：在 shuffle 之前，先在上游按 (rule_id, key, 切片) 聚合一次，网络上传的不再是一条条数据，而是聚合后的中间结果。它最适合「少数几个 key 占了大部分流量」的情况。运行下面的实验，拖动热点比例感受一下：热点越集中，单个任务的压力越大。

开启两阶段聚合后，热点 key 的数据在每个上游任务里先合并成一条中间结果再发给下游，最忙的那个任务就不必逐条处理全部热点数据了。

测试与控制台

规则引擎是给很多人用的平台作业，测试必须扎实（可以参考 Flink 的测试文档）：

• 单元测试：切片分配、聚合函数、表达式这些纯逻辑直接测。

• 算子测试：用 Flink 的测试工具驱动 ETL 和聚合算子，手动推进 Watermark，检查窗口在正确的时间触发、状态按时清理。

• 作业测试：source 和 sink 是可插拔的。本地运行时，规则、日志、告警都读写本地文件，不需要 Kafka 也能端到端跑通整个作业。

本地还可以配一个控制台 QFlink-Web：在页面上用表单或 JSON 编辑规则，下发、下线规则，查看告警和指标走势，还能按模板生成测试日志。它从不直接和作业通信，只往规则通道里写、从告警通道里读——本地是文件，线上是 Kafka，换环境不用改页面。可以把它看作本文开头那个实验的「真实版」。

业界的做法

• 社区提出过动态 CEP 的提案，希望 Flink CEP 也能动态更新规则。

• 业界已有对应实现：阿里云的商业版本（介绍文章），以及字节跳动的实践。

• 本文处理的是统计规则（一段时间内的总和、次数、去重数）。像「点击 → 加入购物车 → 删除」这样的序列规则需要 CEP，思路上同样可以用「广播规则 + 自己写算子」，但 CEP 的规则配置复杂得多，设计好用的前端、降低使用门槛才是关键。

总结

回头看，规则引擎做的事可以浓缩成三句话：

1. 规则即数据：规则变成一条广播流，作业不停，规则随时生效。

2. 动态分组 = 提前算好 key：ETL 算子按规则算出 key，下游永远按 (rule_id, key) 分组。

3. 动态窗口 = 自己写窗口算子：按规则查窗口，用切片让每条数据只记一次，再处理好 Watermark 和状态清理。

一路走下来，不只得到了一个能用的规则引擎，也把 Flink 的广播状态、窗口、定时器和状态管理都摸了一遍，算是向「知其所以然」又迈了一步~

参考资料

• QFlink 规则引擎代码

• Advanced Flink Application Patterns Vol.1: Case Study of a Fraud Detection System

• afedulov/fraud-detection-demo

• Flink 广播状态

• Flink 源码阅读笔记：时间和窗口

• Flink SQL 窗口表值函数

• Flink 序列化调优

• AviatorScript