Flink 时间、窗口与状态

本章目标：能讲清事件时间、水位线、窗口、状态、Checkpoint 和精确一次，能写一个带事件时间窗口的 PyFlink 程序，并回答实时计算最常考的四个原理题。 前置知识：先读 Flink 基础；了解 Kafka 的 partition 和 offset。本章偏选学与面试，重点是机制和取舍。

1. 为什么需要它

1.1 数据到达顺序不等于发生顺序

用户 10:00:01 下单，手机信号差，事件 10:00:08 才到服务器；另一个用户 10:00:03 下单，10:00:04 就到了。

发生时间：  10:00:01 下单       10:00:03 下单
到达顺序：        10:00:04 到 → 10:00:08 到

如果按“到达时间”切窗口，10:00:01 的订单会被算到错误的分钟里。实时统计必须能按事件真正发生的时间计算，还要能处理乱序和迟到。

1.2 流处理不能只做逐条转换

“把金额乘以 2”可以逐条做；“每分钟每个品类 GMV”需要记住窗口内的中间结果。这个中间结果就是状态，作业可能运行几个月、状态几十 GB，必须考虑：

• 状态存在哪、怎么清理；
• 作业挂了，状态和消费位置怎么恢复；
• 恢复后会不会重复计算。

这就是水位线、窗口、状态、Checkpoint 和 Exactly-Once 要解决的问题。

2. 核心概念

2.1 三种时间：事件时间、处理时间、摄入时间

电商 GMV、DAU、转化率必须用事件时间。否则补数据、重启作业都会得到不同结果。

2.2 水位线：什么时候不再等迟到数据

水位线（Watermark）是一个时间戳，表示“事件时间小于等于这个值的数据，理论上都已经到了”。它是 Flink 处理乱序的机制。

常见生成方式：

• 有界乱序：Watermark = 当前最大事件时间 - 允许乱序时长。例如允许 5 秒乱序，最大事件时间是 10:00:10，水位线就是 10:00:05。
• 单调递增：假设数据不会乱序，水位线等于最大事件时间，实际很少用。

收到事件时间： 10:00:10, 10:00:04, 10:00:07
允许乱序：     5 秒
水位线变化：   10:00:05（最大 10）→ 不变 → 不变
10:00:04 的数据：早于水位线，属于迟到数据

窗口什么时候触发？以 1 分钟滚动窗口为例，窗口范围是 [10:00:00, 10:01:00)。当水位线大于等于 10:01:00 时，Flink 认为这个窗口的数据到齐了，触发计算并输出。

面试高频：水位线不是“数据必须到齐”的承诺，而是一种容忍乱序的启发式策略：等多久由你配置，等不到的只能按迟到数据另行处理。

空闲分区问题：一个 source 有多个 partition，如果某个 partition 长时间没有数据，它的水位线不推进，可能拖住整个作业。可以用 with_idleness 配置空闲检测，让空闲分区不再影响全局水位线。

2.3 迟到数据：allowed lateness 与侧输出

水位线过了窗口结束时间后，窗口默认触发并销毁，之后到达的同窗口数据会被丢弃。两种处理方式：

1. Allowed Lateness：允许窗口在触发后保留一段时间。迟到数据到达时，重新触发计算并更新结果。例如允许迟到 1 分钟，窗口 [10:00, 10:01) 在 10:01 触发后，状态保留到 10:02。
2. Side Output：把超过允许迟到时间的数据打到一个侧输出流，单独存储、告警或离线补算。

窗口触发 → 输出结果
   │
   ├── 迟到但在 allowed lateness 内 → 更新结果
   └── 超过 allowed lateness → 侧输出/丢弃

注意：allowed lateness 会增加状态保留时间和重复输出的可能。下游如果是普通追加写，更新结果会变成两条；要用 upsert 或按窗口键覆盖。

2.4 窗口：滚动、滑动、会话

滚动（5 分钟）：[0,5) [5,10) [10,15)
滑动（长度 5，步长 1）：[0,5) [1,6) [2,7) ...
会话（间隙 30 秒）：[事件A 事件B] ---30秒空闲--- [事件C]

窗口聚合分两类：

• 增量聚合：来一条更新一次累加器，例如 reduce、aggregate，状态小；
• 全量聚合：窗口触发时遍历所有元素，例如 apply、process，灵活但状态大。

能增量就增量，这是实时作业状态控制的基本原则。

2.5 状态：keyed state 与 operator state

Keyed State 只能在 keyBy 之后使用。Flink 按 key group 管理状态，扩缩容时以 key group 为单位迁移。

状态后端决定状态存在哪：

• HashMapStateBackend：状态放 JVM 堆内存，快，但受内存限制，GC 压力大；
• EmbeddedRocksDBStateBackend：状态放 RocksDB（本地磁盘），支持超大状态和增量 checkpoint，读写有序列化成本。

状态 TTL：给状态设置过期时间，避免用户维度状态无限增长。例如“用户最近一次登录”保留 7 天：

SET 'table.exec.state.ttl' = '7d';

DataStream API 里通过 StateTtlConfig 给单个状态描述符开启 TTL。

2.6 Checkpoint 与 Savepoint

Checkpoint 是 Flink 周期性做的自动快照，用来故障恢复。

Source 注入 barrier ──→ 算子收到 barrier 后快照自己的状态
                         ↓
              状态写到 HDFS/S3/本地文件系统
                         ↓
              JobManager 记录本次 checkpoint 完成

关键点：

• barrier 随数据流向下游传递，算子只快照 barrier 之前的数据，保证状态一致；
• checkpoint 是异步的，不长时间阻塞数据处理；
• 反压严重时可以用 unaligned checkpoint 减少等待；
• checkpoint 保存了算子状态和 source 的消费位置，恢复时从最近一次成功的 checkpoint 继续。

Savepoint 是手动触发的快照，通常用于升级作业、迁移集群、修改逻辑前留档。它和 checkpoint 格式类似，但由用户管理，不会自动清理。

2.7 Exactly-Once：两阶段提交与幂等写入

“精确一次”不是“数据只发一次”，而是“最终结果只生效一次”。端到端要满足三段：

1. Source 可重放：Kafka 能按 offset 重新读；
2. 状态和 offset 一致：checkpoint 同时保存计算状态和消费位置；
3. Sink 不重复生效：事务提交或幂等写入。

两阶段提交（2PC） 用于 Kafka sink 这类支持事务的系统：

Checkpoint 开始
  → sink 开启事务，把结果写入但先不提交
  → barrier 到达，sink 做 pre-commit
  → checkpoint 完成
  → JobManager 通知 sink 正式 commit
失败时：checkpoint 未完成 → 事务回滚/超时中止

Flink 的 Kafka sink 提供 Semantic.EXACTLY_ONCE，底层就是两阶段提交。要注意事务超时时间必须小于 broker 允许的最大事务超时，否则事务会被 broker 中止。

幂等写入适用于 Doris、MySQL、ClickHouse 这类有主键/唯一键的系统：

结果按 (window_start, category) 唯一键 upsert
同一批次重放两次 → 第二次覆盖第一次 → 结果不变

这时即使 sink 是 at-least-once，最终结果也是精确一次。工程上这往往比事务更简单。

注意：Exactly-Once 只保证 Flink 作业内部和受控 sink 的结果；如果下游还要发邮件、调外部接口，副作用无法自动回滚，需要业务侧去重。

2.8 反压：下游处理不过来

反压（Backpressure）是下游消费速度小于上游生产速度时，压力沿数据流向上游传递的现象。

Flink 使用基于 credit 的流量控制：下游告诉上游“我还有多少缓冲”，上游按额度发送，避免把下游打爆。

常见原因：

• sink 写入慢（数据库、Doris、外部接口）；
• 某个 key 数据倾斜，单个子任务处理慢；
• 算子里有同步阻塞调用；
• 并行度不足或 slot 不够；
• GC 频繁、状态过大。

排查看 backPressuredTimeMsPerSecond、outPoolUsage、inPoolUsage 等指标，先定位瓶颈算子，再决定加并行度、优化 sink 批量写、拆分热点 key 或异步 I/O。

3. 动手示例

3.1 事件时间滚动窗口（PyFlink 本地）

新建 realtime_demo/flink_event_time_demo.py：

from pyflink.common import Duration, Row, Types, WatermarkStrategy
from pyflink.common.time import Time
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.datastream.window import TumblingEventTimeWindows


class EventTimeAssigner(TimestampAssigner):
    def extract_timestamp(self, value, record_timestamp):
        return value[0]


env = StreamExecutionEnvironment.get_execution_environment()
env.set_parallelism(1)

rows = [
    Row(1700000000000, "O1001", "手机", 2999.0),
    Row(1700000010000, "O1002", "图书", 59.0),
    Row(1700000020000, "O1003", "手机", 1999.0),
    Row(1700000030000, "O1004", "图书", 89.0),
]

ds = env.from_collection(
    rows,
    type_info=Types.ROW([Types.LONG(), Types.STRING(), Types.STRING(), Types.DOUBLE()]),
)

watermarks = (
    WatermarkStrategy.for_bounded_out_of_orderness(Duration.of_seconds(5))
    .with_timestamp_assigner(EventTimeAssigner())
)

result = (
    ds.assign_timestamps_and_watermarks(watermarks)
    .key_by(lambda r: r[2])
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .reduce(lambda a, b: Row(a[0], a[1], a[2], a[3] + b[3]))
    .map(lambda r: (r[2], r[3]), output_type=Types.TUPLE([Types.STRING(), Types.DOUBLE()]))
)

result.print()
env.execute("event time window demo")

运行：

python realtime_demo/flink_event_time_demo.py

四条数据都在同一个 1 分钟窗口内，按品类聚合后输出“手机”和“图书”的金额合计。因为 from_collection 有界，输入结束后水位线推进到最大，窗口立即触发。

3.2 加上 allowed lateness 和侧输出

在窗口流上增加允许迟到和侧输出：

from pyflink.datastream import OutputTag

late_tag = OutputTag(
    "late",
    Types.ROW([Types.LONG(), Types.STRING(), Types.STRING(), Types.DOUBLE()]),
)

windowed = (
    ds.assign_timestamps_and_watermarks(watermarks)
    .key_by(lambda r: r[2])
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowed_lateness(Time.minutes(1))
    .side_output_late_data(late_tag)
)

result = windowed.reduce(lambda a, b: Row(a[0], a[1], a[2], a[3] + b[3]))
result.print()
result.get_side_output(late_tag).print()

含义：窗口触发后仍保留 1 分钟状态，这期间到达的迟到数据会更新窗口结果；超过 1 分钟的迟到数据进入 late 侧输出，可以单独落表做离线补算。

3.3 Checkpoint 和状态 TTL 配置

核心配置可以写在 flink-conf.yaml：

execution.checkpointing.interval: 60s
execution.checkpointing.timeout: 2min
execution.checkpointing.min-pause: 30s
execution.checkpointing.max-concurrent-checkpoints: 1
state.backend.type: hashmap
state.checkpoints.dir: hdfs://namenode:8020/flink/checkpoints

PyFlink 代码里也可以开启：

env.enable_checkpointing(60_000)
checkpoint_config = env.get_checkpoint_config()
checkpoint_config.set_checkpoint_timeout(120_000)
checkpoint_config.set_min_pause_between_checkpoints(30_000)
checkpoint_config.set_max_concurrent_checkpoints(1)

Flink SQL 的 keyed 状态过期：

SET 'table.exec.state.ttl' = '24h';

TTL 不能乱设：设太短，迟到数据或跨天累计会算错；设太长，状态膨胀，checkpoint 变慢。

3.4 端到端精确一次的选型

先问下游支不支持幂等，再决定要不要上事务。不要为了“听起来高级”强行 2PC。

4. 常见坑

1. 用处理时间做业务统计。作业重启后结果会变，补历史数据也对不上；业务口径优先事件时间。
2. 水位线设置过小。乱序数据大量被当成迟到丢弃；设置过大则结果输出延迟高，要按业务乱序程度权衡。
3. 多个 source 并行度不一致导致水位线不推进。检查空闲分区，必要时开启 idleness。
4. allowed lateness 设置过长。窗口状态长期不释放，checkpoint 越来越大。
5. 窗口结果追加写下游。迟到更新会产生重复行，下游必须支持 upsert 或按窗口键覆盖。
6. 状态 TTL 设得太短。跨天累计、长会话会被提前清掉，指标偏低。
7. 只开 checkpoint 不配存储和超时。默认存 JobManager 内存，生产不可靠；要配 HDFS/S3 等持久化路径。
8. 认为开了 2PC 就万事大吉。事务超时、下游不支持读已提交、外部副作用都会破坏端到端一致性。
9. 反压时盲目加并行度。如果瓶颈是单个 key 或外部数据库，加并行度只会增加下游压力。

5. 练习

练习 1：选择时间语义

下列场景分别应该用什么时间，为什么：实时大屏 GMV、服务器 CPU 告警、用户浏览会话分析。

练习 2：水位线计算

允许乱序 5 秒，依次收到事件时间 10:00:10、10:00:04、10:00:07、10:00:12。写出每次收到后的水位线，并指出哪条是迟到数据。

练习 3：选择窗口类型

需求分别是“每分钟统计一次最近 5 分钟的支付金额”“每 5 分钟统计一次各品类订单数”“统计用户一次浏览会话内的商品数”，分别选什么窗口？

练习 4：allowed lateness 和侧输出怎么配合

业务允许窗口结果晚到 2 分钟，超过 2 分钟的数据不能丢，要进入单独的表。说明实现方式，并指出下游写入要注意什么。

练习 5：状态、Checkpoint 和 Savepoint

判断对错并说明理由：A. checkpoint 和 savepoint 都是自动触发的；B. 状态 TTL 过期后，checkpoint 里也不会再保留；C. 作业升级逻辑前应手动做 savepoint。

练习 6：设计端到端精确一次

链路是 Kafka → Flink 聚合 → Doris，Doris 表有 (window_start, category) 唯一键。说明怎样做到结果不重不漏。如果 sink 换成 Kafka，又要改什么？

6. 面试延伸

面试题：水位线解决什么问题？

是什么：水位线是一个事件时间进度标记，表示“小于该时间的数据基本到齐”，用来触发窗口计算。

解决什么问题：网络延迟导致数据乱序，不能按到达顺序切窗口；水位线让 Flink 在等待一段时间后按事件时间触发，兼顾正确性和延迟。

我的实操边界：我在本地 PyFlink 里配置过 5 秒有界乱序水位线，观察过窗口触发；了解空闲分区和 allowed lateness，但没有在生产环境调过大规模乱序和空闲 source。

面试题：Flink 的精确一次怎么实现？

是什么：端到端精确一次 = 可重放 source + checkpoint 保存状态和 offset + sink 事务或幂等写入。

解决什么问题：故障恢复后重复消费或漏消费，导致结果翻倍或缺失。

我的实操边界：我理解 checkpoint barrier、两阶段提交和 Doris 唯一键 upsert 的差别，本地没搭过 Kafka 事务 sink；生产级故障演练没有做过。

面试题：Checkpoint 和 Savepoint 的区别？

是什么：两者都是状态快照，checkpoint 自动用于故障恢复，savepoint 手动用于升级和迁移。

解决什么问题：checkpoint 保证作业挂了能恢复；savepoint 保证改逻辑、换集群时能带走状态。

我的实操边界：我本地开启过 checkpoint，了解存储路径、超时和并发数配置；没有做过跨集群 savepoint 迁移。

面试题：反压是什么，怎么处理？

是什么：下游处理速度跟不上上游，压力沿数据流反向传递，导致上游也慢下来。

解决什么问题：防止下游被压垮，同时让瓶颈暴露出来。

我的实操边界：我理解 credit-based 流控和常见指标，本地小作业没有触发过严重反压；没有做过生产作业的线程栈和 GC 分析。

处理思路：定位瓶颈算子，看倾斜、sink、GC、资源，再通过批量写、异步 I/O、拆热点 key、加资源解决。

7. 本章验收

• 能用自己的话区分事件时间、处理时间和摄入时间，并为三个业务场景选对时间语义。
• 能根据“最大事件时间 - 允许乱序”算出水位线，并判断一条数据是否迟到。
• 能说出滚动、滑动、会话窗口的区别，并为“最近 5 分钟”需求选出滑动窗口。
• 能解释 allowed lateness 和侧输出各自处理哪一类迟到数据。
• 能说清 keyed state、状态后端、TTL 的作用，以及 checkpoint 和 savepoint 的区别。
• 能画出 Kafka → Flink → Kafka 的两阶段提交流程，并说明 Doris 链路为什么可以用唯一键 upsert 代替事务。
• 能列出反压的三个常见原因和对应的排查方向。