上一篇解决了 Structured Streaming 的基本模型。这一篇进入时间语义——事件时间、水位线和状态管理。

流处理中的"时间"不是一个简单的概念。一条消息有两个时间戳：事件时间（Event Time，事件实际发生的时间）和处理时间（Processing Time，Spark 收到消息的时间）。两者可能差距很大——网络延迟、消费积压、跨时区传输都会导致消息乱序到达。

本文只抓一个问题：Watermark 如何告诉 Spark “多晚的数据还值得等”，以及 State Store 如何持久化窗口状态。

事件时间 vs 处理时间

1
2
3
4
5
6
7
8
9
10
11
12
事件发生                网络传输              Spark 收到
  t=10:00:01          ~~~延迟 3s~~~          t=10:00:04  (Processing Time)
  t=10:00:02          ~~~延迟 1s~~~          t=10:00:03
  t=10:00:03          ~~~延迟 8s~~~          t=10:00:11  ← 乱序到达

用处理时间做窗口聚合:
  窗口 [10:00:00, 10:00:05): 收到 2 条 → 第三条在 10:00:11 才到达
  结果不准确——同一事件可能因为网络延迟被分到错误的窗口

用事件时间做窗口聚合:
  窗口 [10:00:00, 10:00:05): 事件时间在此范围的 3 条全部归入
  结果准确——但需要等待迟到数据

Structured Streaming 通过 withWatermark 声明事件时间列和允许的最大迟到时间：

1
2
3
4
5
6
7
8
9
val events = spark.readStream
  .format("kafka")
  .load()
  .select($"value".cast("string"), $"timestamp".as("event_time"))

val windowed = events
  .withWatermark("event_time", "10 minutes")  // 允许迟到 10 分钟
  .groupBy(window($"event_time", "5 minutes"))
  .count()

Watermark 的推进机制

单输入下，Watermark 根据已观察到的最大事件时间减去延迟阈值推进。微批计算出的新边界供后续批次使用，窗口算子按窗口结束时间判断状态清理；它不是对每条输入立即应用的墙上时钟截止线。

1
2
3
4
5
6
7
8
9
10
11
Batch 0: 最大事件时间 = 10:15
  Watermark = 10:15 - 10min = 10:05
  → 计算候选边界 10:05，后续批次可据此清理已过期窗口

Batch 1: 最大事件时间 = 10:22
  Watermark = 10:22 - 10min = 10:12
  → 计算候选边界 10:12，生效批次取决于执行推进
  → 边界生效后，已结束窗口可按输出模式输出并清理

Batch 2: 一条事件时间 = 10:08 的数据到达
  若其所属窗口已随生效的 Watermark 清理 → 不再更新该窗口

Watermark 只会向前推进，永远不会倒退。即使某个 batch 的数据事件时间很早，Watermark 也不会降低。这保证了已关闭的窗口不会被重新打开。

Watermark 的阈值是一个权衡：阈值越大，能容纳的迟到数据越多，但窗口关闭越慢、状态保留越久、内存消耗越大。阈值越小，内存占用小、延迟低，但更多迟到数据会被丢弃。

图 1：Watermark 为窗口状态清理提供事件时间边界

flowchart TB
  E[已观察到的事件时间] --> M[维护最大事件时间]
  M --> W[结合延迟阈值计算 Watermark]
  W --> O[算子判断窗口是否可清理]
  O --> K[仍保留的窗口状态]
  O --> R[已清理窗口不再更新]
  L[迟到事件] --> O

图 1 描述的是事件时间推进，不是墙上时钟超时。小于延迟阈值的数据享有不被丢弃的保证；超过阈值的数据不保证一定丢弃，还要看生效的 watermark、算子和窗口状态。多个输入默认按较慢输入形成全局 watermark。

State Store：状态的持久化

有状态操作（聚合、去重、session 窗口）需要在 batch 之间保留中间状态。Structured Streaming 使用 State Store 来管理这些状态。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
State Store 架构:

  Executor 内存
  ┌────────────────────────┐
  │  State Store (HashMap) │  当前 batch 的状态
  │  key → value           │
  └────────┬───────────────┘
           │ 每个 batch 结束后
           ▼
  Checkpoint (HDFS/S3)
  ┌────────────────────────┐
  │  状态快照 + 增量 delta  │  持久化的状态
  │  用于故障恢复           │
  └────────────────────────┘

默认的 State Store 实现是 HDFSBackedStateStore：状态以 HashMap 形式存在 Executor 内存中，每个 batch 结束后把增量变更（delta）写入 checkpoint 目录。每隔 N 个 batch 写一次全量快照，防止 delta 文件累积过多。

故障恢复时，State Store 从最近的快照加上后续的 delta 文件重建状态。这个过程可能需要几秒到几十秒，取决于状态大小和 delta 数量。

Watermark 的另一个作用是状态清理：当 Watermark 推进超过某个窗口的结束时间时，该窗口的状态可以安全清除。对于持续新增 key 或窗口的聚合，没有相应清理机制会使状态持续增长；Complete 模式还会保留全部聚合结果。

窗口类型

Structured Streaming 支持三种窗口：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
滚动窗口 (Tumbling Window):
  window(event_time, "5 minutes")
  |-------|-------|-------|-------|
  [0, 5)  [5, 10) [10,15) [15,20)
  每个事件属于且仅属于一个窗口

滑动窗口 (Sliding Window):
  window(event_time, "10 minutes", "5 minutes")
  |------------|
       |------------|
            |------------|
  窗口大小 10 分钟，滑动步长 5 分钟
  一个事件可能属于多个重叠窗口

Session 窗口 (Spark 3.2+):
  session_window(event_time, "5 minutes")
  |--gap<5min--|    |--gap<5min--|
  [事件簇 A]        [事件簇 B]
  两个事件间隔超过 gap 就切分为不同 session

实验：观察 Watermark 行为

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
import org.apache.spark.sql.functions._
import java.sql.Timestamp

// 模拟带事件时间的数据
val data = spark.readStream
  .format("rate")  // 自动生成递增数据
  .option("rowsPerSecond", 10)
  .load()
  .withColumn("event_time",
    ($"timestamp".cast("long") - (rand() * 30).cast("long")).cast("timestamp"))

// 带 Watermark 的窗口聚合
val windowed = data
  .withWatermark("event_time", "10 seconds")
  .groupBy(window($"event_time", "10 seconds"))
  .count()

val query = windowed.writeStream
  .outputMode("update")
  .format("console")
  .option("truncate", false)
  .trigger(org.apache.spark.sql.streaming.Trigger.ProcessingTime("5 seconds"))
  .start()

// 观察输出中:
// 1. 窗口计数随 batch 递增
// 2. 旧窗口在 Watermark 推进后停止更新
// 3. State Store 中的状态数量（在 Spark UI Structured Streaming 页面查看）

在 Spark UI 的 Structured Streaming 页面可以看到 “Aggregation State” 指标，显示 State Store 中的 key 数量。随着 Watermark 推进，过期窗口被清理，key 数量保持稳定而不是无限增长。

模式提炼

Watermark 体现了延迟-完整性权衡的通用模式：系统必须在"等待更多数据以获得更准确的结果"和"尽快输出结果以降低延迟"之间做选择。Watermark 是这个权衡的显式参数化。

同样的权衡在分布式系统中反复出现。Google Cloud Dataflow 的 Watermark 机制是 Structured Streaming 的直接灵感来源。数据库的 MVCC 读取需要决定"读快照的时间戳"——等太久影响延迟，快照太旧可能读到不一致的数据。分布式一致性中的"最终一致"也是同一个权衡：允许临时不一致来换取更低的延迟。

工程迁移表

Spark 概念 Flink 对应 Kafka Streams 对应 数据库对应
Watermark Watermark（相同概念） Stream Time MVCC 读快照时间戳
事件时间窗口 EventTime Window TimeWindows 时间范围分区
State Store Keyed State (RocksDB) State Store (RocksDB/InMemory) 物化视图状态
Checkpoint Checkpoint（Chandy-Lamport） Changelog + Standby 数据库检查点
迟到数据丢弃 Late element 侧输出 Grace period 无直接对应

常见误解

误解一：“Watermark 保证不丢数据”。Watermark 定义了一个截止时间，它是事件时间延迟的单向保证，超过阈值的数据可能被丢弃。这是设计如此，不是 bug。如果业务不能容忍丢弃，需要加大 Watermark 阈值（代价是更高的延迟和更大的状态）。

误解二：“没有 Watermark 的聚合查询可以正常运行”。可以在支持的输出模式下运行，但不会仅因事件时间过去就清理全部状态。持续增加的 key 或窗口会使状态不断增长，固定且有限的 key 集合则不必然无限增长。窗口状态需要合适的清理边界；Complete 模式保留全部聚合结果，有界键域或其他超时策略则需要分别估算状态上限。

误解三：“State Store 的状态存在 HDFS 上”。状态的工作副本在 Executor 内存中（HashMap），HDFS 上只存检查点（快照 + delta）。状态提交仍需要持久化 I/O。上述内存工作副本描述适用于 HDFSBackedStateStore；RocksDBStateStore 使用本地 RocksDB 与检查点，不能统一解释为 JVM HashMap。

练习

  1. 对同一个流式查询分别设置 Watermark 阈值为 10 秒和 10 分钟，观察 State Store 中的 key 数量变化趋势。Watermark 越大，峰值 key 数量越多还是越少？

  2. 构造一个 session 窗口查询（Spark 3.2+），用 Rate Source 生成数据。在某个时刻暂停输入 30 秒再恢复，观察 session 窗口是否在暂停期间关闭。

  3. 查看 Checkpoint 目录的 state/ 子目录，找到快照文件和 delta 文件。统计 10 个 batch 后的文件数量。修改 spark.sql.streaming.stateStore.maintenanceInterval 后再观察。

系列导航

序号 主题 状态
00 导读：为什么 Spark 的核心是一张 DAG 已完成
01 RDD：弹性分布式数据集的五大属性 已完成
02 宽依赖与窄依赖：Stage 是怎样划分出来的 已完成
03 DAG Scheduler 与 Task Scheduler：从逻辑计划到物理执行 已完成
04 Shuffle 机制：数据跨分区交换的代价与优化 已完成
05 内存管理与 Tungsten：堆外内存、序列化与代码生成 已完成
06 存储体系：Block Manager、广播变量与累加器 已完成
07 容错机制：Lineage、Checkpoint 与推测执行 已完成
08 Spark SQL 与 Catalyst：从 SQL 文本到物理算子树 已完成
09 DataFrame 与 Dataset：类型安全与性能的折中 已完成
10 Adaptive Query Execution：运行时改写执行计划 已完成
11 数据源 API 与谓词下推：让存储层帮忙过滤 已完成
12 Structured Streaming：把流当成无界表 已完成
13 水位线与状态管理：流处理的时间语义 本篇
14 Spark on YARN / K8s：资源管理与部署模式 下一篇
15 动态资源分配与作业调度：多租户集群的资源博弈
16 性能诊断：Spark UI、Event Log 与调优思路
17 Spark 与 Flink：两种计算引擎的设计选择

参考资料