上一篇解决了数据源 API 和谓词下推。这一篇进入流处理——Structured Streaming。

Spark 的流处理经历了两代：DStream（基于 RDD 的离散化流）和 Structured Streaming（基于 DataFrame/Dataset 的结构化流）。DStream 已经停止演进，Structured Streaming 是当前唯一推荐的流处理 API。

Structured Streaming 的核心模型可以用一句话概括：把流数据看成一张不断追加新行的无界表，复用 Spark SQL 的 Catalyst 优化器和 Tungsten 执行引擎来处理。

本文只抓一个问题：无界表模型如何工作，以及 MicroBatch 执行引擎如何把连续的流切成离散的批次。

无界表模型

1
2
3
4
5
6
7
8
传统流处理的心智模型:
  消息 → 处理函数 → 输出
  (一条一条处理，每条消息触发一次计算)

Structured Streaming 的心智模型:
  数据源不断追加行到一张"输入表"
  查询在整张表上持续运行
  结果写入一张"输出表"

用户写的查询代码和批处理完全相同：

1
2
3
4
5
6
7
8
9
10
11
12
13
val lines = spark.readStream
  .format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load()

val words = lines.as[String].flatMap(_.split(" "))
val wordCounts = words.groupBy("value").count()

val query = wordCounts.writeStream
  .outputMode("complete")
  .format("console")
  .start()

这段代码和批处理的唯一区别是 readStream/writeStream 替代了 read/write。groupBy.count 的逻辑完全相同。Structured Streaming 引擎负责把这个批查询"增量化"——每次只处理新到的数据，而不是重新扫描全部历史数据。

三种输出模式

输出模式（Output Mode）决定了每个 micro-batch 写入输出表的行为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
Append 模式:
  只输出新增的行（不会修改已输出的行）
  适用: 无聚合的流式 ETL、带 watermark 的窗口聚合
  限制: 不能用于无 watermark 的聚合（因为聚合结果可能随时变化）

Complete 模式:
  每次输出完整的聚合结果表
  适用: 全局聚合（如 wordcount 的全量结果）
  限制: 输出表不能太大（每次全量输出）

Update 模式:
  只输出本次 batch 中发生变化的行
  适用: 聚合结果的增量更新
  限制: 需要下游 sink 支持 upsert 语义

选择输出模式的经验法则：无聚合查询用 Append，全局聚合用 Complete，有 watermark 的窗口聚合用 Append 或 Update。

MicroBatch 执行引擎

Structured Streaming 的默认执行引擎是 MicroBatch。它把连续的输入流切成离散的小批次（micro-batch），每个批次触发所需的 Spark Job，复杂查询可能包含多个 Job：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
时间线:
  ──────┬──────┬──────┬──────┬──────→
        │ B-0  │ B-1  │ B-2  │ B-3
        │      │      │      │
        ▼      ▼      ▼      ▼
     处理新    处理新   处理新   处理新
     到的数据  到的数据 到的数据 到的数据

每个 Batch 的执行:
  1. 从 Source 获取新数据的 offset 范围 [startOffset, endOffset]
  2. 构建 DataFrame 查询计划（只包含新数据）
  3. 经过 Catalyst 优化
  4. 执行查询，结果写入 Sink
  5. 把 endOffset 提交为新的 checkpoint

触发间隔（Trigger）控制批次的生成频率：

1
2
3
4
5
6
7
8
9
10
11
// 固定间隔触发（每 10 秒一个 batch）
query.writeStream.trigger(Trigger.ProcessingTime("10 seconds"))

// 尽可能快地触发（上一个 batch 结束后立即开始下一个）
query.writeStream.trigger(Trigger.ProcessingTime(0))

// 只处理当前可用数据然后停止（用于测试和增量 ETL）
query.writeStream.trigger(Trigger.Once())

// Available-now: 处理当前所有可用数据（可能多个 batch），然后停止
query.writeStream.trigger(Trigger.AvailableNow())

MicroBatch 的延迟下限由批次间隔和执行时间决定。如果一个 batch 的执行时间是 2 秒，触发间隔设为 1 秒，实际延迟至少 2 秒（batch 执行完才能输出结果）。

Source 和 Sink 抽象

Source 负责提供数据并跟踪读取进度。可恢复的 Source 需要记录进度并支持重放；Socket Source 仅用于测试，不提供这样的端到端容错保证。

1
2
3
4
5
内置 Source:
  Kafka Source     → offset = topic-partition-offset 元组
  File Source      → offset = 已处理的文件列表
  Socket Source    → 无可靠 offset（仅用于测试）
  Rate Source      → 自动生成递增数据（用于测试和基准）

Sink 接收输出，交付语义由具体连接器及写入协议决定。可重放输入、可靠 checkpoint 和幂等或事务式输出共同决定端到端保证，不能仅凭使用了 Structured Streaming 就认定 exactly-once。

1
2
3
4
5
6
内置 Sink:
  Kafka Sink     → 写入 Kafka topic
  File Sink      → 写入 Parquet/JSON/CSV 文件
  Console Sink   → 打印到控制台（测试用）
  Memory Sink    → 写入内存表（测试用）
  ForeachBatch   → 用户自定义的批处理函数

图 1：微批恢复依赖输入重放、checkpoint 与输出提交

flowchart TB
  S[可重放 Source] --> B[执行 batch N]
  C[(Checkpoint：offset 与状态)] --> B
  B --> O[Sink 写入]
  O --> D{写出已提交但执行端未确认}
  D -->|重试同一 batch| B
  O --> I[输出端幂等键或提交协议]
  I --> R[控制重复写入效果]

图 1 中 checkpoint 能保存恢复所需进度，却无法让任意外部写入自动幂等。Kafka Sink 的保证是 at-least-once；foreachBatch 默认也是 at-least-once，可按 batchId 结合事务或去重协议实现需要的输出语义。

Checkpoint 与 Exactly-Once

Structured Streaming 在 checkpoint 目录记录恢复进度和状态；能否获得端到端 exactly-once 还取决于 Source 与 Sink：

1
2
3
4
5
6
Checkpoint 目录结构:
  /checkpoint/
  ├── commits/          已提交的 batch ID
  ├── offsets/           每个 batch 的 source offset 范围
  ├── sources/           source 的元数据
  └── state/             有状态操作的状态数据

恢复流程：查询重启后，结合 checkpoint 的 offset、commit 和状态记录确定要恢复的批次。已经写出但尚未确认提交的 batch 可能重放；支持幂等或事务提交的 Sink 才能将这种重执行收敛到一次输出效果。Kafka Sink 仍可能产生重复记录。

实验：Socket 流式 Wordcount

终端 1：启动 netcat 服务

1
nc -lk 9999

终端 2：在 spark-shell 中执行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
val lines = spark.readStream
  .format("socket")
  .option("host", "localhost")
  .option("port", 9999)
  .load()

val words = lines.as[String].flatMap(_.split(" "))
val counts = words.groupBy("value").count()

val query = counts.writeStream
  .outputMode("complete")
  .format("console")
  .trigger(org.apache.spark.sql.streaming.Trigger.ProcessingTime("5 seconds"))
  .start()

在 netcat 终端输入文本，每 5 秒可以在 spark-shell 的控制台看到更新后的 wordcount 结果。在 Spark UI 的 Structured Streaming 页面可以看到每个 batch 的输入行数、处理时间和延迟。

模式提炼

Structured Streaming 体现了批流统一的设计模式：用同一套 API 和引擎处理有界数据（批）和无界数据（流），流是批的推广，批是流的特例。

这个模式的优势是减少了概念和代码的重复。用户不需要为批处理和流处理各写一套逻辑，不需要维护两套引擎的配置和调优经验。

Flink 从另一个方向达到了同样的目标：Flink 把批看成有界流，Spark 把流看成追加的批。两种视角在实践中都工作得很好，区别在于默认的延迟特性——Flink 的逐条处理天然低延迟，Spark 的 micro-batch 延迟取决于批次间隔。

工程迁移表

Spark 概念 Flink 对应 Kafka Streams 对应 数据库对应
Structured Streaming Flink SQL / DataStream KStream / KTable 物化视图增量刷新
无界表模型 Dynamic Table Changelog Stream 持续查询（Continuous Query）
MicroBatch 引擎 无（逐条/mini-batch） 无（逐条拉取） 批量刷新物化视图
Source offset Checkpoint barrier Consumer offset WAL LSN
Output Mode (Append/Complete/Update) Changelog Mode (Insert/Upsert/Retract) 无直接对应 全量/增量物化视图
Checkpoint 目录 Checkpoint（Chandy-Lamport） Changelog topic 物化视图的检查点

常见误解

误解一：“Structured Streaming 是实时处理”。MicroBatch 的延迟下限是一个 batch 的执行时间（通常几百毫秒到几秒），不是真正的逐条实时处理。Continuous Processing 模式（实验性，Spark 2.3+）可以做到毫秒级延迟，但功能受限，不支持聚合和多数复杂操作。

误解二：“readStream 和 read 的查询可以互换”。大部分查询逻辑可以互换，但流查询对某些操作有限制：聚合能否运行取决于输出模式，无 watermark 的 Complete 模式聚合可以运行但会保留全部聚合状态；排序仅在聚合后的 Complete 模式等受支持条件下可用，limit 不支持。

误解三：“Exactly-Once 意味着每条数据只被处理一次”。Exactly-Once 保证的是输出效果等价于每条数据只被处理一次，不是执行过程中的每条数据物理上只被处理一次。Batch 重试时数据可能被重新处理，但通过幂等 Sink 和 offset 管理保证最终输出不重复。

练习

  1. 修改上面的 Socket Wordcount 示例，把 outputMode 从 “complete” 改为 “update”，观察输出行为的差异。再改为 “append”，观察是否报错以及原因。

  2. 给 Kafka Source 的 Structured Streaming 查询配置 checkpoint 目录，运行后 kill 并重启查询。验证是否从 checkpoint 恢复，是否出现重复计数。

  3. 在 Spark UI 的 Structured Streaming 页面中，找到 “Input Rate”、“Processing Rate” 和 “Batch Duration” 图表。如果 Processing Rate 持续低于 Input Rate，说明什么问题？应该如何应对？

系列导航

序号 主题 状态
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：两种计算引擎的设计选择

参考资料