4. Apache Spark Structured Streaming：流式DataFrame/Dataset API、微批处理与连续处理模式

好，咱们进入第四章。这一章我打算聊聊 Spark Structured Streaming，这是 Lakehouse 实时数据处理的核心引擎。说实话，我最早接触流处理的时候，用的还是 Spark Streaming（DStream 那套），后来 Structured Streaming 出来，我一开始还有点抵触——毕竟老 API 用顺手了。但真正上手之后，我发现这玩意儿的设计思路确实更优雅。

说白了，Structured Streaming 的核心思想就一句话：把流数据当成一张无限追加的表。你想想看，批处理是对一张静态表做查询，流处理就是对一张不断有新数据进来的动态表做查询。这个抽象一旦建立起来，很多问题就变得简单了。

4.1 流式DataFrame/Dataset API：从批到流的无缝切换

Structured Streaming 最让我喜欢的一点，就是它的 API 和批处理几乎一模一样。你在批处理里怎么用 DataFrame/Dataset，在流处理里就怎么用。我刚开始给团队推这套东西的时候，就告诉他们：你不需要学一套全新的 API，你只需要把 spark.read 换成 spark.readStream，把 spark.write 换成 spark.writeStream。

核心概念：流式 DataFrame 是 Unbounded Table（无界表）的抽象。每条新到的数据就是表里新加的一行。你对这个表做的所有操作（select、filter、groupBy、join），最终都会转换成对增量数据的连续计算。

来看一个最简单的例子。假设我们要从 Kafka 读取 JSON 格式的点击流数据：

// 读取流数据
val inputDF = spark
  .readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "click_events")
  .load()

// 解析 JSON
val parsedDF = inputDF
  .selectExpr("CAST(value AS STRING) as json")
  .select(from_json($"json", schema).as("data"))
  .select("data.user_id", "data.page", "data.timestamp")

// 聚合计算
val resultDF = parsedDF
  .withWatermark("timestamp", "10 minutes")
  .groupBy(window($"timestamp", "5 minutes"), $"page")
  .agg(count("*").as("click_count"))

// 输出到 Delta 表
val query = resultDF
  .writeStream
  .format("delta")
  .option("checkpointLocation", "/path/to/checkpoint")
  .table("click_analytics")

query.awaitTermination()

嗯，这里要注意几个关键点。第一，readStream 返回的是一个 DataStreamReader，它不会立即开始消费数据，而是定义了一个流式计算的 逻辑计划。第二，writeStream 才是真正触发计算的地方，它会启动一个持续运行的查询（Continuous Query）。

我的经验：在开发阶段，我习惯用 format("console") 来调试输出，这样可以在终端看到每条触发间隔的数据结果。但千万别在生产环境这么干——console sink 会把所有数据打印到日志里，磁盘很快就爆了。

4.2 微批处理模式：默认的可靠选择

微批处理（Micro-Batch Processing）是 Structured Streaming 的默认执行模式。它的工作原理很简单：把连续不断的流数据切成一个个小批次，每个批次作为一个 Spark 作业来执行。

我遇到过不少刚接触流处理的朋友，他们总觉得微批处理不够「实时」。其实这是个误解。微批处理的延迟通常在 100ms 到几秒之间，对于绝大多数业务场景来说完全够用。而且它的优势很明显：

  • Exactly-Once 语义：基于 checkpoint 和 WAL（Write-Ahead Log），保证每条数据恰好被处理一次
  • 容错性：任务失败后可以从上一个 checkpoint 恢复，不会丢数据
  • 背压处理：自动根据处理速度调整数据拉取速率，防止系统过载

微批处理的执行流程大致是这样的：

  1. 触发间隔：每隔 trigger(ProcessingTime("10 seconds")) 指定的时间，触发一次微批
  2. 数据拉取：从 source（Kafka、文件等）拉取当前批次的新数据
  3. 增量执行：对这批数据执行定义好的转换逻辑
  4. 结果输出：将计算结果写入 sink（Delta 表、Kafka 等）
  5. 提交 offset：记录当前批次处理到的位置，用于故障恢复

避坑指南：我曾经在一个项目里把触发间隔设成了 100ms，结果发现集群 CPU 全耗在任务调度上了，实际处理能力反而下降了。后来我把间隔调到 2 秒，吞吐量提升了 3 倍。记住：触发间隔不是越小越好，要结合数据量和处理逻辑来调优。

4.3 连续处理模式：真正的低延迟

如果你对延迟的要求特别苛刻——比如毫秒级的实时风控、交易监控——那微批处理可能就不够用了。这时候就需要 连续处理（Continuous Processing） 模式。

连续处理是 Spark 2.3 引入的实验性功能。它不再把数据切成批次，而是使用长运行的任务（Long-Running Tasks）来持续不断地处理数据。每个任务维护一个持久的连接，数据一到就立即处理。

对比维度 微批处理 连续处理
延迟 100ms ~ 几秒 ~1ms（亚毫秒级）
Exactly-Once ✅ 支持 ✅ 支持（部分 sink）
容错恢复 基于 checkpoint，成熟稳定 基于 offset 追踪，恢复较慢
支持的 Source Kafka、文件、Socket 等 仅 Kafka（目前）
支持的 Sink Delta、Kafka、Console 等 Kafka、Console、Memory
适用场景 大多数实时场景 超低延迟、高吞吐场景

启用连续处理很简单，只需要在 writeStream 时指定 trigger 类型：

val query = resultDF
  .writeStream
  .trigger(Trigger.Continuous("1 second"))  // 连续处理模式
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("topic", "output_topic")
  .option("checkpointLocation", "/path/to/checkpoint")
  .start()

注意这里的 Trigger.Continuous("1 second") 并不是说每隔 1 秒触发一次，而是指 checkpoint 的间隔。实际的数据处理是持续不断的。

我的建议：除非你的业务真的需要毫秒级延迟，否则我建议先用微批处理。连续处理虽然延迟低，但生态还不够完善，支持的 source/sink 有限，而且故障恢复的代价比微批处理大得多。我在生产环境只用过一次连续处理——那是一个高频交易的风控系统，延迟超过 5ms 就会造成损失。其他场景，微批处理完全够用。

4.4 两种模式的选择策略

怎么选？我一般按这个思路来判断：

  • 延迟要求 > 1秒：用微批处理，默认配置就行
  • 延迟要求 100ms ~ 1秒：微批处理 + 调优（减小触发间隔、优化序列化）
  • 延迟要求 < 10ms：考虑连续处理，但要评估 source/sink 是否支持
  • 数据量极大（百万级/秒）：微批处理更稳定，背压机制更成熟

下面这张图展示了两种模式在 Lakehouse 架构中的位置和关系：

Structured Streaming 执行模式对比 数据源（Kafka / 文件） 微批处理模式 连续处理模式 触发间隔（如 10s） 拉取增量数据 执行 Spark 作业 长运行任务持续处理 亚毫秒级延迟 仅支持 Kafka Delta Lake / Sink 微批处理：成熟稳定，Exactly-Once 连续处理：实验性，超低延迟

4.5 输出模式与容错机制

不管用哪种模式，输出模式（Output Mode）都是必须理解的概念。它决定了每个触发间隔后，计算结果如何写入 sink。

  • Append Mode：只输出新增的行。适用于 select、filter、map 等不会修改已有结果的操作
  • Update Mode：只输出被更新的行。适用于聚合操作，比如 groupBy + count，每次只输出 count 发生变化的分组
  • Complete Mode：每次输出全部结果。适用于小规模聚合，比如全局计数。数据量大的时候千万别用，否则每次输出都是全量数据

避坑指南：我见过有人用 Complete Mode 输出一个按用户 ID 聚合的统计结果，结果用户量一上来，每次输出几百万行，直接把下游 Kafka 打挂了。记住：Complete Mode 只适合结果集很小的场景，比如全局计数器、配置表等。

容错方面，Structured Streaming 的 checkpoint 机制是关键。它会记录三样东西：

  1. Offset 信息：每个 source 分区当前处理到的位置
  2. 状态数据：聚合操作的中间状态（比如 window 里的 count 值）
  3. 已提交批次：哪些批次已经成功写入 sink

有了这些信息，任务重启后就能精确地从上次中断的地方继续处理，不会丢数据，也不会重复处理。嗯，这就是 Exactly-Once 语义的保障。

最后说一句，Structured Streaming 的 API 设计确实很巧妙。它把流处理的复杂性封装在引擎内部，对外暴露的是一套和批处理几乎一致的接口。你不需要关心数据是怎么分片的、任务是怎么调度的、容错是怎么实现的——这些 Spark 都帮你搞定了。你只需要专注于业务逻辑本身。

← 上一章 📖 返回目录 下一章 →