上一篇解决了数据源 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 作业执行：

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 必须能报告当前的 offset（标记读到了哪里），并能重放从某个 offset 开始的数据。

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

Sink 负责接收输出并保证 Exactly-Once 语义（配合 checkpoint）。每个 Sink 必须是幂等的——如果 batch 重试，写入相同的数据不会产生重复。

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

Checkpoint 与 Exactly-Once

Structured Streaming 通过 checkpoint 目录保证端到端的 Exactly-Once 语义：

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

恢复流程：查询重启后，从 checkpoint 读取最后一个已提交的 batch ID 和对应的 source offset，从该 offset 之后继续处理。如果上一个 batch 执行了但没提交（Executor 故障），重新执行该 batch（Source 重放 offset 范围内的数据），配合幂等 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 的无界聚合（内存会无限增长）、不支持 sort（无界数据无法排序）、不支持 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：两种计算引擎的设计选择

参考资料