Apache Spark 4.3 发布：Real-Time Mode 支持有状态处理，已上线 Databricks Runtime 19

Databricks：Blog·2026-11-05 08:00· 52 分钟前AI 评分63

推出 Apache Spark® 4.3 - 现已在 Databricks Runtime 19 中提供

AI 导读

Databricks 宣布 Apache Spark 4.3 发布，并已在 Databricks Runtime 19 提供。

正文 · 原文

引言

Apache Spark 4.3 将状态化处理引入实时模式，在三个具有代表性的状态化基准测试中实现低于 150 毫秒的 P99 端到端延迟。流式聚合、去重和自定义状态化处理现在可以在 Structured Streaming 的低延迟执行模式中运行，将其扩展到需要跨事件记住上下文的应用程序。

其影响不止于更快的管道。对于受支持的工作负载，你不再仅仅为了实现毫秒级延迟而需要单独的流处理引擎。从欺诈检测到持续更新的用户活动和设备监控，开发者可以使用熟悉的 Structured Streaming API，在同一个用于批处理、分析和 AI 的开放引擎中构建状态化应用程序。

状态化实时流处理是 Spark 4.3 的头条亮点，但改进遍及整个引擎。此版本带来五项关键优势：

• 构建状态化实时应用程序：在实时模式中使用流式聚合、去重和 transformWithState，在事件到达时维护上下文。
• 使用 SCD Type 2 Auto CDC 保留历史：使用 Python、SQL 和 Spark Connect 跟踪记录随时间的变化，减少自定义变更处理逻辑
• 在小数据上更快迭代：Project Feather 针对不必要的执行开销，而 Arrow 原生缓存支持更高效的数据复用。
• 将更多应用程序连接到 Spark：Spark Connect 可观测性、JDBC 集成和 Python 的改进使远程执行更易于开发和运维。
• 直接在 SQL 中表达更丰富的分析：VARIANT 转换、JSON_TABLE、时间连接和纳秒时间戳预览扩展了对嵌套数据和时间导向数据的原生处理。

这些变化共同帮助开发者使用一个开放引擎来探索数据、维护不断变化的上下文，并驱动分析和 AI 应用程序。

Spark 4.3 还标志着季度功能发布节奏的开始，从每六个月发布一次改为每三个月发布一次。这一更短的周期将已完成的改进更快地交到用户手中，并为社区提供更频繁的机会来提供反馈并塑造未来方向。

实时流处理：具有毫秒级延迟的状态化应用程序

从处理事件到维护上下文

欺诈信号不仅取决于最新交易，还取决于之前发生的活动。个性化需要反映用户的近期行为。设备监控需要在新读数到达时更新计数器和聚合。这些应用程序既需要低延迟，也需要状态。

Spark 4.3 将实时模式扩展到这些状态化处理模式，使其成为完整的流处理引擎。它增加了对流式聚合、dropDuplicates 以及 Scala/Java transformWithState 的支持（SPARK-58635、 SPARK-57228）。开发者可以使用 Structured Streaming 现有的编程模型维护键控状态并更新结果，而不必仅仅为了更低延迟而采用另一个引擎。

在三个有状态基准测试中，P99 延迟均低于 150 毫秒

在 SPARK-58884 报告的基准测试中，三个具有代表性的有状态工作负载均实现了低于 150 毫秒的 P99 端到端延迟：

这些结果展示了在去重、窗口聚合和自定义有状态处理场景下的低延迟执行能力。它们是对三个受测工作负载的测量结果，并非对每条管道的延迟保证。

熟悉的 API，低延迟执行

实时模式构建在 Structured Streaming 之上，而非引入独立的编程模型。有状态算子继续使用 Spark 现有的状态存储和检查点机制来实现持久化和恢复。

示例：为每个设备维护一个累计计数。给定一个由 Kafka 支持的流式 DataFrame events，其中包含 device_id 列：

如果设备 A、A 和 B 的事件到达，Spark 会维护状态并将计数更新为 A=2 和 B=1。当 A 的另一个事件到达时，其状态更新为 A=3。

为持续数据流打造的执行引擎

支持有状态的实时模式需要对 Spark 调度和连接各阶段的方式进行根本性改变。在传统的阻塞式 shuffle 中，生产阶段必须完成后其下游阶段才能开始。这种模型无法支持长时间运行的流式任务之间的持续数据流。

Spark 4.3 引入了原生并发阶段调度和流水线化 shuffle，允许生产者和消费者阶段同时运行（SPARK-54699、 SPARK-58076、 SPARK-56664）。Streaming Shuffle 为实时模式实现了这一能力，将记录增量式地传输给负责每个键的下游任务，而不是等待整个生产阶段完成。

从概念上讲，执行方式从：

生产者阶段 → 物化 shuffle → 消费者阶段

变为：

生产者阶段 ⇄ 流式 shuffle ⇄ 消费者阶段

Spark 4.3 新增了 PipelinedShuffleDependency，这是一种新型 shuffle 依赖，允许消费者在生产者的运行过程中就开始读取（SPARK-58185）。调度器将由这些依赖连接的阶段视为一个流水线组，即一个协调调度单元。其各阶段一起调度，一个阶段失败会导致整个组失败（SPARK-58263、 SPARK-58398）。简而言之，它们同运行、共失败。

由于流水线组中的所有阶段并发运行，Spark 只有在有足够的任务槽位容纳其所有任务时才会接纳该组。否则，它会快速失败而不是挂起。

Spark 根据依赖类型将每个 shuffle 路由到相应的 shuffle 管理器：

• 常规的 ShuffleDependency 使用阻塞式 shuffle 管理器，它会在下游阶段读取之前物化所有 shuffle 输出。SortShuffleManager 是内置实现。
• PipelinedShuffleDependency 使用流水线化 shuffle 管理器，它在输出产生时即可供读取。StreamingShuffleManager 是内置实现（SPARK-57141）。

依赖类型定义了执行契约；shuffle manager 提供数据传输机制。调度器协调流水线化的各个阶段，而无需知道是哪个工作负载创建了它们。

这一基础超越了 Real-Time Mode。并发阶段调度、流水线化数据交换和协同执行是 Spark 引擎的通用能力，而非一次性的流式基础设施。在 Spark 4.3 中，它们实现了低延迟的有状态处理。从更长远来看，它们为其他需要突破传统“一次一个阶段”执行模型的工作负载提供了基础。

关于 Real-Time Mode 的设计和应用场景背景，请参阅 Real-Time Mode 简介。那篇文章描述的是 Databricks 的实现；本文讨论的 Spark 4.3 功能范围和基准测试结果即为上文所列内容。

Project Feather：加速小数据上的迭代

LLM 辅助开发通常遵循一个短反馈循环：生成一个转换、在样本上运行、检查结果并改进代码。每次迭代可能只处理一个小文件或几行数据，但任务调度和 shuffle 的开销可能超过计算本身。

Project Feather 针对这一固定执行开销，使 Spark 在小型数据集上探索和验证 SQL 与 DataFrame 代码时响应更迅速。开发者从样本迁移到更大工作负载时可以使用相同的 API，无论代码是自己编写还是借助 AI 助手完成。

Spark 4.3 为符合条件的小型单分区查询片段引入了可选的、无 shuffle 的单任务执行（SPARK-57851）。当符合条件的扫描报告其输出已位于一个分区时，受支持的下游操作可以避免不必要的 shuffle。

示例：运行并检查一个小型聚合。使用以单个文件存储的小型 Parquet 数据集，其中包含一个 category 列：

六种情况：

category, Exchange, SPARK_LOCAL_CONNECT_REUSE=1, zip, counter_diff and jaro_winkler_similarity.

结果是每个类别一个计数。对于符合条件的查询，物理计划可以省略原本会引入 shuffle 的 `Exchange`。这支持相同的运行并检查工作流，无论转换是由开发者编写还是由编码助手建议。

在默认的资格限制下，符合条件的文件扫描读取一个非空的小文件，且该文件能容纳在单个分区内；而符合条件的本地内存关系最多包含 1,000 行。支持的查询形态包括选定的扫描、聚合、去重操作、窗口、排序和限制。

Feather 还增加了可选的 Arrow 原生 Dataset 缓存（SPARK-57268），以 Arrow 格式存储缓存批次，并带有压缩和列统计信息。Arrow 支持的输入可以避免不必要的转换，从而支持反复探索同一数据的工作流。单任务优化和 Arrow 缓存均为可选。要启用它，请在启动新的 Spark 应用程序时设置静态配置 spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer。更广泛的进程内本地重分区是后续工作，目标是在下一个版本中实现（SPARK-57399）。

这些改进只是开始。Spark 社区将在未来版本中继续投资于小数据性能，降低执行开销并缩短交互式探索和 LLM 辅助开发的反馈循环。

开源 Auto CDC 中的 SCD Type 2：在数据变化时保留历史

当前状态表可以告诉你客户今天住在哪里，但无法告诉你下单时该客户属于哪个地区。报表、审计和历史分析往往两者都需要。Spark 4.3 为 Auto CDC 增加了端到端的缓慢变化维度（SCD）Type 2 支持（SPARK-56249），因此管道可以跟踪每条记录的历史，而不是在每次变更时覆盖最新值。该能力可通过 Python、SQL 和 Spark Connect 使用（SPARK-58321、SPARK-58247、SPARK-58261），并允许用户自动应用变更，而无需手动执行合并操作。

示例：随时间跟踪客户状态和地区。给定一个包含客户 ID、事件时间、状态和地区的 customer_changes 管道源：

或者，也可以用 SQL 表达：

在 SCD-2 输出表中，每个唯一键由一系列记录表示，每条记录使用两个额外的元数据列 __START_AT 和 __END_AT 标记其有效期的开始和结束。例如，当一条变更记录表明 id 为 17 的客户从地区 West 迁移到 East 时，管道会保留地区为 West 的先前记录，并将其结束时间标记为该变更记录的事件时间；同时插入一条新记录，其地区为 East，并以该记录的事件时间作为开始时间。之后，分析可以使用事件时间序列中相关时间点适用的地区。

在底层，Auto CDC 将声明式流定义与目标模式推导和专用的 SCD Type 2 流处理连接起来（SPARK-58319、SPARK-57395、SPARK-58321）。键标识实体，排序决定变更顺序，历史跟踪列指定哪些属性变更会创建历史版本。引擎还会校验保留列、排序类型和历史跟踪配置（SPARK-57251、SPARK-58313、SPARK-58391），并将较窄的微批次与演进后的目标模式进行协调（SPARK-58418）。

Auto CDC 使用 MERGE 将每个微批次应用到目标表，因此目标表必须位于其连接器实现了 Data Source V2 行级操作的目录中。Spark 内置的文件格式（包括 Parquet）不符合要求。因此，常见的选择是 Iceberg。

Spark Connect：更易于开发、观测和运维

应用程序和 AI 代理需要从自己的进程中调用 Spark、管理会话，并将失败的请求与服务器上运行的工作关联起来。Spark 4.3 通过会话和操作可见性、请求归因以及更广泛的客户端支持，为这些工作流增强了 Spark Connect。

Spark Connect 通过 gRPC 将 DataFrame 操作作为逻辑计划发送，通过服务器的 Spark 引擎执行，并以 Arrow 批次流式返回结果。Spark 4.3 增加了 REST 端点来列出和检查 Connect 会话与操作，并增加了操作 ID，用于跨客户端和服务器跟踪执行（SPARK-57941、SPARK-58547）。对于 POSIX 系统上的本地 PySpark 开发，实验性的 `SPARK_LOCAL_CONNECT_REUSE=1` 选项可在多次运行之间复用持久化的 Connect 服务器，避免重复的启动开销（SPARK-57787）。PySpark 的 SQLContext 兼容包装器也可与 Spark Connect 配合使用（SPARK-57021）。

Spark Connect JDBC 驱动新增了对 ARRAY、MAP 和 STRUCT 结果的支持，并增强了结果集处理能力（SPARK-57380、SPARK-57274），支持 SQL 类型元数据和更兼容的 schema 内省（SPARK-57627），并改进了异常处理（SPARK-57933、SPARK-58619）。API 覆盖范围也进一步扩展，新增了 DataFrame.zip、SCD Type 2 Auto CDC，以及纳秒时间戳类型和字面量（SPARK-57247、SPARK-58247、SPARK-57160、SPARK-57161）。

此外，社区工作还包括 Spark Connect Gateway（SPARK-58455），用于会话感知路由和多租户运行；Rust 客户端及基于 Rust 的 Python 客户端（SPARK-58960），提供另一条客户端实现路径；以及一项能力发现提案（SPARK-56398），用于在提交查询前识别服务器支持的操作。这些工作有各自的开发时间线，并未捆绑为 4.3 的功能。

PySpark：更丰富的 API 与更好的 Python 执行

Python 工作流通常分步构建派生列、合并结果，并在 pandas、Arrow 和 Spark 之间移动数据。Spark 4.3 通过 DataFrame API 让更多此类工作可以直接完成，同时改进了底层执行路径。DataFrame.zip 新增了一种合并兼容列投影的方式，无需下探到 RDD API，在使用 Spark Connect 时同样适用（SPARK-55886、SPARK-57247）。

示例：将标识符与派生值合并。从同一输入创建两个投影，并将它们的列合并在一起：

结果将 id=1 与 doubled=20 配对，将 id=2 与 doubled=40 配对。`zip` 合并的是同一底层计划上的纯投影变换；这些投影之下的过滤、排序、连接或聚合必须在两侧完全一致，才能保持行对齐。

更广泛的执行工作聚焦于数据传输、批处理和原生计算。基于 Pickle 的 UDF 输入批次现在可以按字节数而非仅按行数限制，从而在行大小不一的情况下提供更好的控制（SPARK-57593）。批处理的 pandas-on-Spark 投影和更多原生 NumPy 实现减少了可避免的开销（SPARK-58230、SPARK-58234），同时 pandas 到 Spark 的转换新增了对 PyArrow ChunkedArray 列的支持（SPARK-46776）。

该版本还为现有向量函数新增了 Python 和 Scala 绑定（SPARK-57284），新增了分组协方差和描述性统计（SPARK-40336、SPARK-40337）、滚动和扩展中位数与标准误操作（SPARK-40341、SPARK-40340），以及 show_versions 诊断功能（SPARK-40373）。Traceback 局部变量为选择性启用，因为它们可能包含敏感数据（SPARK-57271）。

Spark SQL：为分析与 AI 准备更好的上下文

事件负载以嵌套文档形式到达，而间隔很近的事件可能需要比微秒更高的精度。准备这些输入不应要求为每种变换单独构建自定义处理层。Spark 4.3 扩展了用于半结构化数据和时间分析的原生 SQL，包括文档操作、时间匹配，以及选择性启用的纳秒时间戳预览（SPARK-56822）。

示例：区分同一微秒内的两个事件。启用预览（该功能尚不稳定，可能发生破坏性变更），并比较相隔一纳秒的时间戳（SPARK-56969、SPARK-56965、SPARK-57211）：

比较结果为 true：最后一位数字被保留而非丢弃。该预览支持 TIMESTAMP_NTZ(p) 和 TIMESTAMP_LTZ(p)，精度范围为 7 到 9。

这不仅仅是接受一个新的 SQL 类型名称。纳秒支持贯穿解析、内部值表示、类型转换、表达式、Spark Connect、缓存和数据源集成（SPARK-56822）。新增的 timestamp_nanos 和 unix_nanos 函数负责数值转换（SPARK-57526、SPARK-57527）。存储和连接器支持仍因各数据路径而异，因此必须对预览进行端到端评估。

将不断变化的 JSON 载荷转化为可用数据

应用事件、文档导出和模型响应通常以嵌套 JSON 形式到达。Spark 4.3 的 JSON_TABLE（SPARK-58366）结合了数组展开和类型化字段提取，而 VARIANT 转换让管道无需自定义文档处理 UDF 即可更新或删除字段（SPARK-57804、SPARK-57417）。二者共同简化了为分析、增强和检索管道准备内容的过程。

示例：提取相关文档章节。

这会生成一个类型化行，管道可在其他下游操作之前用源元数据对其进行增强。JSON_TABLE 通过对每个行项进行共享遍历来解析其列路径，避免为每一列单独解析。另外，Spark 将 VARIANT 字段提取下推扩展到支持聚合、排序或连接的查询（SPARK-58089）。

后续 SQL/JSON 工作（SPARK-58686）以 JSON_VALUE、JSON_EXISTS 和 JSON_QUERY 扩展了这一方向，分别用于标量提取、存在性检查和片段检索（SPARK-58685、SPARK-58687、SPARK-58723）。显式的缺失值处理和错误处理将帮助应用更可预测地处理不断变化的模式。这些是后续开发，并非 4.3 的额外功能。

将事件与当时可用的信息进行匹配

历史分析和模型评估需要事件发生时已有的上下文，而非后来添加的信息。Spark 4.3 的 SQL ASOF JOIN（预览功能，可能发生破坏性变更）直接表达这种时间匹配，取代了手写的范围连接加排序（SPARK-58092）。

示例：重建预测时可用的特征。此处，available_at 记录了某个特征版本何时可用（SPARK-58122、SPARK-58123）：

10:03 的预测匹配的是 10:00 可用的特征版本，而非 10:05 的版本。专用的排序合并执行路径使这种匹配成为引擎操作，而非应用特定的查询逻辑。只要可用时间戳准确反映了当时已知的信息，就能为训练数据、评估和事件分析提供更简单的时点增强。

值得注意的改进与致谢

Spark 4.3 还为日常数据处理引入了新的 API 和性能改进。SQL 新增了用于将数组展开为行的 ANSI UNNEST，以及用于将累积计数器转换为增量的 `counter_diff` 窗口函数和用于字符串相似度比较的 `jaro_winkler_similarity`。支持的文件源新增了对读取 tar、zip 和 7z 归档文件的可选支持，从而省去了单独的提取步骤。该引擎改进了聚合执行，并默认对符合条件的 UNION 操作启用全阶段代码生成。一项可选优化会在符合条件的外连接和反连接中，将空连接键分散到各个 shuffle 分区。嵌套数据处理方面，增强了对结构体数组中未使用字段的裁剪，并为更多数组高阶函数提供代码生成。向量化 Parquet 读取改进了类型转换和字典解码。

Spark 4.3 体现了 Apache Spark 社区的强大实力，超过 200 名贡献者帮助处理了 1,100 多个 Jira 工单。我们感谢每一位为促成此版本而贡献代码、评审、测试、文档和反馈的人。

开始使用 Spark 4.3

请访问 Apache Spark 下载页面 获取版本下载和文档，并在升级前查阅迁移指南。有关 Databricks 上的可用性，请参阅 Databricks Runtime 发行说明。

来源：Databricks：Blog · databricks.com