实时监控是 Flink 一个重要而复杂的应用场景，一般不会只用 SQL 来做。这篇文章偏要试试：从一个简单的需求出发，只用 SQL 应对越来越复杂的要求，看看 SQL 的边界能被推到哪里。

全文的演进路线是这样的：

阶段	规则数	做法	改一条规则要做什么	主要问题
一	1	一句 SQL	改代码、重启作业	没有
二	几条	多句 SQL 用 UNION ALL 合并	改代码、重启作业	重复消费、代码膨胀
三	很多	规则存进表，SQL 关联规则表	改表里的一行	每条数据都要查规则表
四	很多	规则表以流的方式读入	改表里的一行	告警、窗口、状态难以精细控制

准备

有一张日志流表，数据来自仓库里的 sources/common-data.txt，由 QFlink 的 qfile 连接器逐行读出：

sql

CREATE TABLE IF NOT EXISTS `dwd_log` (
    `time` STRING COMMENT '事件时间',
    `key_word` STRING COMMENT '日志标识',
    `key1` INT COMMENT '指标',
    `key2` STRING COMMENT '维度'
) WITH (
    ...
);

数据形如：

json

{"time": "2023-04-09 15:40:05", "key_word": "stdout", "key1": 5, "key2": "val1"}

我们要按照一些规则，从这条日志流里筛出需要关注的事件。

阶段一：只有一条规则

规则 1：关键字为 stdout 的日志，每天每个 key2 下 key1 出现的次数超过 1 次就告警。

一句 SQL 就能实现：

sql

SELECT
    1 AS `rule_id`,
    DATE_FORMAT(`time`, 'yyyy-MM-dd 00:00:00') AS `window_start`,
    `key2` AS `key`,
    MAX(`time`) AS `alert_time`,
    MAP['key1_cnt', CAST(COUNT(`key1`) AS DOUBLE)] AS `alert_metrics`
FROM `dwd_log`
WHERE `key_word` = 'stdout'
GROUP BY `key2`, DATE_FORMAT(`time`, 'yyyy-MM-dd 00:00:00')
HAVING COUNT(`key1`) > 1;

shell

tutorials/run.sh sql_is_all_your_need/flink_dynamic_sql/step1_start.sql

text

6> +U[1, 2023-04-09 00:00:00, val2, 2023-04-09 15:42:05, {key1=2.0}]

val2 当天出现了 2 次，超过阈值 1，输出一条告警。

单条、简单的统计规则，用 Flink SQL 写非常方便。可规则越来越多呢？

阶段二：规则开始增加

规则 2：每个关键字每小时内不同 key2 的数量超过 2 个就告警。

最简单的办法是再起一个作业。如果想在同一个作业里完成，可以把两条规则用 UNION ALL 拼起来：

sql

SELECT
    1 AS `rule_id`,
    DATE_FORMAT(`time`, 'yyyy-MM-dd 00:00:00') AS `window_start`,
    `key2` AS `key`,
    MAX(`time`) AS `alert_time`,
    MAP['key1_cnt', CAST(COUNT(`key1`) AS DOUBLE)] AS `alert_metrics`
FROM `dwd_log`
WHERE `key_word` = 'stdout'
GROUP BY `key2`, DATE_FORMAT(`time`, 'yyyy-MM-dd 00:00:00')
HAVING COUNT(`key1`) > 1

UNION ALL

SELECT
    2 AS `rule_id`,
    DATE_FORMAT(`time`, 'yyyy-MM-dd HH:00:00') AS `window_start`,
    `key_word` AS `key`,
    MAX(`time`) AS `alert_time`,
    MAP['key2_cnt', CAST(COUNT(DISTINCT `key2`) AS DOUBLE)] AS `alert_metrics`
FROM `dwd_log`
GROUP BY `key_word`, DATE_FORMAT(`time`, 'yyyy-MM-dd HH:00:00')
HAVING COUNT(DISTINCT `key2`) > 2;

shell

tutorials/run.sh sql_is_all_your_need/flink_dynamic_sql/step2_grow.sql

text

6> +U[1, 2023-04-09 00:00:00, val2, 2023-04-09 15:42:05, {key1_cnt=2.0}]
11> +U[2, 2023-04-09 15:00:00, stdout, 2023-04-09 15:43:05, {key2_cnt=3.0}]
11> -U[2, 2023-04-09 15:00:00, stdout, 2023-04-09 15:43:05, {key2_cnt=3.0}]
11> +U[2, 2023-04-09 15:00:00, stdout, 2023-04-09 15:44:05, {key2_cnt=4.0}]

规则 2 的结果先是 3，来了 val4 之后变成 4：下游先收到撤回旧值的 -U，再收到新值 +U。达到阈值以后每次更新都会再输出一次，后面「SQL 的边界」一节会讲到这一点。

逻辑简单，但每多一条规则，数据就要被多处理一遍，规则也硬编码在代码里。必须在规则多到失控之前，找到更好的办法。

阶段三：把规则变成数据

规则多起来以后，问题集中在三点：

重复处理：每条规则都把数据过一遍，浪费计算资源；
代码膨胀：几十条规则拼在一起，没法维护；
硬编码：改一个阈值也要改代码、重新上线。

要解决它们，就得找到一种SQL 本身不变、只改规则的写法。仔细看规则 1 和规则 2，会发现它们由同样的五个部分组成，每一部分恰好对应 SQL 的一个子句：

一条监控规则的五个部分，恰好对应 SQL 的五个子句

既然每条规则都只是这五个部分取值不同，就可以把规则抽象成一张表，存在 Paimon 里：

sql

CREATE TABLE IF NOT EXISTS `dim_rule` (
    `rule_id` INT COMMENT '规则 ID',
    `job_id` STRING COMMENT '处理这条规则的作业 ID',
    `filter` STRING COMMENT '过滤规则',
    `key` STRING COMMENT '分组规则',
    `window` STRING COMMENT '窗口规则',
    `aggregate` ROW<`name` STRING, `input` STRING, `method` STRING> COMMENT '聚合规则',
    `threshold` STRING COMMENT '阈值规则',
    PRIMARY KEY (`rule_id`) NOT ENFORCED
);

prepare.sql 建好这张 Paimon 表，写进规则 1 和规则 2：

shell

tutorials/run.sh sql_is_all_your_need/flink_dynamic_sql/prepare.sql

规则 1 在这张表里就是一行：

字段	值	含义
rule_id	1
job_id	test	由哪个作业处理
filter	`EXPR$0 == 'stdout'`	关键字为 stdout
key	`EXPR$2`	按 key2 分组
window	`DAY`	按天
aggregate	`{name: key1_cnt, input: EXPR$1, method: COUNT}`	对 key1 计数，结果命名为 key1_cnt
threshold	`key1_cnt > 1`	计数超过 1 告警

规则里引用字段时写的是 EXPR$0、EXPR$1、EXPR$2，分别对应 key_word、key1、key2，原因后面会说。

有了规则表，就可以把每条日志和它要遵守的规则关联起来，再按规则去处理。这需要几个 UDF 来增强 SQL 的表达能力（具体实现）：

函数	类型	作用
`dynamic_filter`	标量函数	按过滤规则或阈值规则的表达式判断是否满足
`dynamic_key`	标量函数	按分组规则从日志里取出分组值
`dynamic_window`	标量函数	按窗口规则算出日志所属窗口的起点，相当于可配置的 `DATE_FORMAT`
`dynamic_agg`	聚合函数	按聚合规则计算，输出「指标名 → 值」的映射

重构后的 SQL 是这样的：

sql

SELECT
    `rule_id`,
    `window_start`,
    `key`,
    MAX(`time`) AS `alert_time`,
    -- 按聚合规则计算
    dynamic_agg(`data`, `aggregate`) AS `alert_metrics`
FROM (
    SELECT
        `rule_id`,
        dynamic_window(`time`, `window`) AS `window_start`,
        dynamic_key(`data`, `key`) AS `key`,
        `aggregate`,
        `threshold`,
        `time`,
        -- data 是 ROW(key_word, key1, key2)：规则会用到哪个字段事先不知道，所以整行都带上
        `data`
    FROM `dwd_log` AS `log`
    LEFT JOIN
        -- 查询 Paimon 规则维表
        `dim_rule` FOR SYSTEM_TIME AS OF `log`.`proc_time` AS `rule`
    -- 只取本作业负责的规则
    ON `rule`.`job_id` = 'test'
    WHERE
        -- 按过滤规则筛选日志
        dynamic_filter(`data`, `filter`) AND `rule_id` IS NOT NULL
) `log_with_rule`
GROUP BY
    -- 按规则分组
    `key`, `rule_id`, `window_start`, `threshold`, `aggregate`
HAVING
    -- 按阈值规则判断聚合结果
    dynamic_filter(dynamic_agg(`data`, `aggregate`), `threshold`);

这样一来，SQL 固定不变，新增、修改、删除规则都只是对规则表的一次增删改。一条日志如果命中三条规则，会被关联成三行，各自按自己的规则分组、聚合和判断。

shell

tutorials/run.sh sql_is_all_your_need/flink_dynamic_sql/step3_explode.sql

text

1> +U[1, 2023-04-09 00:00:00, val2, 2023-04-09 15:42:05, {key1_cnt=2.0}]
15> +U[2, 2023-04-09 15:00:00, stdout, 2023-04-09 15:43:05, {key2_cnt=3.0}]
15> -U[2, 2023-04-09 15:00:00, stdout, 2023-04-09 15:43:05, {key2_cnt=3.0}]
15> +U[2, 2023-04-09 15:00:00, stdout, 2023-04-09 15:44:05, {key2_cnt=4.0}]

结果和阶段二完全一样，但 SQL 里已经没有任何一条具体的规则了。两条规则都来自 dim_rule，加规则只需要往表里插一行。

还会有什么问题吗？

阶段四：规则表变成一条流

上面用的是维表关联：每来一条日志，都要去规则表里查一次。日志和规则都多了以后，查询压力会成为瓶颈。可以用异步查询、缓存等 Lookup Join 的优化缓解，但治标不治本。

好在 Paimon 支持流式读取：可以先把规则表全量读进 Flink，之后持续捕获规则的变更。规则存在 Flink 的状态里，日志来了直接在本地匹配，不再访问外部存储。只需改一行：

sql

SELECT ...
FROM (
    SELECT ...
    FROM `dwd_log` AS `log`
    LEFT JOIN
        -- 改成普通的关联：规则表以流的方式读入
        `dim_rule` AS `rule`
    ON `rule`.`job_id` = 'test'
    ...
) `log_with_rule`
...

shell

tutorials/run.sh sql_is_all_your_need/flink_dynamic_sql/step4_challenge.sql

输出和阶段三一样，区别只在规则怎么到达：从每条日志查一次规则表，变成规则表本身作为一条流读进状态。

规则表以流的方式读入，与日志关联后按规则过滤、分组、开窗、聚合并判断阈值

用 SQL 实现动态规则监控的目标似乎达到了。但真的是这样吗？

真相：SQL 的边界

真正上线后，还会遇到一系列棘手的问题：

告警控制：SQL 在一个周期内每次更新都会输出，达到阈值后会反复告警。想控制告警次数，只能交给下游处理。
窗口类型：SQL 只能在周期内实时累积，做不到「每天定时检查一次」这类按时间触发的逻辑。
状态控制：规则修改后，之前累积的状态没法精细地清理或重置；作业长期运行后，状态会越积越多，不堪重负。

这些问题要求对状态、定时器和输出有精细的控制，已经超出了 SQL 的能力范围，需要回到 DataStream API 自己实现算子，这就是 Flink 规则引擎一文的做法，那篇文章里还有一个可以直接操作的欺诈检测实验。

不过回到实际工作中，大部分监控需求其实就是阶段一那样确定的规则，用 SQL 都能方便地解决。从长远看，上层的数据应用会变得越来越简单：对最终用户来说，所有数据都可以用 SQL 来分析，这就是我理解的「SQL Is All Your Need」的愿景。

总结

这篇文章看到了 SQL 解决复杂问题的潜力，也看到了它的边界。把变化的部分抽象成数据、让不变的代码去解释数据，这个思路本身比 SQL 更重要：换成 DataStream API，规则引擎用的也是同样的抽象。

随着大模型的发展，也许有一天可以直接用自然语言处理数据。技术在变，保持跟进就好。

试试看

阶段三运行时，另开一个终端，照着 prepare.sql 写一句 INSERT 往 dim_rule 加一条规则，用 tutorials/run.sh 执行，看新规则能不能在不重启作业的情况下生效。

本文的全部 SQL 在 QFlink，UDF 在 qflink-sql-udf，运行说明见 QFlink 教程第 03 章。

参考资料

全文评论

正在加载评论…