前四天走完了 Flink 的入门 → DataStream API → 窗口原理 → Flink SQL 窗口与 JOIN。本篇聚焦"Day4 没覆盖的生产级能力"：Flink SQL 命令行与执行模式、各类连接器（CDC/Doris/文件监控）、SQL Hint 与 DISTINCT 优化、Window Join、CEP 模式检测、Checkpoint 故障恢复、PyFlink 自定义函数。内容定位在"把 SQL 从写 demo 用到生产环境"。
一、SQL 命令行与执行模式
1.1 启动 SQL Client
有集群时先起 Yarn Session（Day1 的 Standalone 本地集群也可以，生产推荐 Yarn）：
# 启动 Yarn 会话集群（-d 后台运行，-tm 4G 指定 TaskManager 内存）
yarn-session.sh -tm 4G -d
# 进入 SQL Client
sql-client.sh
Client 启动后，先设置输出模式（表格模式便于查看实时流更新）：
SET 'sql-client.execution.result-mode' = 'table';
-- table 和 tableau 两种都可以，table 更直观
1.2 批模式 vs 流模式
Flink SQL 是批流一体的，同一套 SQL 可以切换两种执行模式：
-- 流处理模式（默认）
SET 'execution.runtime-mode' = 'streaming';
-- 批处理模式
SET 'execution.runtime-mode' = 'batch';
| 模式 | 适用数据 | 执行方式 | 输出特点 |
|---|---|---|---|
| streaming（流） | 有界/无界流都可 | 持续运行、持续输出 | 输出 INSERT/UPDATE/DELETE 三种变更 |
| batch（批） | 仅无界流不行 | 跑完退出 | 输出最终结果，只有 INSERT |
关键区别：流模式下无界表的聚合状态会无限增长，必须用窗口或加状态 TTL；批模式下程序跑完自动释放所有状态。
1.3 一个例子看清差异
用 filesystem connector 读 HDFS 上的 CSV 文件（有界流）：
CREATE TABLE students (
id STRING, name STRING, age INT, sex STRING, clazz STRING
) WITH (
'connector' = 'filesystem',
'path' = 'hdfs://master:9000/data/students',
'format' = 'csv',
'csv.field-delimiter' = ','
);
批模式执行聚合：
SET 'execution.runtime-mode' = 'batch';
INSERT INTO clazz_num
SELECT clazz, COUNT(1) AS num FROM students GROUP BY clazz;
-- 执行一次，输出最终结果后作业退出
流模式执行聚合：
SET 'execution.runtime-mode' = 'streaming';
INSERT INTO clazz_num
SELECT clazz, COUNT(1) AS num FROM students GROUP BY clazz;
-- 作业持续运行，每来一条新数据就输出新结果（包含 UPDATE）
流模式处理有界流的特点：作业不会自动结束，你需要手动 cancel（./bin/flink cancel <job-id>）。
二、连接器实战
Flink SQL 通过 connector 参数连接各种外部系统，这是流处理"数据进得来、结果出得去"的基础。
2.1 测试连接器 datagen
快速生成随机数据做功能验证，不需要依赖 Kafka。无界流，每秒生成指定条数。
CREATE TABLE students_datagen (
id STRING,
name STRING,
age INT,
sex STRING,
clazz STRING
) WITH (
'connector' = 'datagen',
'rows-per-second' = '5', -- 每秒 5 条
'fields.id.length' = '10', -- 字符串长度
'fields.name.length' = '2',
'fields.sex.length' = '1',
'fields.clazz.length' = '4',
'fields.age.min' = '1', -- 数值范围
'fields.age.max' = '100'
);
SELECT * FROM students_datagen;
实战用途：本地写 SQL 逻辑时先用 datagen 验证，不用起 Kafka。确认逻辑对了再换 Kafka。
2.2 filesystem 连接器
连接 HDFS/S3/本地文件系统。有界流 vs 无界流的关键差异靠 source.monitor-interval：
-- 有界流：一次性读取目录下现有文件，读完退出
CREATE TABLE students_batch (...) WITH (
'connector' = 'filesystem',
'path' = 'hdfs://master:9000/data/students',
'format' = 'csv'
);
-- 无界流：每隔 5 秒扫描目录，有新文件就读进来
CREATE TABLE students_stream (...) WITH (
'connector' = 'filesystem',
'path' = 'hdfs://master:9000/data/students',
'format' = 'csv',
'source.monitor-interval' = '5000'
);
流模式写入 HDFS 的坑：GROUP BY 产生的是更新结果流（同一个 key 会多次输出 UPDATE），直接写入 filesystem 会每行一条，文件无限增长。解决方案：
-- 方案 1：用 canal-json format 写变更流（每行带 INSERT/UPDATE/DELETE 类型）
CREATE TABLE sink_canal (...) WITH (
'connector' = 'filesystem',
'path' = '...',
'format' = 'canal-json' -- 变更流格式
);
-- 方案 2：只写 INSERT 流（过滤掉 UPDATE/DELETE）
INSERT INTO sink_json
SELECT * FROM (...) WHERE op = 'INSERT';
2.3 mysql-cdc 连接器（增量同步）
Day4 的 Regular Join/Lookup Join 都是读 JDBC（有界流，只读一次）。如果要实时同步 MySQL 的增删改，用 mysql-cdc——基于 Debezium，先全表扫描，再监控 Binlog。
前置要求（MySQL 端）：
- Binlog 格式必须是 ROW：
SET GLOBAL binlog_format = 'ROW'; - 账号需要
REPLICATION SLAVE, REPLICATION CLIENT, SELECT权限
CREATE TABLE clazz_num_cdc (
clazz STRING,
num BIGINT,
PRIMARY KEY (clazz) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'master',
'port' = '3306',
'username' = 'root',
'password' = '123456',
'database-name' = 'bigdata',
'table-name' = 'clazz_num'
);
SELECT * FROM clazz_num_cdc;
CDC 输出的流包含三种变更：
+IINSERT：新增行-U/+UUPDATE：旧行 + 新行-DDELETE：删除行
把这个流接到其他 Sink（Doris、Kafka）就能实现实时数据同步管道。
2.4 Doris 连接器
Doris 是 OLAP 型数据库，列式存储、向量化执行，实时数仓的典型存储。写入 Doris 有几个避坑点：
CREATE TABLE students_doris (
id STRING, name STRING, age INT, sex STRING, clazz STRING,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector' = 'doris',
'fenodes' = 'master:8034', -- Doris FE HTTP 端口
'benodes' = 'master:8041', -- Doris BE brpc 端口（可选）
'table.identifier' = 'bigdata.students',
'username' = 'root',
'password' = '123456',
'sink.label-prefix' = 'doris_label' -- 幂等写入标签前缀
);
INSERT INTO students_doris SELECT * FROM students_batch;
版本冲突坑：Flink Doris Connector 版本和 Flink/Doris 版本必须匹配。典型错误：
NoClassDefFoundError: org/apache/doris/flink/DorisExecutionEnv
对策：下载对应版本的 flink-doris-connector-<flink-version>-<doris-version>.jar，替换 Flink 的 lib/ 目录里的旧包后重启集群。
Doris 侧建表（用 UNIQUE KEY 模型，支持主键去重）：
CREATE TABLE IF NOT EXISTS students (
id VARCHAR(20), name STRING, age INT, sex STRING, clazz STRING
)
UNIQUE KEY(id)
DISTRIBUTED BY HASH(id) BUCKETS 10
PROPERTIES ("replication_num" = "1");
2.5 print / blackhole（测试 Sink）
print：把结果打印到 TaskManager 日志里，快速验证；blackhole：什么都不做，纯测试吞吐（压测用）。
-- print
CREATE TABLE print_table (...) WITH ('connector' = 'print');
INSERT INTO print_table SELECT * FROM students_datagen;
-- blackhole：把并发提升到 50000 测试极限吞吐
INSERT INTO blackhole_table
SELECT * FROM students_datagen /*+ OPTIONS('rows-per-second'='50000') */;
三、并行度与 SQL Hint
3.1 设置并行度
并行度 = 每个 TaskManager 的 Slot 数 × TaskManager 数。Flink 没显式设置时默认为 1：
SET 'parallelism.default' = '2';
每个算子独立并行度的设置（Hint 方式）：
-- 只给本次查询设置 Source 并行度
SELECT * FROM cars /*+ OPTIONS('scan.parallelism'='4') */;
3.2 SQL Hint 实战
Hint 可以在不重新建表的情况下，临时修改连接器参数，常用于测试不同的消费起点或并行度：
-- 临时让这个查询从 earliest-offset 开始消费
SELECT * FROM cars /*+ OPTIONS('scan.startup.mode'='earliest-offset') */;
Hint 不影响表定义，只在本次查询里生效。常用 Hint：
| Hint 参数 | 作用 |
|---|---|
scan.startup.mode | 修改 Kafka 消费起点 |
scan.startup.timestamp-millis | 从指定时间戳开始 |
scan.parallelism | Source 并行度 |
四、DISTINCT 与 ROW_NUMBER 去重
4.1 DISTINCT 的问题
流模式下直接用 DISTINCT 是全量去重，所有历史数据都要保存在状态里：
-- 状态无限增长，跑久了 OOM
SELECT DISTINCT * FROM students_datagen;
4.2 用 ROW_NUMBER 替代 DISTINCT
原理：按去重字段分区，取每个分区的第一条，状态只需要保存去重字段的主键值，内存占用大幅降低：
INSERT INTO blackhole_table
SELECT id, name, age, sex, clazz FROM (
SELECT *,
ROW_NUMBER() OVER (PARTITION BY id ORDER BY PROCTIME()) AS r
FROM students_datagen /*+ OPTIONS('rows-per-second'='50000') */
) AS a
WHERE r = 1;
对比：
| 方案 | 状态大小 | 说明 |
|---|---|---|
DISTINCT * | 存所有已出现过的完整行 | 数据量指数增长 |
ROW_NUMBER ... WHERE r = 1 | 只存已出现过的 id | O(N)，可控 |
五、Window Join（窗口内关联）
Day4 讲了 Regular / Interval / Temporal / Lookup 四种 JOIN，漏掉了Window Join——把两个流先套同一个窗口，再在窗口内关联。
5.1 语法
SELECT a.id, a.name, b.cid, b.score
FROM (
SELECT * FROM TABLE(
TUMBLE(TABLE a_stream, DESCRIPTOR(eventtime), INTERVAL '1' MINUTES)
)
) AS a
JOIN (
SELECT * FROM TABLE(
TUMBLE(TABLE b_stream, DESCRIPTOR(eventtime), INTERVAL '1' MINUTES)
)
) AS b
ON a.id = b.sid
AND a.window_start = b.window_start
AND a.window_end = b.window_end;
5.2 为什么状态有界
两个流各自开窗 → 窗口结束就输出 → 关联时只需要处理同一窗口时间范围内的数据，窗口关了数据就清空。状态大小 = 窗口大小内的数据量 × 并行度，有界。
5.3 实战：学生信息 + 成绩（按分钟关联）
CREATE TABLE students_kafka_eventtime (
id STRING, name STRING, age INT, sex STRING, clazz STRING,
eventtime TIMESTAMP(3),
WATERMARK FOR eventtime AS eventtime
) WITH (
'connector' = 'kafka', 'topic' = 'students',
'properties.bootstrap.servers' = 'master:9092',
'properties.group.id' = 'testGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'csv', 'csv.ignore-parse-errors' = 'true'
);
-- 输入：id,name,age,sex,clazz,eventtime
-- 例：1500100001,施笑槐,22,女,文科六班,2026-10-10 10:10:04
CREATE TABLE scores_kafka_eventtime (
sid STRING, cid STRING, score DOUBLE,
eventtime TIMESTAMP(3),
WATERMARK FOR eventtime AS eventtime
) WITH (
'connector' = 'kafka', 'topic' = 'scores',
'properties.bootstrap.servers' = 'master:9092',
'properties.group.id' = 'testGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'csv', 'csv.ignore-parse-errors' = 'true'
);
-- 输入：sid,cid,score,eventtime
-- 例：1500100001,1000001,98,2026-10-10 10:10:04
-- 1分钟窗口内关联，状态只有1分钟
SELECT a.id, a.name, b.cid, b.score, a.window_start, a.window_end
FROM (
SELECT * FROM TABLE(
TUMBLE(TABLE students_kafka_eventtime, DESCRIPTOR(eventtime), INTERVAL '1' MINUTES)
)
) a
JOIN (
SELECT * FROM TABLE(
TUMBLE(TABLE scores_kafka_eventtime, DESCRIPTOR(eventtime), INTERVAL '1' MINUTES)
)
) b
ON a.id = b.sid
AND a.window_start = b.window_start
AND a.window_end = b.window_end;
5.4 五种 JOIN 完整对比（Day4 + Day5）
| JOIN 类型 | 状态 | 典型场景 |
|---|---|---|
| Regular Join | 无限（危险） | 有界流全量关联 |
| Interval Join | 有界（时间区间） | 双流时间相近关联 |
| Temporal Join | 小（版本表常驻） | 关联"当时的版本" |
| Lookup Join | 极小（实时查库） | 流关联外部维表 |
| Window Join | 有界（窗口大小） | 同一时间窗口内的双流关联 |
六、CEP 模式检测（MATCH_RECOGNIZE）
Day4 没覆盖的全新能力：用 SQL 在流里检测复杂事件模式，典型场景——反欺诈、告警、趋势检测。
6.1 语法骨架
SELECT * FROM 表名
MATCH_RECOGNIZE (
PARTITION BY 分组键 -- 按谁分组（如账户 ID）
ORDER BY 时间列 -- 必须按时间升序
MEASURES -- 匹配成功后输出哪些字段
A.xxx AS a_xxx,
B.yyy AS b_yyy
PATTERN (规则) -- 模式序列，如 (A B)、(A{3} B)、(A B+ C)
DEFINE -- 定义每个模式变量的条件
A AS 条件1,
B AS 条件2
) AS T;
6.2 案例 1：交易欺诈检测
业务规则：同一个账户，如果出现一笔小于 $1 的交易，紧接着出现一笔大于 $500 的交易，判定为可疑。
先建表：
CREATE TABLE transactions (
id STRING,
price DOUBLE,
eventtime TIMESTAMP(3),
WATERMARK FOR eventtime AS eventtime
) WITH (
'connector' = 'kafka', 'topic' = 'transactions',
'properties.bootstrap.servers' = 'master:9092',
'properties.group.id' = 'testGroup',
'scan.startup.mode' = 'latest-offset',
'format' = 'csv', 'csv.ignore-parse-errors' = 'true'
);
往 Kafka 输入数据：
001,100,2026-11-09 10:10:10 -- 正常
001,0.5,2026-11-09 10:10:11 -- A 条件触发（< $1）
001,600,2026-11-09 10:10:13 -- B 条件触发（> $500），匹配成功！
001,400,2026-11-09 10:10:20 -- 没有 < $1 的前置，不匹配
001,0.4,2026-11-09 10:10:21 -- 又一个 A
001,100,2026-11-09 10:10:22 -- 金额不够大，不匹配
001,1000,2026-11-09 10:10:24 -- B 条件触发，匹配成功！
用 LAG 也能做，但 CASE WHEN 复杂，而且要手工排 state。CEP 写法更清晰：
SELECT * FROM transactions
MATCH_RECOGNIZE (
PARTITION BY id
ORDER BY eventtime
MEASURES
A.price AS a_price, -- 匹配到的 A 的价格
B.price AS b_price, -- 匹配到的 B 的价格
A.eventtime AS a_time,
B.eventtime AS b_time
PATTERN (A B)
DEFINE
A AS price < 1,
B AS price > 500
) AS T;
输出：每次匹配成功输出一行，包含 A 和 B 两个事件的字段。
6.3 案例 2：连续 N 次小额后大额 + 时间限制
规则升级：连续 3 次 < $1 的交易 + 紧跟着一笔 > $500 + 必须在 5 秒内完成。
SELECT * FROM transactions
MATCH_RECOGNIZE (
PARTITION BY id
ORDER BY eventtime
MEASURES
FIRST(A.price) AS first_a_price,
AVG(A.price) AS avg_a_price,
SUM(A.price) AS sum_a_price,
B.price AS b_price,
A.eventtime AS a_time,
B.eventtime AS b_time
PATTERN (A{3} B) WITHIN INTERVAL '5' SECOND -- A 连续 3 次，且整体必须在 5 秒内
DEFINE
A AS price < 1,
B AS price > 500
) AS T;
PATTERN 语法说明：
| 写法 | 含义 |
|---|---|
(A B) | A 后面紧跟 B |
(A{3} B) | A 连续出现 3 次后面跟 B |
(A B+ C) | A → B（1 次或多次）→ C |
... WITHIN INTERVAL '5' SECOND | 整个模式匹配必须在 5 秒内完成 |
MEASURES 里可用的聚合函数：FIRST()、LAST()、AVG()、SUM()、MAX()、MIN()，都是针对匹配到的该模式变量的所有事件计算。
6.4 案例 3：股票价格持续下跌期
找出单只股票价格连续下跌（中间允许小幅反弹一次）的时期。
CREATE TABLE ticker (
symbol STRING,
price DOUBLE,
tax DOUBLE,
rowtime TIMESTAMP(3),
WATERMARK FOR rowtime AS rowtime
) WITH (...);
输入数据：
ACME,12,1,2026-11-09 10:00:00
ACME,17,2,2026-11-09 10:00:01 -- 下跌
ACME,19,1,2026-11-09 10:00:02 -- 下跌
ACME,21,3,2026-11-09 10:00:03 -- 下跌
ACME,25,2,2026-11-09 10:00:04 -- 上涨，C 触发
ACME,18,1,2026-11-09 10:00:05
...
SELECT * FROM ticker
MATCH_RECOGNIZE (
PARTITION BY symbol
ORDER BY rowtime
MEASURES
A.price AS a_price,
B.price AS b_price,
C.price AS c_price,
A.rowtime AS a_time,
C.rowtime AS c_time
AFTER MATCH SKIP PAST LAST ROW -- 匹配完成后，从最后匹配行的下一行继续
PATTERN (A B+ C) -- A → B（下跌，1次或多次）→ C（反弹）
DEFINE
B AS (LAST(price, 1) IS NULL AND B.price < A.price)
OR B.price < LAST(price, 1), -- 比上一行低
C AS C.price > B.price -- 比上一行高，反弹
) AS T;
AFTER MATCH SKIP PAST LAST ROW：匹配到一次后，从模式的最后一个匹配行的下一行重新开始。不设置的话默认会从下一行开始，具体行为看 Flink 版本。
6.5 CEP vs LAG 的取舍
| 维度 | LAG | CEP (MATCH_RECOGNIZE) |
|---|---|---|
| 模式复杂度 | 只能写简单相邻行逻辑 | 支持序列、数量、时间窗口 |
| 可读性 | CASE WHEN 嵌套多了就乱 | PATTERN 规则清晰 |
| 灵活性 | 需要预定义所有中间条件 | 聚合 FIRST/LAST 很方便 |
| 代价 | 每次新事件都要更新所有聚合 | 只在匹配成功时输出一次 |
什么时候用哪个：简单相邻行逻辑（如"上一次价格 - 当前价格"）用 LAG；复杂序列（如"连续 N 次 A 后接 B"、“趋势检测”）用 CEP。
七、Checkpoint 与故障恢复
Day3 讲过 Checkpoint 概念，这里讲SQL 层面怎么配置和怎么恢复。
7.1 flink-conf.yaml 配置
# 开启 checkpoint，每 10 秒一次
execution.checkpointing.interval: 10000
# 作业取消后是否保留 checkpoint（恢复时要用）
execution.checkpointing.externalized-checkpoint-retention: RETAIN_ON_CANCELLATION
# 同时进行的 checkpoint 数量（一般保持 1，避免互相影响）
execution.checkpointing.max-concurrent-checkpoints: 1
# 两个 checkpoint 之间的最小间隔（避免堆积）
execution.checkpointing.min-pause: 0
# 精确一次语义（生产必须 EXACTLY_ONCE，性能差一点）
execution.checkpointing.mode: EXACTLY_ONCE
# 超时时间（默认 10 分钟，作业大时要调大）
execution.checkpointing.timeout: 10min
# 允许连续失败几次 checkpoint 后作业挂
execution.checkpointing.tolerable-failed-checkpoints: 0
# 状态后端（hashmap 内存、rocksdb 磁盘）
state.backend: hashmap
# Checkpoint 持久化路径（HDFS/S3/OSS）
state.checkpoints.dir: hdfs://master:9000/flink/checkpoints
state.savepoints.dir: hdfs://master:9000/flink/savepoints
7.2 SQL 文件提交
编写 word_count.sql：
CREATE TABLE words (word STRING) WITH (...);
CREATE TABLE print_table (
word STRING, num BIGINT
) WITH ('connector' = 'print');
INSERT INTO print_table
SELECT word, COUNT(1) AS num FROM words GROUP BY word;
首次提交（会从头消费）：
sql-client.sh -f word_count.sql
7.3 故障恢复
作业中途挂了（比如重启 Yarn、kill 进程），怎么恢复？
- 找到 checkpoint 目录：HDFS 上
state.checkpoints.dir下有每个作业的随机 ID 目录，里面有chk-N子目录：
hdfs dfs -ls /flink/checkpoints/
# c56e7c3dcf8352a285dfd954084c6258/
hdfs dfs -ls /flink/checkpoints/c56e7c3dcf8352a285dfd954084c6258
# chk-23/
- SQL 里加恢复路径（放在 INSERT 语句前面）：
SET 'execution.savepoint.path' =
'hdfs://master:9000/flink/checkpoints/c56e7c3dcf8352a285dfd954084c6258/chk-23';
- 重新提交：
sql-client.sh -f word_count.sql
# 作业从 chk-23 恢复，数据不丢不重
7.4 EXACTLY-ONCE vs AT-LEAST-ONCE
| 模式 | 语义 | 代价 |
|---|---|---|
| EXACTLY_ONCE | 数据恰好一次，不丢不重 | Checkpoint 过程中会阻塞数据流（对齐栅栏） |
| AT_LEAST_ONCE | 数据至少一次，可能重复 | 不阻塞，吞吐高，但下游要能容忍重复 |
生产推荐 EXACTLY_ONCE，除非对吞吐量要求极高且下游天然幂等。
八、PyFlink 自定义函数
Flink SQL 内置函数不够时，可以用 Python 写 UDF。
8.1 编写 Python UDF
# upper.py
from pyflink.table.udf import udf
@udf(result_type='STRING')
def upper(s: str):
"""把字符串转大写"""
return s.upper()
8.2 启动 SQL Client 加载 Python
# --pyFiles 加载 Python 文件，--pyExecutable 指定 Python 解释器
sql-client.sh \
--pyFiles upper.py \
--pyExecutable /opt/miniconda3/envs/flink_env/bin/python3
8.3 注册与调用
CREATE FUNCTION python_upper AS 'upper.upper' LANGUAGE PYTHON;
SELECT python_upper('shujia'); -- 输出 SHUJIA
SELECT python_upper(name) FROM students_batch; -- 批量转大写
注意：upper 是文件名，upper() 是函数名，完整限定为 upper.upper。
8.4 PyFlink 环境要求
- Python 版本 >= 3.8；
- 安装 pyflink：
pip install apache-flink==1.18.0（和 Flink 版本对齐）； - 集群上所有 TaskManager 都能访问到这个 Python 解释器路径。
九、典型坑点
坑点 1：filesystem 连接器扫不到新文件
现象：上传文件到目录，SQL 没有输出。
原因：忘了加 'source.monitor-interval'，默认一次性读完就结束了。
对策：加监控间隔（毫秒）：'source.monitor-interval' = '5000'。
坑点 2：mysql-cdc 全量扫描锁大表
现象：启动 CDC 任务后业务查询被阻塞。
原因：默认全量扫描用 LOCK TABLES 会对表加读锁。
对策：MySQL 开 gtid_mode=ON，CDC 配置 'scan.incremental.snapshot.enabled' = 'true'（增量快照模式，不锁表）。
坑点 3：Doris 写入报版本不兼容
现象：NoClassDefFoundError 或 NoSuchMethodError。
对策：连接器版本必须匹配 Flink 和 Doris，替换 lib/ 下的旧 jar，重启集群（sql-client 里热加载没用）。
坑点 4：DISTINCT 后作业越来越慢
原因：全量去重把所有历史数据都存在状态里。
对策：用 ROW_NUMBER + PARTITION BY 替代，只存去重主键。
坑点 5：Window Join 漏数据
原因：两个流各自有水位线，但 window_join 取的是各自窗口内的数据。如果某个流数据少导致窗口迟迟不触发，另一流的数据已经被清理了。
对策：保证两个流水位线都在推进（空闲分区用 withIdleness），或放宽 WHERE 里的时间条件到 INTERVAL JOIN。
坑点 6：CEP 匹配后结果丢了
原因：AFTER MATCH SKIP 控制不正确，导致同一个事件被多次匹配或跳过。
对策：测试时先去掉 AFTER MATCH，用默认行为验证结果，再加。
坑点 7：Checkpoint 太大导致超时
原因：窗口状态、CDC 状态没有配置好 TTL，状态无限增长。
对策：SET 'table.exec.state.ttl' = '1h'；或用 RocksDB 状态后端（磁盘+内存，比 hashmap 省内存）。
坑点 8：保存点路径写错导致恢复失败
现象：execution.savepoint.path 设置了路径但作业还是从头消费。
原因：路径里的 chk-23 目录不存在，或路径格式错误。
对策：先 hdfs dfs -ls 确认目录存在，路径开头必须是 hdfs://（不是 hdfs:/）。
坑点 9：PyFlink 加载 UDF 报 ModuleNotFoundError
现象：No module named 'upper'。
原因：--pyFiles 路径没找到文件，或 Python 解释器版本不对。
对策：用绝对路径，比如 --pyFiles /home/flink/upper.py；确认 python3 -c "import upper" 在集群节点上能跑通。
坑点 10：Hint 改不了 connector 参数
原因：Hint 只能改可变参数（并行度、消费起点），connector URL 等必选参数不能被 Hint 覆盖。
对策：建表时就要确定好必选参数，Hint 只用于可调项。
十、五天知识串联
| 天 | 主题 | 核心关键词 |
|---|---|---|
| Day1 | 入门与环境 | 概念、架构、WordCount、提交作业 |
| Day2 | DataStream API | Source / map / keyBy / sum / Sink / 并行度 |
| Day3 | 窗口原理 | 事件时间、水位线、四类窗口、允许迟到 |
| Day4 | Flink SQL 窗口与 JOIN | TVF、事件时间 DDL、Temporal/Lookup/Interval Join |
| Day5 | 生产实战 | connector、datagen、mysql-cdc、doris、Hint、DISTINCT、Window Join、CEP、Checkpoint、PyFlink UDF |
Day4 vs Day5 的分工（刻意避重）：
| 能力 | Day4 深度 | Day5 深度 |
|---|---|---|
| 窗口 TVF（TUMBLE/HOP/CUMULATE/SESSION） | ✅ 原理 + 实战 | 跳过，直接用 |
| 事件时间 DDL | ✅ 完整 | 跳过 |
| ORDER BY / OVER | ✅ 完整 | 跳过 |
| Regular/Interval/Temporal/Lookup Join | ✅ 完整 | 跳过，新增 Window Join |
| CEP (MATCH_RECOGNIZE) | ❌ 没讲 | ✅ 完整（三个案例） |
| Checkpoint 配置与恢复 | Day3 提概念 | ✅ SQL 层面完整 |
| PyFlink UDF | ❌ 没讲 | ✅ 完整 |
| 连接器（datagen/mysql-cdc/doris） | ❌ 没讲 | ✅ 完整 |
| DISTINCT 优化 | ❌ 没讲 | ✅ 完整 |
Day6 可以继续学：状态管理（TTL、MapState/ValueState）、RocksDB 状态后端调优、Flink CDC 全链路实时数仓、CEP 复杂模式高级语法、Flink on K8s 部署、Table API 和 SQL 互操作。
转载自 CSDN-专业IT技术社区
原文链接：https://blog.csdn.net/YEEMOLY/article/details/170556365
