上一篇解决了 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): 收到 3 条 → 但事件实际分布在不同时刻
  结果不准确——同一事件可能因为网络延迟被分到错误的窗口

用事件时间做窗口聚合:
  窗口 [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 的数据被丢弃
  → 窗口 [10:00, 10:05) 和 [10:05, 10:10) 可以关闭并输出

Batch 2: 一条事件时间 = 10:08 的数据到达
  10:08 < Watermark 10:12 → 丢弃

Watermark 只会向前推进，永远不会倒退。即使某个 batch 的数据事件时间很早，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 推进超过某个窗口的结束时间时，该窗口的状态可以安全清除。没有 Watermark 的聚合操作，状态会无限增长。

窗口类型

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 的状态永远不会被清理，最终耗尽 Executor 内存。生产环境中有状态查询必须配置 Watermark。

误解三：“State Store 的状态存在 HDFS 上”。状态的工作副本在 Executor 内存中（HashMap），HDFS 上只存检查点（快照 + delta）。正常执行时状态读写不涉及 HDFS I/O，只有 batch 结束时的检查点写入和故障恢复时的状态加载需要访问 HDFS。

练习

  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：两种计算引擎的设计选择

参考资料