大数据开发说穿了，就是数据从一个存储系统出发，经过计算引擎加工，再写进另一个存储系统。如果把每个存储系统都抽象成一张表，用 SQL 来描述加工过程，它和传统数据库里的查询就没有本质区别了。

这篇文章用 Apache Paimon（存储）和 Flink SQL（计算）实现一组常见的数据开发场景，作为「SQL Is All Your Need」系列的第一步。全文的链路是这样的：

全文的链路：原始日志经 ETL 变成明细，再分别算出每分钟指标、告警和模式匹配

准备：一张不断写入的日志表

数据采集不在本文范围内。prepare.sql 建好一张 Paimon 表 ods_log，再写进 5 条原始日志，相当于一张不断接收日志的表：

shell

tutorials/run.sh sql_is_all_your_need/prepare.sql

建表部分是这样的：

sql

-- 用 Paimon Catalog 管理元数据，表的定义会持久化下来
CREATE CATALOG my_catalog WITH (
    'type' = 'paimon',
    'warehouse' = 'file:/tmp/paimon'
);
USE CATALOG my_catalog;

CREATE TABLE IF NOT EXISTS ods_log (
    log STRING
) WITH (
    -- 只追加、不更新
    'write-mode' = 'append-only'
);

每条日志形如：

plaintext

2023-04-09 15:40:05 stdout {"key1":5,"key2":"val1"}
2023-04-09 15:41:05 stdout {"key1":4,"key2":"val2"}
2023-04-09 15:42:05 stdout {"key1":1,"key2":"val2"}
2023-04-09 15:43:05 stdout {"key1":5,"key2":"val3"}
2023-04-09 15:44:05 stdout {"key1":6,"key2":"val4"}

一行里混着时间、关键字和一段 JSON，下游没法直接用。

ETL：把日志变成结构化的表

第一步是清洗：用正则拆出时间、关键字和 JSON，再从 JSON 里取出字段。借助 Flink 的流处理能力，这条 INSERT 一旦提交，就会变成一个持续运行的管道，新日志写进 ods_log，几秒后就出现在 dwd_log 里。

sql

CREATE TABLE IF NOT EXISTS `dwd_log` (
    `time` STRING,
    `key_word` STRING,
    `key1` INT,
    `key2` STRING,
    -- 事件时间：用日志里记录的时间，而不是数据到达的时间
    `rowtime` AS TO_TIMESTAMP_LTZ(UNIX_TIMESTAMP(`time`), 0),
    -- 水位线：容忍 10 秒以内的乱序
    WATERMARK FOR `rowtime` AS `rowtime` - INTERVAL '10' SECOND
) WITH (
    'write-mode' = 'append-only'
);

INSERT INTO `dwd_log`
SELECT
    REGEXP_EXTRACT(`log`, '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*?) (.*)', 1) AS `time`,
    REGEXP_EXTRACT(`log`, '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*?) (.*)', 2) AS `key_word`,
    JSON_VALUE(REGEXP_EXTRACT(`log`, '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*?) (.*)', 3), '$.key1' RETURNING INT) AS `key1`,
    JSON_VALUE(REGEXP_EXTRACT(`log`, '(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}) (.*?) (.*)', 3), '$.key2') AS `key2`
FROM `ods_log`;

建表语句里有两个流处理特有的概念：

事件时间：数据在业务上发生的时间。网络抖动会让数据晚到、乱序，按事件时间统计，结果才不受到达顺序影响。
水位线（Watermark）：告诉引擎「事件时间早于某个时刻的数据应该都到齐了」。上面的定义表示最多等 10 秒的迟到数据，水位线越过窗口的结束时间，窗口才会输出结果。

跑一下，再看看 dwd_log 里有什么（这里用批模式，读完 5 条日志就结束）：

shell

tutorials/run.sh --batch sql_is_all_your_need/flink_sql/etl.sql
tutorials/run.sh --batch tutorials/show.sql --table dwd_log --columns '`time`, key_word, key1, key2'

text

+----+---------------------+----------+------+------+
| op |                time | key_word | key1 | key2 |
+----+---------------------+----------+------+------+
| +I | 2023-04-09 15:40:05 |   stdout |    5 | val1 |
| +I | 2023-04-09 15:41:05 |   stdout |    4 | val2 |
| +I | 2023-04-09 15:42:05 |   stdout |    1 | val2 |
| +I | 2023-04-09 15:43:05 |   stdout |    5 | val3 |
| +I | 2023-04-09 15:44:05 |   stdout |    6 | val4 |
+----+---------------------+----------+------+------+

show.sql 打印任意一张表。批模式下 SELECT * 碰到 rowtime 这种时间属性列会报 Flink 的规划错误，所以用 --columns 列出其余几列。

分析：按窗口统计

有了结构化数据，就可以算指标了，比如每分钟统计一次每个关键字下有多少种不同的 key2。Flink SQL 用窗口表值函数（Windowing TVF）表达窗口：TUMBLE 把时间切成首尾相接、互不重叠的一分钟，每条数据恰好属于一个窗口。

5 条日志各落进一个 1 分钟窗口；最后一条日志 15:44:05，水位线因此停在 15:43:55

sql

CREATE TABLE IF NOT EXISTS dws_metric (
    `window_start` STRING,
    `window_end` STRING,
    `dim` STRING,
    `metric` BIGINT,
    PRIMARY KEY (`window_start`, `window_end`, `dim`) NOT ENFORCED
);

INSERT INTO dws_metric
SELECT
    DATE_FORMAT(`window_start`, 'yyyy-MM-dd HH:mm:ss') AS `window_start`,
    DATE_FORMAT(`window_end`, 'yyyy-MM-dd HH:mm:ss') AS `window_end`,
    `key_word` AS `dim`,
    COUNT(DISTINCT `key2`) AS `metric`
FROM TABLE(
    TUMBLE(TABLE `dwd_log`, DESCRIPTOR(`rowtime`), INTERVAL '1' MINUTES))
GROUP BY `key_word`, `window_start`, `window_end`;

先用批模式跑：

shell

tutorials/run.sh --batch sql_is_all_your_need/flink_sql/analysis.sql
tutorials/run.sh --batch tutorials/show.sql --table dws_metric

text

+----+---------------------+---------------------+--------+--------+
| op |        window_start |          window_end |    dim | metric |
+----+---------------------+---------------------+--------+--------+
| +I | 2023-04-09 15:40:00 | 2023-04-09 15:41:00 | stdout |      1 |
| +I | 2023-04-09 15:41:00 | 2023-04-09 15:42:00 | stdout |      1 |
| +I | 2023-04-09 15:42:00 | 2023-04-09 15:43:00 | stdout |      1 |
| +I | 2023-04-09 15:43:00 | 2023-04-09 15:44:00 | stdout |      1 |
| +I | 2023-04-09 15:44:00 | 2023-04-09 15:45:00 | stdout |      1 |
+----+---------------------+---------------------+--------+--------+

生产环境中数据量大了会遇到性能问题，Flink 提供了 MiniBatch、Local-Global、Split Distinct 等调优手段，原理可以参考大数据性能调优。

一行切换成批处理

实时流偶尔会出问题，比如上游数据延迟或者逻辑有 bug，这时通常要用离线方式重新计算修正。传统的 Lambda 架构里，实时和离线是两套存储、两套引擎、两份代码，修数时要把逻辑在另一套系统里再写一遍。

Paimon + Flink 的组合则是流批一体的：同一张表既能流式读写，也能批量读写；同一段 SQL 只需加一行配置，就能切成批模式运行：

sql

SET 'execution.runtime-mode' = 'batch';

-- 下面是和流模式完全相同的 INSERT INTO dws_metric ...

这才是真正意义上的流批一体：同一套 API、同一种开发方式完成流计算和批计算，从根本上保证两者的处理逻辑和结果一致。Paimon 支持按主键更新，批模式算出来的结果会直接覆盖之前的错误值。

同一段 SQL 用流模式再跑一遍，就能看到两种模式的区别。先把表重建，流模式的作业不会自己结束，等十几秒看到它提交后按 Ctrl-C：

shell

tutorials/run.sh sql_is_all_your_need/prepare.sql
tutorials/run.sh --batch sql_is_all_your_need/flink_sql/etl.sql
tutorials/run.sh sql_is_all_your_need/flink_sql/analysis.sql      # 十几秒后 Ctrl-C
tutorials/run.sh --batch tutorials/show.sql --table dws_metric

text

+----+---------------------+---------------------+--------+--------+
| op |        window_start |          window_end |    dim | metric |
+----+---------------------+---------------------+--------+--------+
| +I | 2023-04-09 15:40:00 | 2023-04-09 15:41:00 | stdout |      1 |
| +I | 2023-04-09 15:41:00 | 2023-04-09 15:42:00 | stdout |      1 |
| +I | 2023-04-09 15:42:00 | 2023-04-09 15:43:00 | stdout |      1 |
+----+---------------------+---------------------+--------+--------+

同一段 SQL，流模式少了两个窗口。水位线停在 15:43:55（上图），流模式只有在水位线越过窗口终点时才输出这个窗口，15:43 和 15:44 两个窗口还在等可能迟到的数据；批模式知道输入已经读完，所有窗口直接输出。修数、回溯时用批模式跑同一段 SQL，结果和流模式最终一致，但不用等水位线。

事件驱动：实时监控

实时场景里另一类重要应用是事件驱动：不是为了出报表，而是为了在某件事发生时立刻做出反应，典型的就是监控和风控。比如监控每个关键字当天的累计值，超过 10 就告警：

sql

-- 状态保留 2 天，过期的分组状态会被清理
SET 'table.exec.state.ttl' = '2 d';

CREATE TABLE IF NOT EXISTS `dws_alert` (
    `window_start` STRING,
    `key` STRING,
    `time` STRING,
    `metric` BIGINT,
    PRIMARY KEY (`window_start`, `key`) NOT ENFORCED
);

INSERT INTO `dws_alert`
SELECT
    DATE_FORMAT(`time`, 'yyyyMMdd') AS `window_start`,
    `key_word` AS `key`,
    MAX(`time`) AS `time`,
    SUM(`key1`) AS `metric`
FROM `dwd_log`
GROUP BY DATE_FORMAT(`time`, 'yyyyMMdd'), `key_word`
HAVING SUM(`key1`) > 10;

这里没有用窗口函数，而是普通的 GROUP BY 加 HAVING：每来一条数据就更新一次累计值，一旦超过阈值就立刻输出，不必等到一天结束。

shell

tutorials/run.sh --batch sql_is_all_your_need/flink_sql/event_driven.sql
tutorials/run.sh --batch tutorials/show.sql --table dws_alert

text

+----+--------------+--------+---------------------+--------+
| op | window_start |    key |                time | metric |
+----+--------------+--------+---------------------+--------+
| +I |     20230409 | stdout | 2023-04-09 15:44:05 |     21 |
+----+--------------+--------+---------------------+--------+

当天 key1 累加到 5 + 4 + 1 + 5 + 6 = 21，超过 10，写出一条告警。在流模式下，这条 SQL 每来一条数据结果就更新一次，下游会先收到「撤回旧值」、再收到新值，这就是 Flink SQL 回撤机制与问题讲的回撤。

复杂事件处理：MATCH_RECOGNIZE

有些监控关心的不是某个数值，而是一串事件的形态，比如「某个指标连续下降，然后反弹」。Flink SQL 支持 SQL 标准中的 MATCH_RECOGNIZE 子句，可以像写正则表达式一样描述事件序列。

以「key1 先连续下降、再回升」为例：

时间	15:40	15:41	15:42	15:43	15:44
key1	5	4	1	5	6
角色	起点	下降	下降（谷底）	回升	—

sql

SELECT *
FROM dwd_log
    MATCH_RECOGNIZE (
        PARTITION BY key_word      -- 每个关键字单独匹配
        ORDER BY rowtime           -- 按事件时间排序
        MEASURES
            START_ROW.rowtime AS start_ts,
            LAST(VAL_DOWN.rowtime) AS bottom_ts,
            LAST(VAL_UP.rowtime) AS end_ts
        ONE ROW PER MATCH
        AFTER MATCH SKIP TO LAST VAL_UP
        -- 模式：一个起点，接一次或多次下降，再接一次上升
        PATTERN (START_ROW VAL_DOWN+ VAL_UP)
        DEFINE
            VAL_DOWN AS
                (LAST(VAL_DOWN.key1, 1) IS NULL AND VAL_DOWN.key1 < START_ROW.key1) OR
                    VAL_DOWN.key1 < LAST(VAL_DOWN.key1, 1),
            VAL_UP AS
                VAL_UP.key1 > LAST(VAL_DOWN.key1, 1)
    ) MR;

PATTERN 里的 VAL_DOWN+ 和正则中的 + 一样表示「一次或多次」，DEFINE 则定义每个变量需要满足的条件。上面这组数据会匹配出一行结果：从 15:40 开始，15:42 到达谷底，15:43 回升。

shell

tutorials/run.sh --batch sql_is_all_your_need/flink_sql/cep.sql

text

+----+----------+-------------------------+-------------------------+-------------------------+
| op | key_word |                start_ts |               bottom_ts |                  end_ts |
+----+----------+-------------------------+-------------------------+-------------------------+
| +I |   stdout | 2023-04-09 15:40:05.000 | 2023-04-09 15:42:05.000 | 2023-04-09 15:43:05.000 |
+----+----------+-------------------------+-------------------------+-------------------------+

模式匹配的语法需要专门学习，SQL 也因此变得更复杂。但这也说明，随着 SQL 标准的发展，SQL 的表达能力在不断增强。

一个冷知识

SQL 是图灵完备的。

带递归公用表表达式（WITH RECURSIVE）的 SQL 理论上可以表达任何计算。当然，「SQL Is All Your Need」只是一句口号，没有人会只用 SQL 去完成复杂灵活的任务，但单纯从理论上说，这句话也不算错。

总结

这篇文章用一套技术栈走完了清洗、分析、修数、监控和模式匹配。示例虽然简单，但能看出大数据技术正在走向融合：存储抽象成表，计算统一成 SQL，流和批只是同一份逻辑的两种运行方式。

试试看

改 prepare.sql 里的日志，比如把最后一条的时间改成 15:50:05，再用流模式跑分析：水位线推进到 15:49:55，哪些窗口会补出来？
打开 analysis.sql 里被注释掉的 table.optimizer.distinct-agg.split.enabled，把 INSERT INTO dws_metric 改成 EXPLAIN INSERT INTO dws_metric 再用流模式运行：计划里的去重被拆成了两层，先按 MOD(HASH_CODE(key2), 1024) 分桶去重，再把各桶的计数加起来。为什么能这样拆，见少算一点：大数据计算优化的统一视角。

本文的全部 SQL 在 QFlink，运行说明见 QFlink 教程第 01 章。

参考资料

全文评论

正在加载评论…