流式计算的理论基础：时间、窗口与一致性

流式计算不只是"把批处理变快"——它是一个完全不同的计算范式，背后有一套独特的理论体系。这套理论由 Google 的 DataFlow 论文（2015 年）奠定基础，随后被 Apache Flink、Spark Structured Streaming 等系统实现落地。

本文从理论角度系统梳理流式计算的核心概念：时间模型、Watermark、窗口、触发器、状态一致性、Exactly-Once 语义，以及流批统一的理论基础。理解这些理论，是真正掌握 Flink 等流处理系统的前提。

一、流与批：两种截然不同的计算模型

数据的有界性：根本分歧

批处理假设数据是有界的（Bounded）：先收集，再计算，输出一个确定的结果。

批处理模型：
  f(D) → R
  D 是有限数据集，f 是计算函数，R 是确定的结果

  一天的订单数据 → 计算 → 今天的总销售额（确定值）

流处理假设数据是无界的（Unbounded）：数据像河流一样持续流入，计算结果随时间不断更新。

流处理模型：
  f(S, t) → R(t)
  S 是无界数据流，f 是持续计算函数，R(t) 是在时刻 t 的当前结果

  持续流入的订单 → 持续计算 → 当前累计销售额（实时更新）

这不只是工程实现的差异，而是数学模型的根本不同。批处理求的是一个静态函数的值；流处理求的是一个时序函数，它的"答案"在不断演化。

Lambda 架构：批流并存时代的妥协

早期（2012 年前后），人们没有成熟的流处理框架，于是提出了 Lambda 架构：用批处理保证准确性，用流处理保证低延迟，两套系统并行运行，结果合并：

graph TD
    RAW[原始数据]
    RAW --> BATCH[批处理层 Hadoop
每天跑一次
精确结果 T+1]
    RAW --> SPEED[速度层 Storm
实时处理
近似结果 秒级]
    BATCH --> SERVE[服务层
合并两层结果
对外提供查询]
    SPEED --> SERVE

    style RAW fill:#f9f,stroke:#333
    style BATCH fill:#bbf,stroke:#333
    style SPEED fill:#fbb,stroke:#333
    style SERVE fill:#bfb,stroke:#333

Lambda 架构的存在本质上是在承认：当时的流处理系统无法同时保证低延迟和高正确性。随着 Flink 的成熟，这个妥协方案已经逐渐退出历史舞台。

二、时间模型：流处理最深的理论问题

流处理里最核心也最容易被低估的理论问题是：时间有两个维度，而且它们可以不一致。

Event Time 与 Processing Time

Event Time（事件时间）：
  事件真实发生的时刻，记录在数据里
  例：用户在 10:00:01 点击了按钮，这个时间戳写在日志里

Processing Time（处理时间）：
  数据被流处理系统看到的时刻，即"系统时钟"
  例：这条日志在 10:00:05 到达 Flink，被处理的时间是 10:00:05

两者之间的差距叫做摄入延迟（Ingestion Lag），来源于：

• 网络传输延迟（手机 App 在弱网下积攒日志，网络恢复后批量上传）
• 消息队列堆积（Kafka 消费者滞后）
• 时钟不同步（不同服务器时钟偏差）

真实场景：用户手机在地铁里断网，累积了 5 分钟的行为日志，出站后上传

Event Time 视角：
  10:00:01 点击 → 10:00:03 点击 → 10:00:05 点击 → 10:04:58 点击

Processing Time 视角（这些数据在 10:05:10 到达系统）：
  10:05:10 处理 [et=10:00:01]
  10:05:10 处理 [et=10:00:03]
  10:05:10 处理 [et=10:00:05]
  10:05:10 处理 [et=10:04:58]  ← 5 分钟前的数据，现在才到

如果用 Processing Time 统计"每分钟点击次数"，这 4 次点击都会被算进 10:05 这一分钟，结论完全错误。正确答案应该是：10:00 有 3 次，10:04 有 1 次。

结论：只有基于 Event Time 的计算才能得到语义正确的结果，但这带来了一个根本性的困难：系统无法知道某个时间窗口的数据是否已经全部到达。

乱序（Out-of-order）：流处理的常态

即使同一批数据，Event Time 也可能乱序：

数据到达顺序（Processing Time 排序）：
  [et=10:00:05] [et=10:00:03] [et=10:00:08] [et=10:00:01] [et=10:00:06]
       ↑                              ↑
  先到的是 05                    后到的是 01（乱序！）

原因：不同请求走了不同的网络路径，延迟不同；
      微服务多实例各自攒批发送，攒满触发时间不同

乱序是流处理的常态，不是特例。任何实用的流处理系统都必须正确处理乱序数据。

三、Watermark：对时间进度的系统性断言

Watermark 是解决"何时关闭窗口"问题的核心机制，也是整个流处理理论最精妙的部分。

Watermark 的形式化定义

Watermark(t) 是流中的一个特殊标记，表示：

系统断言：在此标记之后，不会再出现 event_time ≤ t 的数据。

graph LR
    D1["[et=10:00:05]"]
    D2["[et=10:00:03]"]
    W1{{"W(10:00:02) 窗口≤10:00:02 可以关闭"}}
    D3["[et=10:00:08]"]
    W2{{"W(10:00:07) 窗口≤10:00:07 可以关闭"}}
    D4["[et=10:00:01?] 迟到数据！ et < W(10:00:07)"]

    D1 --> D2 --> W1 --> D3 --> W2 --> D4

    style W1 fill:#ff9,stroke:#c90
    style W2 fill:#ff9,stroke:#c90
    style D4 fill:#fcc,stroke:#c00

当 Watermark 超过某个窗口的结束时间，该窗口就可以安全关闭并输出结果了。

Watermark 的生成策略

最常见的 Watermark 生成策略是：

W(t_processing) = max(event_time seen so far) - allowed_lateness

其中 allowed_lateness 是允许的最大延迟容忍时间（如 5 分钟）

例：
  已观察到的最大 event_time = 10:05:30
  allowed_lateness = 5 分钟
  Watermark = 10:05:30 - 5min = 10:00:30

  意思是：我猜测 ≤ 10:00:30 的数据应该都到了
  所以 [10:00:00, 10:00:30] 之前的窗口可以关闭

这个 allowed_lateness 参数揭示了一个无法消除的根本权衡：

allowed_lateness 设置	结果延迟	正确性	适用场景
0（激进 Watermark）	最低	差（大量数据被视为迟到）	允许一定数据丢失的监控指标
5 分钟（保守）	较高（5 分钟）	好（覆盖大多数延迟）	实时报表，对准确性要求高
30 分钟（极保守）	高	极好	金融结算等精确场景

这个权衡没有标准答案，取决于业务对延迟和准确性的相对重视程度。

多流场景：Watermark 的取最小值原则

当多个并行流合并时（如 Flink 的多个 Partition），Watermark 取所有上游的最小值：

graph LR
    P1["Partition 1 W(10:05:30)"]
    P2["Partition 2 W(10:03:10)"]
    P3["Partition 3 W(10:05:15)"]
    MERGE{{"合并算子 Watermark = min = 10:03:10"}}
    OUT["下游 Watermark = 10:03:10"]
    IDLE["⚠️ 若某 Partition 停止发数据 Watermark 永远不推进 → 所有窗口无法关闭 → 需 idle source 处理机制"]

    P1 --> MERGE
    P2 --> MERGE
    P3 --> MERGE
    MERGE --> OUT
    P2 -. "瓶颈" .-> IDLE

    style MERGE fill:#ff9,stroke:#c90
    style IDLE fill:#fcc,stroke:#c00
    style P2 fill:#fbb,stroke:#c00

四、窗口：给无界流分段计算

无界流必须切成有限片段才能计算聚合，这就是窗口（Window）的作用。窗口定义了"在哪段数据上做计算"。

滚动窗口（Tumbling Window）

大小固定，不重叠，每条数据恰好属于一个窗口。适合：每分钟统计一次、每小时汇总一次。特点：计算结果之间相互独立，无重叠，状态不复用。

gantt
    title 滚动窗口（窗口大小 = 1 分钟，无重叠）
    dateFormat mm:ss
    axisFormat %M:%S
    section 窗口
    W1 [0,60)  : 00:00, 60s
    W2 [60,120) : 01:00, 60s
    W3 [120,180) : 02:00, 60s
    W4 [180,240) : 03:00, 60s

滑动窗口（Sliding Window）

窗口大小固定，但步长小于窗口大小，窗口间有重叠。同一条数据可能属于多个窗口（最多 size/slide 个）。适合：过去 5 分钟的滑动平均、移动窗口统计。代价：每条数据被计算 size/slide 次，计算量更大，状态也更多。

gantt
    title 滑动窗口（窗口大小 5 分钟，步长 1 分钟）
    dateFormat mm:ss
    axisFormat %M:%S
    section 窗口
    W1 [0,5min)   : 00:00, 300s
    W2 [1,6min)   : 01:00, 300s
    W3 [2,7min)   : 02:00, 300s
    W4 [3,8min)   : 03:00, 300s

会话窗口（Session Window）

基于活动间隔动态划分，窗口大小不固定。适合：用户会话分析，同一次访问中的行为归为一组。特点：窗口边界由数据本身决定，不是预先固定的。挑战：需要动态合并窗口（当两个 session 之间的间隔被一条数据填上时）。

gantt
    title 会话窗口（间隔超过 30 分钟则分割）
    dateFormat HH:mm
    axisFormat %H:%M
    section 用户活动
    点击点击点击 (Session 1) : 10:00, 15m
    30分钟无活动             : crit, 10:15, 30m
    点击点击 (Session 2)     : 10:45, 10m
    30分钟无活动             : crit, 10:55, 30m
    点击 (Session 3)         : 11:25, 5m

窗口的理论本质：数据分配函数

从理论角度，窗口可以被看作一个函数：

assign(event) → Set<Window>

对每条输入数据，返回它属于哪些窗口

滚动窗口：assign(event) = { [floor(et/size)*size, floor(et/size)*size + size) }
滑动窗口：assign(event) = { 所有包含 et 的 [n*slide, n*slide+size) }
会话窗口：assign(event) = { [et, et+gap) }，之后动态合并相交窗口

这个视角揭示了窗口类型之间的统一性——它们只是数据分配策略不同，底层计算机制是一样的。

五、触发器（Trigger）：何时输出结果

窗口定义了"在哪算"，触发器定义了"什么时候把结果发出去"。两者是正交的设计，可以自由组合。

三种基本触发策略

1. Watermark 触发（默认）
   条件：Watermark 超过窗口结束时间
   语义："我认为这个窗口的数据差不多到齐了"
   特点：结果准确，但延迟 = 窗口等待时间 + allowed_lateness

2. Processing Time 触发
   条件：距离窗口创建已过去 N 秒
   语义：不管数据到没到齐，每隔 N 秒发一次结果
   特点：低延迟，但结果可能不完整（Early firing，早期触发）

3. 数据量触发
   条件：窗口内积累了 N 条数据
   语义：攒够一批就发
   特点：吞吐可预测，但时间不确定

组合触发器：DataFlow 模型的精华

Google DataFlow 论文的核心贡献之一是提出了可组合的触发器：

典型生产模式：
  Early firing（早期）：Processing Time 每 30 秒触发，输出不完整但低延迟的结果
  On-time firing（准时）：Watermark 触发，输出认为完整的结果
  Late firing（迟到）：对超过 Watermark 还到来的数据再次触发

// Flink 实现示例
DataStream<Tuple2<String, Long>> result = stream
    .keyBy(0)
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .trigger(
        ContinuousEventTimeTrigger.of(Time.seconds(10)) // 早期：每 10 秒一次
        // 准时触发由 EventTimeTrigger 默认处理
    )
    .allowedLateness(Time.minutes(5))   // 允许 5 分钟迟到数据
    .sideOutputLateData(lateOutputTag)  // 超级迟到数据发到侧输出
    .sum(1);

这个模式在实践中极为常见：用户先看到一个"大约对"的实时结果，之后等数据到齐了再看到"完全准确"的最终结果，超迟到的数据则进入单独处理流程。

DataFlow 模型的四个问题

Google 的 DataFlow 论文用四个问题统一描述了流处理的所有核心概念：

问题	对应概念	示例
What：计算什么结果？	聚合函数（Sum/Count/Join...）	统计点击次数
Where：在哪些数据上算？	窗口（Tumbling/Sliding/Session）	每 1 分钟
When：什么时候输出？	触发器（Watermark/PT/Count）	数据到齐后 + 每 10 秒早期结果
How：迟到数据怎么处理？	Accumulation Mode	替换已有结果 / 累积

其中 "How" 对应的 Accumulation Mode 有三种：

• Discarding：每次触发只输出这次新增的数据的结果，不累积
• Accumulating：每次触发输出从窗口开始到现在的累积结果（最常用）
• Accumulating & Retracting：不仅发新结果，还发一条"撤回"消息取消上次的结果（下游可以精确维护）

六、状态（State）：流处理的记忆

无状态流处理（如过滤、映射）很简单。真正的挑战来自有状态计算——需要记住过去的数据才能计算当前结果。

状态的类型

Keyed State（键控状态）：
  每个 Key 维护独立的状态，最常见
  例：统计每个用户的点击次数，每个 user_id 对应一个计数器

  ValueState<Long> count;        // 单个值
  ListState<Event> eventList;    // 值列表
  MapState<String, Long> map;    // 键值对
  ReducingState<Long> sum;       // 持续聚合
  AggregatingState<Event, Stats> // 自定义聚合

Operator State（算子状态）：
  整个算子实例共享一个状态，不按 Key 分区
  例：Kafka Source 记录每个 Partition 的消费 offset

Checkpoint：状态一致性的保障

流处理系统长期运行，机器随时可能崩溃。如何保证崩溃恢复后状态仍然正确，是流处理的核心挑战。

Flink 使用 Chandy-Lamport 算法的变体来做分布式快照（Checkpoint）：

sequenceDiagram
    participant JM as JobManager
    participant SRC as Source（Kafka）
    participant OP as 中间算子
    participant STORE as 持久化存储
HDFS/S3

    JM->>SRC: 触发 Checkpoint(ckpt=5)
    SRC->>SRC: 快照自身状态（Kafka offset）
    SRC->>OP: 数据流：[d1][d2][d3][Barrier(ckpt=5)][d4]...
    Note over OP: 收到 Barrier 后
等待所有输入对齐
    OP->>OP: 对自己的状态做快照
    OP->>OP: 向下游广播 Barrier
    SRC->>STORE: 写入状态快照
    OP->>STORE: 写入状态快照
    STORE-->>JM: Checkpoint 完成确认

    Note over JM,STORE: 崩溃恢复：从最近成功的 Checkpoint
恢复状态，Kafka Source 重置 offset 重放数据

状态后端（State Backend）的选择

HashMapStateBackend（原 MemoryStateBackend）：
  状态存在 JVM 堆内存
  优点：访问极快（O(1)，无序列化开销）
  缺点：受 JVM 堆大小限制，GC 压力大，不适合大状态
  适合：状态小、延迟要求极低的场景

EmbeddedRocksDBStateBackend（原 RocksDBStateBackend）：
  状态存在本地磁盘（RocksDB LSM-Tree）
  优点：状态可以远超内存大小，增量 Checkpoint（只保存变化部分）
  缺点：访问需要序列化/反序列化，延迟高 10-100 倍
  适合：状态大（GB 到 TB 级）的场景，如用户画像、设备历史

选择原则：
  状态 < 几 GB → HashMapStateBackend
  状态 > 几 GB 或需要增量 Checkpoint → EmbeddedRocksDBStateBackend

无界状态：流处理的长期挑战

无界数据流会产生不断增长的状态。例如"对历史上所有出现过的 user_id 去重"——如果用户 ID 持续增长，状态将无限膨胀。

解决方案：

• State TTL（状态过期）：超过一定时间未访问的状态自动清理
• 业务设计约束：把"全局去重"改为"7 天内去重"，从语义上限制状态大小
• 布隆过滤器：用概率数据结构近似去重，牺牲极小精度换取 O(1) 的恒定内存

七、Exactly-Once：流处理的语义圣杯

三种处理语义的对比

At-most-once（至多一次）：
  实现：不做任何重试或确认
  结果：数据可能丢失，但不会重复
  适合：监控告警，偶尔丢一条数据可接受

At-least-once（至少一次）：
  实现：崩溃后从上次 offset 重放，但不撤销已写入的结果
  结果：数据不丢失，但可能重复（同一条数据处理两次）
  适合：大多数日志分析，重复多算一次影响不大

Exactly-once（恰好一次）：
  实现：复杂，需要 Source 可重放 + Sink 幂等/事务性写入
  结果：既不丢失也不重复
  适合：金融交易、精确计费等不能有误差的场景

Exactly-Once 的两阶段提交实现

Flink 实现端到端 Exactly-Once（从 Kafka 读到 Kafka 写）的核心是两阶段提交（2PC），与数据库事务中的 2PC 原理相同：

sequenceDiagram
    participant JM as JobManager
    participant SINK as Sink 算子
    participant KAFKA as Kafka（下游）

    rect rgb(220, 240, 255)
        Note over JM,KAFKA: Phase 1 — Pre-commit（预提交）
        JM->>SINK: 触发 Checkpoint
        SINK->>KAFKA: 写入数据到 Kafka Transaction
（数据对消费者不可见）
        SINK->>SINK: 快照状态：Transaction = pre-committed
        SINK-->>JM: Checkpoint 完成
    end

    rect rgb(220, 255, 220)
        Note over JM,KAFKA: Phase 2 — Commit（提交）
        JM->>SINK: 所有算子 Checkpoint 完成通知
        SINK->>KAFKA: Kafka Transaction Commit
        Note over KAFKA: 数据对消费者可见
    end

    rect rgb(255, 240, 220)
        Note over JM,KAFKA: 崩溃恢复 — 场景 A：Checkpoint 未完成时崩溃
        Note over SINK,KAFKA: 未 Commit 的 Transaction 超时自动 Abort
从上一个 Checkpoint 的 offset 重放
重新写入新 Transaction → 最终 Commit
    end

    rect rgb(255, 220, 220)
        Note over JM,KAFKA: 崩溃恢复 — 场景 B：Checkpoint 完成后 Commit 前崩溃
        Note over SINK,KAFKA: 恢复到该 Checkpoint 状态
状态记录有 pre-committed Transaction
恢复后补充执行 Commit → 不会丢失
    end

Exactly-Once 的代价

Exactly-Once 不是免费的：

• 吞吐降低约 10-30%：Barrier 对齐需要等待最慢的上游，造成 stall
• 延迟增加：数据要等到 Checkpoint 完成才对下游可见，延迟 = Checkpoint 间隔
• Sink 必须支持事务：HDFS、Kafka 支持；MySQL 要用 JDBC 事务；部分存储不支持

对于非关键路径，At-least-once 往往是更经济的选择；只有金融级业务才需要全链路 Exactly-Once。

Flink 1.11 引入的 Unaligned Checkpoint（非对齐 Checkpoint）缓解了这个问题：不等待 Barrier 对齐，把未处理的数据一起纳入快照，以换取更低的 stall 时间，但快照大小会增加。

八、流批统一：两个世界的终极融合

批是流的特例

从理论角度，批处理是流处理的特殊情况：有界数据集可以看作一条带有确定终点的流。

批处理 = 流处理中 Watermark = +∞ 的特殊情况

等价关系：
  批处理"对全量数据计算"
  = 流处理"Watermark 推进到无穷大，触发所有窗口的最终计算"

Flink 的设计哲学：
  - 内部只有一套执行引擎（流引擎）
  - 批处理 = BoundedStream，Watermark 在数据结束时设为 MAX_VALUE
  - 自动触发所有未关闭的窗口，得到完整结果

Micro-batch：流是批的极限

反过来也可以成立：流处理是批处理的极限情况——把批的时间窗口缩到无限小，就趋近于流。

Spark Streaming / Spark Structured Streaming 的思路：
  把持续的数据流切成极小的"micro-batch"（如 1 秒一批）
  对每个 micro-batch 跑一次完整的批处理
  模拟出"连续流"的效果

批处理 batch_size → ∞：传统离线批处理
批处理 batch_size → 0：趋近于原生流处理

Micro-batch 的优缺点：
  优点：利用成熟的批处理引擎，Exactly-Once 更容易实现
  缺点：延迟有下限（一个 micro-batch 的时间），无法做到毫秒级

两种路径的本质区别

原生流（Native Streaming）：Flink、Storm
  每条数据来了就处理
  延迟：毫秒级
  状态：持续维护，更复杂
  Exactly-Once：需要 2PC，有成本

Micro-batch：Spark Streaming
  每隔一段时间处理一批
  延迟：秒级（受 batch interval 限制）
  状态：每个 batch 结束后更新，相对简单
  Exactly-Once：利用批处理的原子性，相对容易

"Micro-batch 不是真正的流处理"这个判断在理论上是正确的：
  它的延迟模型、状态模型、时间语义都和原生流有本质差异。
  但在工程实践上，如果业务不需要毫秒级延迟，Spark Structured Streaming
  是更简单、更稳定的选择。

九、关键点总结

• 流处理的本质是在不完整信息下做决策：数据永远可能迟到，系统只能用 Watermark 做有根据的猜测
• Event Time vs Processing Time 是语义正确性的关键：用 Processing Time 计算的结果语义是错的，但实现最简单；用 Event Time 才能得到真正正确的结果，但需要处理乱序和迟到
• Watermark 是时间进度的断言，not 保证：它表达"我认为 ≤t 的数据到齐了"，而不是"一定到齐了"，超过 Watermark 到来的数据就是迟到数据，需要单独处理策略
• 窗口和触发器是正交的设计：窗口决定"在哪算"，触发器决定"什么时候输出"，DataFlow 模型用 What/Where/When/How 四个维度统一了流处理的所有场景
• 有状态计算是流处理的真正难点：Checkpoint（基于 Chandy-Lamport 算法）保证状态一致性，状态后端的选择（内存 vs RocksDB）决定了性能上限
• Exactly-Once 依赖两阶段提交：Checkpoint 触发时 Pre-commit，所有算子快照完成后 Commit，崩溃恢复时补充提交或回滚重试，代价是吞吐和延迟的损失
• 流批统一是理论上的自然结论：批处理是 Watermark=∞ 时的流处理；Micro-batch 是 batch_size→0 时趋近于流处理；两条路径在延迟、状态复杂度、Exactly-Once 实现难度上有本质差异
• 多流 Watermark 取最小值：是保证正确性的必要条件，但 idle source 需要特殊处理，否则会导致 Watermark 停滞