Apache Spark Real-time Mode 深入解析：基于 Structured Streaming 的毫秒级低延迟流处理

• 大数据
• 数据分析
• 批处理
• 流处理
• 机器学习
• 图计算

【免费下载链接】spark

Apache Spark – A unified analytics engine for large-scale data processing

项目地址：
https://gitcode.com/gh_mirrors/sp/spark

点击查看 免费下载

Real-time Mode（实时模式）是 Apache Spark 4.1.0 引入的全新流式执行模式，它复用 Structured Streaming 的 API 与处理保证，将端到端延迟从微批模型约 100 ms 的下限压缩到毫秒级。本文以官方文档 real-time-mode.md 为主线，结合仓库源码（如 RealTimeModeAllowlist.scala、SQLConf.scala），完整讲解其工作原理、pipelined shuffle 机制、支持/不支持的查询范围、Trigger API、配置项与最佳实践，帮助你为欺诈检测、实时告警、实时个性化等运营型负载构建真正"数据一到即处理"的管道。

什么是 Real-time Mode

Real-time Mode 是 Spark 4.1.0 起新增的流式执行模式，面向必须对"数据抵达的瞬间"作出反应的运营型负载，典型场景包括：

• 欺诈检测（fraud detection）
• 实时告警（real-time alerting）
• 实时个性化（live personalization）

它的核心承诺是：使用 Structured Streaming 的 API 与处理保证，实现超低端到端延迟，但部分调度与回调细节与微批（micro-batch）执行不同。

在查询能力上，Real-time Mode 自 Spark 4.1.0 起支持无状态查询（投影、过滤等 map 类操作、union、流-静态表连接）；自 Spark 4.3.0 起支持有状态查询，包括基于 dropDuplicates 的流式去重、流式聚合（groupBy(…).agg(…)）以及 JVM（Scala/Java）transformWithState 算子；自 Spark 4.4.0 起新增支持 dropDuplicatesWithinWatermark。这些有状态操作需要 shuffle，Real-time Mode 以 pipelined shuffle（流水线化 shuffle） 方式执行，使记录仍然可以持续流经而不必等待批边界（详见下文 有状态查询如何工作）。其他有状态操作（如流-流连接、flatMapGroupsWithState）尚不支持。

最需要记住的一点：你传给 trigger 的时长（默认 5 分钟）本质上是检查点间隔（checkpoint interval），而不是输入驱动输出的延迟目标。记录可以被持续处理并连续发出，而不是等待批边界；但批作用域（batch-scoped）的效果，如 watermark 推进和空闲分区上的定时器，仍可能等待该边界。

启用方式非常简洁：在流式写入（writeStream）上设置一个 Real-time trigger，查询的其他部分完全不用改动。

工作原理：从微批到长运行任务

默认情况下，Structured Streaming 把查询作为一系列小批作业执行——即微批模型。每个微批中，driver 规划该批并启动一批短命任务；这些任务读取并处理输入的一个有界切片，driver 提交进度后再规划下一批。每个批固定的规划（planning）与任务调度（task-scheduling）开销给端到端延迟设定了下限（约 100 ms）。

Real-time Mode 通过启动长运行任务（long-running tasks）——每个输入分区一个——来消除这种每批开销。这些任务在一个（长的）批的持续时间内保持存活，并随记录到达而持续处理。由于任务每个批只调度一次（而不是每个数据切片调度一次），记录会流经算子流水线（source → transformations → sink）而无需等待批边界。端到端延迟从微批约 100 ms 的下限下降到大致等于处理并投递一条记录所需的时间（通常只有几毫秒）。

因为记录从不等待批边界，批时长主要控制查询多久做一次检查点进度提交——这正是下一节要解释的内容。

有状态查询如何工作：Pipelined Shuffle 与 Streaming Shuffle

无状态操作是逐记录执行的：每个长运行任务读取一个分区、变换记录并直接投递出去，永远不需要其他分区的数据。有状态操作则不同。流式聚合、dropDuplicates、dropDuplicatesWithinWatermark 或 transformWithState 都按键分组记录，因此给定键的所有记录必须到达同一个任务，无论它到达于哪个分区。

在微批引擎中，这种重新分组通过 shuffle 完成：批的生产者阶段写 shuffle 文件，只有这些文件被完整物化后，消费者阶段才按键分组读回。"先物化、再读取"的边界正是 Real-time Mode 要避免的延迟来源；而且长运行的 Real-time 任务永远不结束它的批，所以普通 shuffle 会死锁——消费者会永远等待一个永不完成的生产者。

因此 Real-time Mode 用两个协同部件以不同方式运行 shuffle：

• Pipelined shuffle 改变的是"调度"。正常情况下调度器只在生产者阶段完成后才运行消费者阶段。对 Real-time 有状态查询，生产者（source 扫描）与消费者（有状态算子）被标记为同一个 pipelined group（流水线组），并同时调度，贯穿整个批。两个阶段谁都不等对方结束，它们像无状态场景一样作为长运行任务并发执行。
• Streaming shuffle 改变的是"数据传输"。因为两个阶段并发运行，shuffle 数据无法先写文件、事后再读。取而代之，生产者的输出逐条记录通过网络直接流式传给消费者任务，记录一经产生就被按键重新分组并交给有状态算子——没有中间文件，也不等待批结束。

两者配合，让有状态 Real-time 查询保持与无状态查询一致的连续逐记录流动：一条记录被读取，通过 streaming shuffle 路由到持有其键的任务，合并进状态，再发出——全程没有批边界。重新分组后的状态保存在 Spark 常规的 state store 中，每个批做一次检查点，因此 exactly-once 与恢复保证与微批引擎相同（见 容错与交付语义）。

这一机制在 Real-time Mode 中自动启用，无需任何配置。它适用于流式路径上的每一个 shuffle——有状态算子所需的 shuffle，以及单纯的 repartition（哈希或 round-robin shuffle、不带任何有状态算子），后者以同样的方式作为 pipelined shuffle 运行。唯一的例外是需要单独预备作业的 shuffle：range 分区（repartitionByRange，或规划为 range shuffle 的 ORDER BY）需要一个采样作业来计算 range 边界，而该作业在 source 持续产出的情况下无法完成，因此这类查询会以 STREAMING_REAL_TIME_MODE.OPERATOR_OR_SINK_NOT_IN_ALLOWLIST 启动失败。

从源码看，这一点在 allowlist 的检查逻辑中有明确体现：RealTimeModeAllowlist.scala 中对 ShuffleExchangeExec 节点额外校验 supportsPartitioning，直接拒绝 RangePartitioning——注释指出构建 RangePartitioner 会运行一个独立的采样作业来计算 range 边界，而该作业无法在 source 持续产出时完成，因此在此处快速失败，而不是让查询卡在一个永不结束的采样作业上。

transformWithState 在 Real-time Mode 中的执行语义

自 Spark 4.3.0 起，JVM transformWithState API 可以在 Real-time Mode 中运行。Scala 与 Java API 均受支持；PySpark 的 transformWithState 与 transformWithStateInPandas API 不受支持。 TimeMode.None、TimeMode.ProcessingTime、TimeMode.EventTime 三种模式均支持。完整的状态处理器 API（状态变量、定时器、TTL、初始状态）参见 transformWithState 指南。TimeMode.EventTime 要求用 withWatermark 声明输入的事件时间 watermark。

API 在微批与 Real-time 两种模式下相同，但输入回调的粒度不同：

• 微批模式：一次 handleInputRows 调用会收到该批中某个分组键的全部输入行；
• Real-time 模式：Spark 为每条非迟到（non-late）输入行调用一次 handleInputRows，迭代器中只有单行。

因此，同时用于两种模式的处理器，必须既能正确处理同一键的多行在一次调用中到达，也能正确处理它们在多次调用中分别到达。

基于时间的操作在长 Real-time 批保持打开期间也会增量执行：

• 处理时间定时器（Processing-time timers） 使用长运行输入批期间的当前 executor 时钟。Spark 会在每条输入行到达状态分区后、以及批完成时各检查一次过期定时器。
• 事件时间定时器（Event-time timers） 要求用 withWatermark 声明输入 watermark，并使用批开始时建立的 watermark。该 watermark 在整个批内保持固定，仅在批与批之间推进。针对它已过期的定时器会在每条输入行后以及批完成时各检查一次。
• TTL 状态 要求 TimeMode.ProcessingTime。过期值在状态被访问时不会返回。Spark 还会在处理输入行的过程中周期性移除过期值，并在批完成时做最终清理。

需要特别说明：定时器检查与 TTL 清理由输入或批完成驱动，没有独立的后台轮询。空闲分区上的处理时间定时器可能要等到另一条行到达或当前批完成才会触发。同样，事件时间 watermark 的推进也受 Real-time 批时长的约束。

当提供了初始状态（initial state）时，Spark 会在启动第一个长运行 Real-time 输入批之前，用一个**有界的引导批（bootstrap batch）**加载并提交它。已经可用的输入会等到初始状态持久化后才被处理。引导批使用普通 shuffle；pipelined shuffle 从随后的输入批开始。TimerValues.getCurrentProcessingTimeInMs() 在 handleInitialState 运行期间返回有界引导批的时间戳；长运行输入批开始后则使用实时的 executor 时钟。

状态变量、TTL 信息与已注册的定时器都会在重启后从检查点恢复。当使用 RocksDB 与 state-store 检查点格式 v2 时，查询可以在微批或 Real-time 模式下恢复同一个兼容检查点（见 State store 软默认值）。

Batch Duration 是 Checkpoint Interval 而非延迟目标

在 Real-time Mode 中，批时长本质上是检查点间隔，而不是输入驱动输出的延迟间隔。使用默认 5 分钟时长时，查询仍可以在几毫秒内发出由输入记录产生的结果。该时长控制的是它多久提交一次进度并启动下一个长运行批；它还可以约束批作用域工作的延迟，包括 watermark 推进与空闲分区上的处理时间定时器。这与微批引擎不同——在微批中，所有输出都要等待批间隔。

注意不要混淆两个概念：

• 5 分钟的默认 trigger 时长：当你未指定时长时使用的检查点节奏；
• 5 秒的允许最小时长：你被允许设置的最小时长（见 启动硬性要求）。

选择批时长是一个权衡：

• 更短的批时长：检查点更频繁，恢复粒度更细（失败后需要重处理的工作更少）。但查询在提交进度、启动下一批的间隙不处理数据，检查点过于频繁会增加这些间隙，从而抬高尾部（p99）延迟，同时带来更多规划与提交开销。
• 更长的批时长：减少上述开销与间隙，代价是恢复粒度更粗（失败后重处理的数据更多）。

时长设置在 Real-time trigger 上（见 启用 Real-time Mode）。

对无状态 Real-time 查询，进度使用异步进度跟踪（asynchronous progress tracking）提交：offset 与 commit 日志在记录处理路径之外写入，使检查点不会阻塞处理。它会在无状态 Real-time 查询中自动启用，并且每个批都会做检查点（Real-time Mode 中异步进度跟踪的检查点间隔固定为 0）。可以通过 asyncProgressTrackingEnabled writer option 关闭它。有状态查询（包括 transformWithState）不支持异步进度跟踪，进度是同步提交的。

仓库中的内部配置印证了这一行为：SQLConf.scala 定义了 spark.sql.streaming.realTimeMode.asyncProgressTrackingByDefault.enabled（内部配置，默认 true，自 4.3.0 起），即无状态查询默认启用异步进度跟踪；对应的执行机制实现见 AsyncProgressTrackingMicroBatchExecution.scala。

与其他执行模式对比

下表概括了 Real-time Mode 与默认微批引擎、以及实验性的 Continuous Processing 模式的关系：

上表中的 Processing Guarantees 列指处理语义（定义见 容错与交付语义）；端到端投递还取决于 sink，与执行模式无关。

Real-time Mode 与 Continuous Processing 都瞄准毫秒级延迟，但差异显著：

• Continuous Processing（Spark 2.3 引入）至今仍是实验性的。它只支持 map 类操作（投影与选择），不支持聚合、连接等有状态操作，并且只提供 at-least-once 保证。由于它无状态，容错与交付语义 中讨论的 exactly-once 处理 保证对它并不适用。这些限制制约了它的采用。
• Real-time Mode 的设计目标是支持所有查询形态（包括有状态操作），同时复用 Spark 成熟组件——状态管理、Catalyst 优化器、现有 SQL 算子。它提供 exactly-once 处理语义。它支持无状态查询，自 4.3.0 起支持 dropDuplicates、有状态聚合与 JVM transformWithState，自 4.4.0 起支持 dropDuplicatesWithinWatermark；其余有状态操作的支持仍在进行中。

对于新的低延迟负载，优先选择 Real-time Mode 而非 Continuous Processing。

启用 Real-time Mode

要让受支持的查询运行在 Real-time Mode，只需在流式写入上设置 Real-time trigger，查询的其他部分保持不变。例如，下面这个查询从 Kafka topic 读取、做无状态变换、再写入另一个 Kafka topic——即使 trigger 是 5 分钟，记录也会以低延迟流过：

Python

spark \\
.readStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("subscribe", "input-topic") \\
.load() \\
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \\
.writeStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("topic", "output-topic") \\
.option("checkpointLocation", "/path/to/checkpoint") \\
.outputMode("update") \\
.trigger(realTime="5 minutes") \\
.start()

Scala

import org.apache.spark.sql.streaming.Trigger

spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes")) // enable Real-time Mode
.start()

Java

import org.apache.spark.sql.streaming.Trigger;

spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.writeStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes")) // enable Real-time Mode
.start();

Trigger API

• Scala 与 Java：trigger 为 Trigger.RealTime(…)，从 org.apache.spark.sql.streaming.Trigger 导入，有几种形式：
  • Trigger.RealTime()——使用默认批时长 5 分钟。
  • Trigger.RealTime("5 minutes")——时长字符串。
  • Trigger.RealTime(300000)——以毫秒为单位的批时长（long）。
  • Trigger.RealTime(5, TimeUnit.MINUTES)——数值配合 java.util.concurrent.TimeUnit。
  • Trigger.RealTime(Duration("10 seconds"))——Scala 的 scala.concurrent.duration.Duration。
• Python：把批时长以字符串形式传给 trigger() 的 realTime 关键字参数，例如 .trigger(realTime="5 minutes")。Python 中该时长是必填的。

启动硬性要求

查询在 Real-time Mode 中启动前必须满足以下全部条件，每个条件都在查询启动时校验：

在源码层面，5 秒默认值与 allowlist 开关均在 SQLConf.scala 中定义：spark.sql.streaming.realTimeMode.minBatchDuration（默认 5000 ms，自 4.1.0 起）与 spark.sql.streaming.realTimeMode.allowlistCheck（默认 true，自 4.1.0 起）。

支持的查询

Real-time Mode 支持无状态的 map 类查询，以及首批有状态查询：去重、聚合与 JVM transformWithState。

以下操作、source 与 sink 均受支持：

无状态操作（自 Spark 4.1.0 起）

• 投影：select、selectExpr、withColumn、drop，以及类型化的 Dataset map / flatMap 操作。
• 选择：where / filter。
• 可编译为投影的表达式——包括 from_json / to_json 等函数以及标量用户自定义函数（UDF）。
• 列生成器（column generators），如 explode。
• union 两个或更多互不相同（distinct）的流式 source。多次引用同一个 source DataFrame 不受支持，会以 STREAMING_REAL_TIME_MODE.IDENTICAL_SOURCES_IN_UNION_NOT_SUPPORTED 失败；请为每个 source 单独创建 DataFrame。union 可以馈给有状态算子，但有状态算子不能出现在 union 之前的输入分支上；这种形态会以 STREAMING_REAL_TIME_MODE.STATEFUL_OPERATORS_BEFORE_UNION_NOT_SUPPORTED 失败。
• 流-静态表连接（stream-static join）：流式 DataFrame 与静态 DataFrame 连接。静态侧必须广播（使用 broadcast(…) hint），因为流-静态连接不得引入 shuffle。

有状态操作（自 Spark 4.3.0 起）

这些操作通过 pipelined shuffle 按键重新分组记录（见 有状态查询如何工作），并把状态保存在 state store 中，每批做检查点：

• 去重：dropDuplicates 自 Spark 4.3.0 起支持。dropDuplicatesWithinWatermark 自 Spark 4.4.0 起支持；它要求 withWatermark，并依据事件时间 watermark 限制状态保留时长。
• 流式聚合：groupBy(…).agg(…)（以及等价的 SQL GROUP BY），包括使用 window(…) 的窗口聚合。count(distinct …) 等 distinct 聚合不受支持（见 暂不支持的操作）。
• JVM transformWithState：Scala 与 Java API，包括 value、list、map 状态；处理时间 TTL；处理时间与事件时间定时器；可选的初始状态；输出事件时间列。增量执行与时间语义详见 transformWithState 在 Real-time Mode 中的执行语义。
• withWatermark（事件时间 watermark 声明）受支持且现在真正生效：它让窗口聚合丢弃迟到输入并驱逐已关闭窗口的状态，从而限制长运行查询累积的状态量。（Real-time Mode 始终运行在 update 输出模式——见 启动硬性要求——所以窗口聚合在窗口每次变化时发出其运行结果，watermark 决定该窗口状态何时被驱逐。）

支持的 Source 与 Sink

• Source：source 必须支持 Real-time Mode。在 Apache Spark 中，Kafka source 支持 Real-time Mode。不支持的 source 会以 STREAMING_REAL_TIME_MODE.INPUT_STREAM_NOT_SUPPORTED 失败。（内置的 rate source 不作为 Real-time source 支持。）
• Sink：
  • Kafka sink。
  • Foreach sink（通过 ForeachWriter），用于逐条记录写入任意外部系统。参见 使用 Foreach。注意 foreachBatch 不受支持，因为它按整批处理而非逐条记录。
  • Console 与 memory sink，适用于开发与调试。

其他 sink 会以 STREAMING_REAL_TIME_MODE.SINK_NOT_SUPPORTED 失败。

Real-time 查询使用的算子与 sink 会在启动前对照 allowlist 检查；不在 allowlist 内的任何内容都会以 STREAMING_REAL_TIME_MODE.OPERATOR_OR_SINK_NOT_IN_ALLOWLIST 失败。

从源码看，allowlist 是一份精确的清单：RealTimeModeAllowlist.scala 中 allowedSinks 包含 ConsoleTable、ContinuousMemorySink、ForeachWriterTable 与 KafkaTable；allowedOperators 包含 ProjectExec、FilterExec、GenerateExec、UnionExec、BroadcastHashJoinExec、ShuffleExchangeExec（需通过分区检查）、StatefulStreamlineAggregateExec、StreamingDeduplicateExec、TransformWithStateExec、EventTimeWatermarkExec 等。这与文档中的支持矩阵一一对应。

暂不支持的操作

以下内容在 Real-time Mode 中尚不支持。除非特别说明，使用了它们的查询会以 STREAMING_REAL_TIME_MODE.OPERATOR_OR_SINK_NOT_IN_ALLOWLIST 启动失败：

• 除上述之外的有状态操作：流-流连接（stream-stream joins）、flatMapGroupsWithState 与 会话窗口聚合（session-window aggregation）。
• PySpark 的 transformWithState 与 transformWithStateInPandas API。Real-time Mode 目前只支持 JVM（Scala/Java）transformWithState 实现。
• Range 分区：repartitionByRange，或规划为 range shuffle 的 ORDER BY / sort，因为计算 range 边界需要单独的采样作业，而该作业在 source 持续产出的情况下无法完成。（普通的 repartition——哈希或 round-robin——是支持的，它以 pipelined shuffle 运行，见 有状态查询如何工作。）

Distinct 聚合（如 count(distinct …)）也不受支持，但这是 Structured Streaming 的通用限制而非 Real-time Mode 特有：任何流式 distinct 聚合都会在分析阶段被拒绝（提示使用 approx_count_distinct()），与 trigger 无关。

更多有状态操作的支持正在持续进行中。

容错与交付语义

Real-time Mode 提供与默认微批引擎相同的 exactly-once 处理保证。需要区分两个不同的保证：

• Exactly-once 处理（processing）：每条输入记录对引擎管理状态（例如聚合计数）的影响，即使在失败与重启后也只生效一次。
• 投递语义（delivery semantics）：一条记录是否可能被多次写入外部系统。这是 sink 的属性，与执行模式无关。

Real-time Mode 在处理层面是 exactly-once 的。端到端投递取决于 sink：执行幂等或事务写入的 sink 可以实现 exactly-once，其他 sink 则是 at-least-once（失败后可能出现重复）。内置的 Kafka sink 提供 at-least-once 投递，无论是否使用 Real-time Mode。Real-time Mode 目前尚未随附 exactly-once sink，但可以实现一个。

内部实现上，offset 在每个批的末尾提交，且此时对应记录已经写入 sink。如果查询在批中途失败，重启时会从最后已提交的 offset 恢复，并可能重写失败前已发出的记录。在 exactly-once 输出重要时，请设计能够容忍重复的 sink——例如幂等写入。

完整示例

以下示例都从 Kafka 读取，并假定有一个正在运行的 Kafka 集群。多数示例展示了 Python、Scala、Java 三种写法。transformWithState 示例以 Scala 展示，同样适用于 Java；其 PySpark API 在 Real-time Mode 中尚不受支持。

流-静态表连接（Stream-static join）

用静态参考数据集为流增补信息。静态侧包在 broadcast(…) 中，使连接作为广播（map 侧）连接执行，从而避免 shuffle。

Python

from pyspark.sql.functions import broadcast

# Static reference data, read once as a batch DataFrame.
reference = spark.read.format("parquet").load("/path/to/reference")

spark \\
.readStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("subscribe", "input-topic") \\
.load() \\
.selectExpr("CAST(key AS STRING) AS joinKey", "CAST(value AS STRING) AS value") \\
.join(broadcast(reference), "joinKey") \\
.writeStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("topic", "output-topic") \\
.option("checkpointLocation", "/path/to/checkpoint") \\
.outputMode("update") \\
.trigger(realTime="5 minutes") \\
.start()

Scala

import org.apache.spark.sql.functions.broadcast
import org.apache.spark.sql.streaming.Trigger

// Static reference data, read once as a batch DataFrame.
val reference = spark.read.format("parquet").load("/path/to/reference")

spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr("CAST(key AS STRING) AS joinKey", "CAST(value AS STRING) AS value")
.join(broadcast(reference), "joinKey")
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes"))
.start()

Java

import static org.apache.spark.sql.functions.broadcast;
import org.apache.spark.sql.streaming.Trigger;

// Static reference data, read once as a batch DataFrame.
Dataset<Row> reference = spark.read().format("parquet").load("/path/to/reference");

spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr("CAST(key AS STRING) AS joinKey", "CAST(value AS STRING) AS value")
.join(broadcast(reference), "joinKey")
.writeStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes"))
.start();

去重

Real-time Mode 支持两种去重操作。两者都通过 pipelined shuffle 按去重键重新分组记录，并把已见过的键保存在 state store 中（见 有状态查询如何工作）。

dropDuplicates（Real-time Mode 支持自 Spark 4.3.0 起）

这个例子在不使用 watermark 的情况下调用 dropDuplicates，因此它会为查询的整个生命周期保留每个不同的 id。当事件时间列包含在去重列中时，dropDuplicates 可以使用 watermark 来限制状态。当事件时间应该限制状态但又不属于去重键时，请使用下面的 dropDuplicatesWithinWatermark。

Python

spark \\
.readStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("subscribe", "input-topic") \\
.load() \\
.selectExpr("CAST(key AS STRING) AS id", "CAST(value AS STRING) AS value") \\
.dropDuplicates(["id"]) \\
.writeStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("topic", "output-topic") \\
.option("checkpointLocation", "/path/to/checkpoint") \\
.outputMode("update") \\
.trigger(realTime="5 minutes") \\
.start()

Scala

import org.apache.spark.sql.streaming.Trigger

spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr("CAST(key AS STRING) AS id", "CAST(value AS STRING) AS value")
.dropDuplicates("id")
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes"))
.start()

Java

import org.apache.spark.sql.streaming.Trigger;

spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr("CAST(key AS STRING) AS id", "CAST(value AS STRING) AS value")
.dropDuplicates("id")
.writeStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes"))
.start();

dropDuplicatesWithinWatermark（Real-time Mode 支持自 Spark 4.4.0 起）

结合事件时间 watermark 使用 dropDuplicatesWithinWatermark，可以在不把事件时间纳入去重键的前提下限制去重状态。这个例子把 Kafka 的记录时间戳别名为 eventTime，并设置 10 分钟的 watermark 延迟阈值。

Python

spark \\
.readStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("subscribe", "input-topic") \\
.load() \\
.selectExpr(
"CAST(key AS STRING) AS id",
"CAST(value AS STRING) AS value",
"timestamp AS eventTime") \\
.withWatermark("eventTime", "10 minutes") \\
.dropDuplicatesWithinWatermark(["id"]) \\
.writeStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("topic", "output-topic") \\
.option("checkpointLocation", "/path/to/checkpoint") \\
.outputMode("update") \\
.trigger(realTime="5 minutes") \\
.start()

Scala

import org.apache.spark.sql.streaming.Trigger

spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr(
"CAST(key AS STRING) AS id",
"CAST(value AS STRING) AS value",
"timestamp AS eventTime")
.withWatermark("eventTime", "10 minutes")
.dropDuplicatesWithinWatermark("id")
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes"))
.start()

Java

import org.apache.spark.sql.streaming.Trigger;

spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr(
"CAST(key AS STRING) AS id",
"CAST(value AS STRING) AS value",
"timestamp AS eventTime")
.withWatermark("eventTime", "10 minutes")
.dropDuplicatesWithinWatermark("id")
.writeStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes"))
.start();

通过去重的记录会在处理时立即发出；而 watermark 推进与状态驱逐在 Real-time Mode 的批边界才生效。因此 trigger 时长决定的是：watermark 越过某个键的过期点之后，清理工作多久有机会移除该状态。

流式聚合

为每个键维护一个运行中的聚合值。Real-time Mode 通过 pipelined shuffle 按分组键重新分组输入，把每条记录合并进 state store 中的运行聚合值，并发出更新后的结果。由于输出模式是 update，每个键在变化时就会被发出，而不是只在批结束时发出。

Python

from pyspark.sql.functions import count

spark \\
.readStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("subscribe", "input-topic") \\
.load() \\
.selectExpr("CAST(key AS STRING) AS id") \\
.groupBy("id") \\
.agg(count("*").alias("cnt")) \\
.selectExpr("CAST(id AS STRING) AS key", "CAST(cnt AS STRING) AS value") \\
.writeStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("topic", "output-topic") \\
.option("checkpointLocation", "/path/to/checkpoint") \\
.outputMode("update") \\
.trigger(realTime="5 minutes") \\
.start()

Scala

import org.apache.spark.sql.functions.count
import org.apache.spark.sql.streaming.Trigger

spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr("CAST(key AS STRING) AS id")
.groupBy("id")
.agg(count("*").as("cnt"))
.selectExpr("CAST(id AS STRING) AS key", "CAST(cnt AS STRING) AS value")
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes"))
.start()

Java

import static org.apache.spark.sql.functions.count;
import org.apache.spark.sql.streaming.Trigger;

spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr("CAST(key AS STRING) AS id")
.groupBy("id")
.agg(count("*").as("cnt"))
.selectExpr("CAST(id AS STRING) AS key", "CAST(cnt AS STRING) AS value")
.writeStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes"))
.start();

JVM transformWithState

在 Real-time trigger 下持续运行一个 JVM 有状态处理器。这个处理器为每个 Kafka 键维护一个运行总和。它用 sum 累加迭代器，因此同一份实现既能处理微批调用中多行的情形，也能处理 Real-time 调用中单行的情形。完整的 API 参见 transformWithState 指南。Java 应用使用相同的 TimeMode、OutputMode 与 Trigger.RealTime 设置；Java 的 transformWithState 重载还额外接受一个输出 encoder。

import org.apache.spark.sql.Encoders
import org.apache.spark.sql.streaming.{
OutputMode, StatefulProcessor, TTLConfig, TimeMode, TimerValues, Trigger, ValueState}

import spark.implicits._

class RunningTotalProcessor
extends StatefulProcessor[String, (String, Int), String] {
@transient private var total: ValueState[Long] = _

override def init(outputMode: OutputMode, timeMode: TimeMode): Unit = {
total = getHandle.getValueState("total", Encoders.scalaLong, TTLConfig.NONE)
}

override def handleInputRows(
key: String,
inputRows: Iterator[(String, Int)],
timerValues: TimerValues): Iterator[String] = {
val previous = if (total.exists()) total.get() else 0L
val updated = previous + inputRows.map(_._2.toLong).sum
total.update(updated)
Iterator.single(s"$key,$updated")
}
}

val totals = spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr(
"CAST(key AS STRING) AS id",
"CAST(CAST(value AS STRING) AS INT) AS delta")
.as[(String, Int)]
.groupByKey(_._1)
.transformWithState(
statefulProcessor = new RunningTotalProcessor,
timeMode = TimeMode.ProcessingTime(),
outputMode = OutputMode.Update())

totals
.writeStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("topic", "output-topic")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("5 minutes"))
.start()

开发调试：写入 console

console sink 把输出打印到 driver 的标准输出，非常适合开发查询时使用。注意：console sink 会缓冲每个批的行，并在批提交时打印，因此它的输出是每批一次——这里是每 30 秒一次——而不是连续的。这使它便于检查结果，但它并不能反映 Real-time Mode 真实的逐记录延迟；要观察逐记录延迟，请使用 Kafka 这类逐行 sink。更短的批时长只是让 console 刷新更频繁。

Python

spark \\
.readStream \\
.format("kafka") \\
.option("kafka.bootstrap.servers", "host1:port1,host2:port2") \\
.option("subscribe", "input-topic") \\
.load() \\
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \\
.where("value IS NOT NULL") \\
.writeStream \\
.format("console") \\
.option("checkpointLocation", "/path/to/checkpoint") \\
.outputMode("update") \\
.trigger(realTime="30 seconds") \\
.start()

Scala

import org.apache.spark.sql.streaming.Trigger

spark
.readStream
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.where("value IS NOT NULL")
.writeStream
.format("console")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("30 seconds"))
.start()

Java

import org.apache.spark.sql.streaming.Trigger;

spark
.readStream()
.format("kafka")
.option("kafka.bootstrap.servers", "host1:port1,host2:port2")
.option("subscribe", "input-topic")
.load()
.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
.where("value IS NOT NULL")
.writeStream()
.format("console")
.option("checkpointLocation", "/path/to/checkpoint")
.outputMode("update")
.trigger(Trigger.RealTime("30 seconds"))
.start();

配置项详解

核心配置

这三个配置的定义与默认值都能在 SQLConf.scala、SQLConf.scala 中找到对应条目。此外，仓库中还定义了另两个内部配置：spark.sql.streaming.realTimeMode.transformWithState.ttlEvictionIntervalMs（内部，默认 1000 ms，自 4.3.0 起，用于 JVM transformWithState 在 Real-time 模式下 TTL 驱逐的阈值间隔）与 spark.sql.streaming.realTimeMode.asyncProgressTrackingByDefault.enabled（内部，默认 true，自 4.3.0 起），后者对应无状态查询默认启用异步进度跟踪。

State store 软默认值

有状态的 Real-time 查询需要一个低延迟且恢复正确（recovery-correct）的 state store 配置。因为该配置对整个引擎而言并不是正确的默认值，Real-time Mode 只在查询启动时、仅对 Real-time 查询自动应用它。这些是软默认值（soft defaults）：每个配置只在你没有自行设置时才生效，因此显式设置的值会被保留——少数与 Real-time Mode 不兼容的显式值除外，它们会在查询启动时被拒绝而不是被保留（见 不兼容配置）。

Real-time 查询必须使用 state-store 检查点格式 v2。以格式 v1 启动 Real-time 查询——例如把现有微批查询切换到 Real-time Mode，或当 spark.sql.streaming.stateStore.checkpointFormatVersion 被固定为 1 时——会以 STREAMING_REAL_TIME_MODE.CHECKPOINT_FORMAT_V1_NOT_SUPPORTED 启动失败。请使用全新的检查点位置；或者，在接受失败时可能丢失状态的前提下，设置 spark.sql.streaming.realTimeMode.dangerouslyAllowCheckpointV1.enabled=true。

不兼容配置

上述默认值只在你没有设置对应配置时应用。如果你把以下某项设置为 Real-time Mode 无法运行的值，查询会以 STREAMING_REAL_TIME_MODE.SQL_CONFIGURATION_NOT_SUPPORTED 启动失败，而不是静默覆盖你的值：

• spark.sql.streaming.stateStore.checkpointFormatVersion 设置为低于 2（除非 spark.sql.streaming.realTimeMode.dangerouslyAllowCheckpointV1.enabled=true）。
• spark.sql.streaming.stateStore.providerClass 设置为 RocksDBStateStoreProvider 以外的 provider。
• spark.sql.execution.sortBeforeRepartition 设置为 true。

最佳实践

注意事项（Caveats）

• Real-time Mode 提供 exactly-once 处理语义，但 sink 在失败后可能收到重复记录。如何为 exactly-once 输出设计 sink 见 容错与交付语义。
• Adaptive Query Execution（AQE）不支持 Real-time Mode 查询。

结语

Real-time Mode 把 Structured Streaming 从"百毫秒级微批"推进到"毫秒级连续处理"：通过每输入分区一个长运行任务消除逐批规划与调度开销，通过 pipelined shuffle + streaming shuffle 让有状态查询也能无批边界地按记录流动，同时完整保留 Catalyst 优化、state store 管理、exactly-once 处理与检查点恢复等 Spark 成熟能力。启动一个 Real-time 查询只需要在 writeStream 上设置 Trigger.RealTime(…)（Python 为 trigger(realTime=…)），并满足输出模式为 update、提供 checkpointLocation、批时长不小于 5 秒等硬性要求。结合本文的源码佐证与完整示例，你可以在当前仓库的 real-time-mode.md 基础上，为欺诈检测、实时告警、实时个性化等场景搭建真正低延迟的流处理管道。

赞

分享

• 大数据
• 数据分析
• 批处理
• 流处理
• 机器学习
• 图计算

【免费下载链接】spark

Apache Spark – A unified analytics engine for large-scale data processing

项目地址：
https://gitcode.com/gh_mirrors/sp/spark

点击查看 免费下载

相关推荐

• claude-skills Spark Engineer 参考手册解读：Apache Spark Structured Streaming 流处理实战模式全解析

  claude skills Spark Engineer 参考手册解读：Apache Spark Structured Streaming 流处理实战模式全解析

  AI 技能
  AI 插件
  后端
  前端
  DevOps

• 如何免费永久保存微信聊天记录：WeChatMsg完整解决方案

  如何免费永久保存微信聊天记录：WeChatMsg完整解决方案 你是否曾因手机丢失、系统更新而丢失珍贵的微信聊天记录？那些与亲友的温馨对话、重要的工作沟通、难忘的

• franz-go架构设计原理：纯Go实现的高性能Kafka客户端

  franz go架构设计原理：纯Go实现的高性能Kafka客户端 franz go是一个功能完整的纯Go语言Kafka客户端库，支持从0.8.0到4.1+的Ka

  消息队列
  后端

上一篇：
LiveSplit计时器全面解析：从新手到高手的进阶指南

下一篇：
Dify工作流HTTP请求终极指南：从入门到精通

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考