本章目标：能写出一个带事件时间、窗口、状态和 Checkpoint 的 Flink 作业骨架，并说清上线、重启、监控和去重怎么做。 前置知识：Flink 基础、Flink 时间、窗口与状态 预计用时：150 分钟
会解释水位线，不等于能交付作业。真实作业要处理：消息重复、迟到数据、状态越来越大、Checkpoint 失败、下游写入不幂等、上线后怎么改并行度。本章按一条实时 GMV 链路，把这些坑逐个拆开。
1. 一条实时作业由什么组成
Kafka(orders)
│
▼
Flink Source ──► 过滤/去重/窗口聚合 ──► Sink(ClickHouse / Kafka / JDBC)
│ │
│ └── 状态：窗口、去重键、维表缓存
└── Checkpoint：把状态定期存到 HDFS/S3/本地
一个能上线的作业至少要回答六个问题：
| 问题 | 本章对应做法 |
|---|---|
| 数据从哪来、到哪去 | Kafka source / ClickHouse sink |
| 按什么时间算 | 事件时间 + 水位线 |
| 状态存什么、存多久 | 窗口状态、去重状态设 TTL |
| 失败怎么恢复 | Checkpoint + 重启策略 |
| 重复消息怎么办 | 主键去重 + 幂等 sink |
| 怎么知道它健康 | 延迟、反压、Checkpoint 指标 |
2. Flink SQL：一条完整的分钟级 GMV 作业
下面是一个可落地的 Flink SQL 骨架。DDL 里的连接信息用变量表示，实际提交时替换。
2.1 定义 Kafka 源表
CREATE TABLE kafka_orders (
order_id STRING,
city STRING,
status STRING,
amount DECIMAL(12, 2),
event_time TIMESTAMP(3),
WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'orders',
'properties.bootstrap.servers' = 'kafka:9092',
'properties.group.id' = 'gmv-job-v1',
'scan.startup.mode' = 'group-offsets',
'format' = 'json',
'json.ignore-parse-errors' = 'true'
);
WATERMARK FOR ... AS event_time - INTERVAL '5' SECOND 表示：事件时间落后当前最大事件时间 5 秒以内，仍然可能到达。水位线是“我认为这个时间点之前的数据基本到齐了”的声明，不是精确时钟。
2.2 定义结果表
CREATE TABLE ads_gmv_minute (
window_start TIMESTAMP(3),
window_end TIMESTAMP(3),
city STRING,
gmv DECIMAL(18, 2),
order_cnt BIGINT,
PRIMARY KEY (window_start, city) NOT ENFORCED
) WITH (
'connector' = 'jdbc',
'url' = 'jdbc:clickhouse://clickhouse:8123/default',
'table-name' = 'ads_gmv_minute',
'username' = 'writer',
'password' = '${CLICKHOUSE_PASSWORD}',
'sink.buffer-flush.max-rows' = '1000',
'sink.buffer-flush.interval' = '2s'
);
PRIMARY KEY 配合 ClickHouse 的 ReplacingMergeTree（或支持 upsert 的表引擎）才能让重复写入收敛。只声明主键、不选对表引擎，重复数据仍然会留在表里。
2.3 窗口聚合
INSERT INTO ads_gmv_minute
SELECT
window_start,
window_end,
city,
SUM(amount) AS gmv,
COUNT(DISTINCT order_id) AS order_cnt
FROM TABLE(
TUMBLE(TABLE kafka_orders, DESCRIPTOR(event_time), INTERVAL '1' MINUTE)
)
WHERE status NOT IN ('canceled', 'unavailable')
GROUP BY window_start, window_end, city;
这段 SQL 做了四件事：按事件时间切 1 分钟滚动窗口、过滤无效订单、按城市聚合、写出窗口结果。窗口结束且水位线越过窗口末尾时，结果才会发出。
注意：ClickHouse JDBC 不是 Flink 官方连接器，生产中通常用 Kafka 中转或专门的 connector。这里只展示 sink 的配置形态，具体依赖以团队环境为准。
3. 动手：用 Python 模拟事件时间窗口
没有 Flink 集群也能验证窗口逻辑。下面的脚本用标准库模拟“事件时间 + 水位线 + 迟到数据”：
from collections import defaultdict
WINDOW_MS = 5000
DELAY_MS = 2000
STREAM = [
(1000, "北京", 120),
(1500, "上海", 80),
(2600, "上海", 95),
(4000, "北京", 60),
(9000, "北京", 30),
(4500, "上海", 50),
]
buckets = defaultdict(int)
late = []
max_seen = 0
for ts, city, amount in STREAM:
max_seen = max(max_seen, ts)
watermark = max_seen - DELAY_MS
start = (ts // WINDOW_MS) * WINDOW_MS
end = start + WINDOW_MS
if end <= watermark:
late.append((ts, city, amount))
continue
buckets[(start, end, city)] += amount
for (start, end, city) in sorted(buckets):
print(f"[{start}, {end}) {city} = {buckets[(start, end, city)]}")
print("late:", late)
输出：
[0, 5000) 上海 = 175
[0, 5000) 北京 = 180
[5000, 10000) 北京 = 30
late: [(4500, '上海', 50)]
事件 4500 到达时，最大事件时间已经是 9000，水位线是 7000，<input type="checkbox" class="task sql-task" data-k="sql:lc:1141" data-label="0, 5000) 窗口已经关闭，所以它被判定为迟到。真实 Flink 里，迟到数据可以丢弃、更新结果，或者通过侧输出流单独收集。
4. 状态、Checkpoint 与 Exactly-Once
4.1 状态是什么
窗口里暂存的聚合值、去重用的订单 ID、维表缓存，都是状态。状态默认在内存里，作业失败就丢；所以要定期做 Checkpoint。
关键配置：
execution.checkpointing.interval: 60s
execution.checkpointing.timeout: 10min
execution.checkpointing.mode: EXACTLY_ONCE
state.backend: rocksdb
state.checkpoints.dir: hdfs://namenode:8020/flink/checkpoints
state.savepoints.dir: hdfs://namenode:8020/flink/savepoints
Checkpoint 是作业运行中自动做的快照，用于失败恢复；Savepoint 是手动触发的快照，用于升级、迁移、改并行度。改代码前先打 Savepoint，是上线的基本纪律。
4.2 Exactly-Once 不是免费的
Flink 内部可以做到 Exactly-Once，但端到端还要看 sink：
Kafka sink 要开启事务或幂等写入；数据库 sink 用主键 upsert 或两阶段提交；文件 sink 先写临时文件，Checkpoint 完成后再原子重命名。
如果 sink 只是普通 INSERT，重复消费就会写出重复行，Flink 内部的状态一致性救不了你。
4.3 去重
订单消息可能重复。常见做法是按主键保留第一条：
SELECT order_id, city, status, amount, event_time
FROM (
SELECT *,
ROW_NUMBER() OVER (
PARTITION BY order_id ORDER BY event_time ASC
) AS rn
FROM kafka_orders
)
WHERE rn = 1;
如果订单会更新，就不能只保留第一条，应该用 upsert-kafka 或按版本号取最新。去重策略取决于业务语义：支付事件只认第一次，订单状态要认最新一次。
4.4 状态 TTL
去重状态如果不清理，会无限增长。给状态设置 TTL：
SET 'table.exec.state.ttl' = '24h';
TTL 太长会占内存，太短会让迟到的重复消息重新出现。按业务允许的最大重复窗口来定。
5. 维表 Join、反压与监控
5.1 维表 Join
实时流要补商品品类、城市名称，常用 lookup join：
SELECT o.order_id, o.city, p.category, o.amount
FROM kafka_orders AS o
LEFT JOIN dim_product FOR SYSTEM_TIME AS OF o.proc_time AS p
ON o.product_id = p.product_id;
维表数据变化后，Flink 需要重新查询或刷新缓存。缓存 TTL 越大，性能越好，但维度更新越慢。要接受“实时链路里的维表不是强一致”的事实。
5.2 反压
反压（backpressure）表示下游处理不过来，压力向上游传导。排查顺序：
- 看 Flink UI 的反压指标，定位卡在哪个算子。
- 看 sink 写入耗时、数据库连接池、外部服务限流。
- 看数据倾斜：某个 key 的数据量远大于其他 key。
- 再考虑加并行度、扩资源、拆分热点 key。
只加内存不一定是解法。如果瓶颈在下游数据库，加 Flink 资源只会把压力更快地打过去。
5.3 监控指标
至少盯这些：消费延迟（当前时间 - 最新处理事件时间）、Checkpoint 成功率和耗时、反压比例、每分钟输入/输出条数、失败重启次数、sink 写入失败数。
> 提示：延迟告警不要只看 Kafka lag。Kafka lag 小但窗口没触发，报表照样是旧的。
6. 常见坑
6.1 用处理时间代替事件时间、水位线设置不当
处理时间取决于机器时钟和到达顺序，重跑结果不一致，业务关心“事件什么时候发生”就用事件时间。水位线太小会把正常迟到数据判成迟到，太大则窗口迟迟不触发、延迟升高；先用业务数据分布估算，再留出可接受的等待时间。
6.2 Checkpoint 频繁超时、改并行度不兼容
状态太大、checkpoint 目录慢、同步 IO 阻塞都会导致超时；先看状态大小和 checkpoint 耗时，再考虑增量 checkpoint、RocksDB 调优或减小状态。从 Savepoint 恢复时，算子状态和并行度可能不兼容，上线前要在测试环境演练恢复流程。
6.3 只测正常数据
实时作业最常见的故障是脏数据：空主键、时间戳为 0、金额为负。用 json.ignore-parse-errors 只能跳过解析失败，业务校验仍要自己做。
7. 在线练习
| 题目 | 考点 |
|---|---|
| [1141. 查询近 30 天活跃用户数"> [0, 5000)` 窗口已经关闭，所以它被判定为迟到。真实 Flink 里，迟到数据可以丢弃、更新结果，或者通过侧输出流单独收集。 |
4. 状态、Checkpoint 与 Exactly-Once
4.1 状态是什么
窗口里暂存的聚合值、去重用的订单 ID、维表缓存，都是状态。状态默认在内存里，作业失败就丢；所以要定期做 Checkpoint。
关键配置：
execution.checkpointing.interval: 60s
execution.checkpointing.timeout: 10min
execution.checkpointing.mode: EXACTLY_ONCE
state.backend: rocksdb
state.checkpoints.dir: hdfs://namenode:8020/flink/checkpoints
state.savepoints.dir: hdfs://namenode:8020/flink/savepoints
Checkpoint 是作业运行中自动做的快照，用于失败恢复；Savepoint 是手动触发的快照，用于升级、迁移、改并行度。改代码前先打 Savepoint，是上线的基本纪律。
4.2 Exactly-Once 不是免费的
Flink 内部可以做到 Exactly-Once，但端到端还要看 sink：
Kafka sink 要开启事务或幂等写入；数据库 sink 用主键 upsert 或两阶段提交；文件 sink 先写临时文件，Checkpoint 完成后再原子重命名。
如果 sink 只是普通 INSERT，重复消费就会写出重复行，Flink 内部的状态一致性救不了你。
4.3 去重
订单消息可能重复。常见做法是按主键保留第一条：
SELECT order_id, city, status, amount, event_time
FROM (
SELECT *,
ROW_NUMBER() OVER (
PARTITION BY order_id ORDER BY event_time ASC
) AS rn
FROM kafka_orders
)
WHERE rn = 1;
如果订单会更新，就不能只保留第一条，应该用 upsert-kafka 或按版本号取最新。去重策略取决于业务语义：支付事件只认第一次，订单状态要认最新一次。
4.4 状态 TTL
去重状态如果不清理，会无限增长。给状态设置 TTL：
SET 'table.exec.state.ttl' = '24h';
TTL 太长会占内存，太短会让迟到的重复消息重新出现。按业务允许的最大重复窗口来定。
5. 维表 Join、反压与监控
5.1 维表 Join
实时流要补商品品类、城市名称，常用 lookup join：
SELECT o.order_id, o.city, p.category, o.amount
FROM kafka_orders AS o
LEFT JOIN dim_product FOR SYSTEM_TIME AS OF o.proc_time AS p
ON o.product_id = p.product_id;
维表数据变化后，Flink 需要重新查询或刷新缓存。缓存 TTL 越大，性能越好，但维度更新越慢。要接受“实时链路里的维表不是强一致”的事实。
5.2 反压
反压（backpressure）表示下游处理不过来，压力向上游传导。排查顺序：
- 看 Flink UI 的反压指标，定位卡在哪个算子。
- 看 sink 写入耗时、数据库连接池、外部服务限流。
- 看数据倾斜：某个 key 的数据量远大于其他 key。
- 再考虑加并行度、扩资源、拆分热点 key。
只加内存不一定是解法。如果瓶颈在下游数据库，加 Flink 资源只会把压力更快地打过去。
5.3 监控指标
至少盯这些：消费延迟（当前时间 - 最新处理事件时间）、Checkpoint 成功率和耗时、反压比例、每分钟输入/输出条数、失败重启次数、sink 写入失败数。
提示：延迟告警不要只看 Kafka lag。Kafka lag 小但窗口没触发，报表照样是旧的。
6. 常见坑
6.1 用处理时间代替事件时间、水位线设置不当
处理时间取决于机器时钟和到达顺序，重跑结果不一致，业务关心“事件什么时候发生”就用事件时间。水位线太小会把正常迟到数据判成迟到，太大则窗口迟迟不触发、延迟升高；先用业务数据分布估算，再留出可接受的等待时间。
6.2 Checkpoint 频繁超时、改并行度不兼容
状态太大、checkpoint 目录慢、同步 IO 阻塞都会导致超时；先看状态大小和 checkpoint 耗时，再考虑增量 checkpoint、RocksDB 调优或减小状态。从 Savepoint 恢复时，算子状态和并行度可能不兼容，上线前要在测试环境演练恢复流程。
6.3 只测正常数据
实时作业最常见的故障是脏数据：空主键、时间戳为 0、金额为负。用 json.ignore-parse-errors 只能跳过解析失败，业务校验仍要自己做。
7. 在线练习
| 题目 | 考点 |
|---|---|
| 1141. 查询近 30 天活跃用户数 | 时间窗口与活跃口径 |
| 1454. 活跃用户 🔒 | 连续活跃与去重 |
| 1321. 餐馆营业额变化增长 | 滚动窗口与累计指标 |
8. 面试延伸
面试题：Flink 的 Checkpoint 和 Savepoint 有什么区别？
是什么：Checkpoint 是运行中自动做的状态快照，用于故障恢复；Savepoint 是手动触发的快照，用于升级和迁移。
解决什么问题：没有快照，作业失败就要从头消费；没有 Savepoint，改代码和并行度就没有安全的回滚点。
我的实操边界：本章给出配置示例和 Python 窗口模拟，未在真实 Flink 集群上执行 Checkpoint 恢复。
面试题：Exactly-Once 怎么保证？
是什么：Flink 内部靠 Checkpoint 和 barrier 对齐保证状态一致；端到端还要 sink 支持事务或幂等。
解决什么问题：避免失败恢复后重复计算或丢数据。
我的实操边界：了解原理和 Kafka 事务 sink 的配置方向，未做过两阶段提交的生产落地。
面试题：实时任务延迟高怎么排查？
是什么：先看反压和 Checkpoint，再看数据倾斜、下游写入和外部依赖。
解决什么问题：延迟可能来自任何一段，盲目加资源可能掩盖真正瓶颈。
我的实操边界：本章列出排查顺序，未接入真实 Flink UI 做调优。
9. 本章验收
- 能画出 Kafka → Flink → ClickHouse 的作业链路，并标出状态和 Checkpoint 位置。
- 能写出带
WATERMARK的 Kafka 源表和 1 分钟滚动窗口聚合。 - 能运行 Python 窗口模拟，并解释 4500 为什么被判为迟到。
- 能说清 Checkpoint、Savepoint、Exactly-Once 三者的关系。
- 能给出至少三种反压排查方向。
- 能说出实时去重和状态 TTL 的取舍。