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

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

Introducing Apache Spark® 4.3 - Now available in Databricks Runtime 19

AI 导读

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

正文 · 原文

Introduction

Apache Spark 4.3 brings stateful processing to Real-Time Mode, achieving sub-150 ms P99 end-to-end latency across three representative stateful benchmarks. Streaming aggregations, deduplication, and custom stateful processing can now run in Structured Streaming's low-latency execution mode, extending it to applications that need to remember context across events.

The impact goes beyond faster pipelines. For supported workloads, you no longer need a separate streaming engine just to achieve millisecond-scale latency. From fraud detection to continuously updated user activity and device monitoring, developers can build stateful applications with familiar Structured Streaming APIs in the same open engine they use for batch processing, analytics, and AI.

Stateful real-time streaming is the headline of Spark 4.3, but the improvements extend across the engine. The release delivers five key benefits:

• Build stateful real-time applications: Use streaming aggregation, deduplication, and transformWithState in Real-Time Mode to maintain context as events arrive.
• Preserve history with SCD Type 2 Auto CDC: Track how records change over time using Python, SQL, and Spark Connect, with less custom change-processing logic
• Iterate faster on small data: Project Feather targets unnecessary execution overhead, while Arrow-native caching supports more efficient data reuse.
• Connect more applications to Spark: Improvements to Spark Connect observability, JDBC integration, and Python make remote execution easier to develop and operate.
• Express richer analytics directly in SQL: VARIANT transformations, JSON_TABLE, temporal joins, and a nanosecond timestamp preview extend native processing of nested and time-oriented data.

Together, these changes help developers use one open engine to explore data, maintain changing context, and power analytical and AI applications.

Spark 4.3 also marks the start of a quarterly feature-release cadence, moving from one release every six months to one every three months. This shorter cycle puts completed improvements into users' hands sooner and gives the community more frequent opportunities to provide feedback and shape what comes next.

Real-Time Streaming: Stateful Applications with Millisecond-Scale Latency

From processing events to maintaining context

A fraud signal depends not only on the latest transaction, but also on the activity that preceded it. Personalization needs to reflect a user's recent behavior. Device monitoring needs to update counters and aggregates as new readings arrive. These applications need both low latency and state.

Spark 4.3 extends Real-Time Mode to these stateful processing patterns, making it a complete streaming engine. It adds support for streaming aggregation, dropDuplicates, and Scala/Java transformWithState (SPARK-58635, SPARK-57228). Developers can maintain keyed state and update results using Structured Streaming's existing programming model, rather than adopting another engine solely for lower latency.

Sub-150 ms P99 latency across three stateful benchmarks

In the benchmarks reported in SPARK-58884, all three representative stateful workloads achieved P99 end-to-end latency below 150 milliseconds:

These results demonstrate low-latency execution across deduplication, windowed aggregation, and custom stateful processing. They are measurements of the three tested workloads, not a latency guarantee for every pipeline.

Familiar APIs, low-latency execution

Real-Time Mode builds on Structured Streaming rather than introducing a separate programming model. Stateful operators continue to use Spark's existing state store and checkpointing mechanisms for durability and recovery.

Example: Maintain a running count for each device. Given a Kafka-backed streaming DataFrame events with a device_id column:

If events arrive for devices A, A, and B, Spark maintains state and updates the counts to A=2 and B=1. When another event for A arrives, its state is updated to A=3.

An execution engine built for continuous data flow

Supporting stateful Real-Time Mode required a fundamental change to how Spark schedules and connects stages. With a traditional blocking shuffle, a producing stage must finish before its downstream stage can begin. That model cannot support continuous data flow between long-running streaming tasks.

Spark 4.3 introduces native concurrent stage scheduling and pipelined shuffles, allowing producer and consumer stages to run at the same time (SPARK-54699, SPARK-58076, SPARK-56664). Streaming Shuffle implements this capability for Real-Time Mode, transferring records incrementally to the downstream task responsible for each key instead of waiting for the entire producing stage to finish.

Conceptually, execution changes from:

Producer stage → materialize shuffle → consumer stage

to:

Producer stage ⇄ streaming shuffle ⇄ consumer stage

Spark 4.3 adds PipelinedShuffleDependency, a new kind of shuffle dependency that allows consumers to start reading while producers are still running (SPARK-58185). The scheduler treats stages connected by these dependencies as a pipelined group, a coordinated scheduling unit. Its stages are scheduled together, and a failure in one stage causes the entire group to fail (SPARK-58263, SPARK-58398). In short, they run together and fail together.

Because all stages in a pipelined group run concurrently, Spark admits the group only when enough task slots are available for all of its tasks. Otherwise, it fails fast rather than hanging.

Spark routes each shuffle to a shuffle manager based on its dependency type:

• A regular ShuffleDependency uses a blocking shuffle manager, which materializes all shuffle output before downstream stages read it. SortShuffleManager is the built-in implementation.
• A PipelinedShuffleDependency uses a pipelined shuffle manager, which makes output available as it is produced. StreamingShuffleManager is the built-in implementation (SPARK-57141).

The dependency type defines the execution contract; the shuffle manager provides the data-transfer mechanism. The scheduler coordinates pipelined stages without needing to know which workload created them.

This foundation extends beyond Real-Time Mode. Concurrent stage scheduling, pipelined data exchange, and coordinated execution are generic Spark engine capabilities, not one-off streaming infrastructure. In Spark 4.3, they enable low-latency stateful processing. Longer term, they provide a foundation for other workloads that need to move beyond the traditional stage-at-a-time execution model.

For background on Real-Time Mode's design and application scenarios, see the Real-Time Mode introduction. That post describes the Databricks implementation; the Spark 4.3 feature scope and benchmark results discussed here are those listed above.

Project Feather: Faster Iteration on Small Data

LLM-assisted development often follows a short feedback loop: generate a transformation, run it on a sample, inspect the results, and refine the code. Each iteration may process only a small file or a few rows, yet task scheduling and shuffling can cost more than the computation itself.

Project Feather targets this fixed execution overhead, making Spark more responsive for exploring and validating SQL and DataFrame code on small datasets. Developers can use the same APIs as they move from a sample to larger workloads, whether they write the code themselves or work with an AI assistant.

Spark 4.3 introduces opt-in, shuffle-free single-task execution for eligible small, single-partition query fragments (SPARK-57851). When a qualifying scan reports that its output is already in one partition, supported downstream operations can avoid an unnecessary shuffle.

Example: Run and inspect a small aggregation. Use a small Parquet dataset stored as a single file, with a category column:

Six cases:

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

The result is one count per category. For an eligible query, the physical plan can omit the `Exchange` that would otherwise introduce a shuffle. This supports the same run-and-inspect workflow whether the transformation was written by a developer or suggested by a coding assistant.

With the default eligibility limits, qualifying file scans read one nonempty small file that fits within a single partition, while qualifying local in-memory relations contain up to 1,000 rows. Supported query shapes include selected scans, aggregations, distinct operations, windows, sorts, and limits.

Feather also adds optional Arrow-native Dataset caching (SPARK-57268), storing cached batches in Arrow format with compression and column statistics. Arrow-backed inputs can avoid unnecessary conversion, supporting workflows that repeatedly explore the same data. Both the single-task optimization and Arrow cache are opt-in. To turn it on, set the static config spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer when starting a fresh Spark application. Broader in-process local repartitioning is follow-on work targeted for the next release (SPARK-57399).

These improvements are just the beginning. The Spark community will continue investing in small-data performance in future releases, reducing execution overhead and shortening feedback loops for interactive exploration and LLM-assisted development.

SCD Type 2 in Open-Source Auto CDC: Preserve History as Data Changes

A current-state table can tell you where a customer lives today, but not which region they belonged to when an order was placed. Reporting, auditing, and historical analysis often need both. Spark 4.3 adds end-to-end Slowly Changing Dimension (SCD) Type 2 support to Auto CDC (SPARK-56249), so pipelines can track the history of every record rather than overwriting the latest value with every change. The capability is available through Python, SQL, and Spark Connect (SPARK-58321, SPARK-58247, SPARK-58261) and allows users to automatically apply changes instead of manual merge operations.

Example: Track customer status and region over time. Given a customer_changes pipeline source with a customer ID, event time, status, and region:

Alternatively, this can be expressed in SQL:

In an SCD-2 output table, every unique key is represented by a sequence of records, each marking the start and end of its validity using two extra metadata columns, __START_AT and __END_AT. For example, when a change record indicates that customer with id 17 moved from region West to East, the pipeline retains the record with the previous region West and marks its end as the event time of the change record; it inserts a new record with region East and the record’s event time as its start. An analysis can then use the region that applied at the relevant point in the event time sequence.

Under the hood, Auto CDC connects the declarative flow definition to target-schema derivation and dedicated SCD Type 2 stream processing (SPARK-58319, SPARK-57395, SPARK-58321). Keys identify the entity, sequencing determines the order of changes, and the history-tracking columns specify which attribute changes create historical versions. The engine also validates reserved columns, sequencing types, and history-tracking configuration (SPARK-57251, SPARK-58313, SPARK-58391), and reconciles narrower microbatches with an evolved target schema (SPARK-58418).

Auto CDC applies each micro-batch to the target using MERGE, hence the target table must be in a catalog whose connector implements Data Source V2 row-level operations. Spark's built-in file formats, including Parquet, don't qualify. Therefore, a common choice is Iceberg.

Spark Connect: Easier to Develop, Observe, and Operate

Applications and AI agents need to invoke Spark from their own processes, manage sessions, and connect a failed request to the work running on the server. Spark 4.3 strengthens Spark Connect for these workflows with session and operation visibility, request attribution, and broader client support.

Spark Connect sends DataFrame operations as logical plans over gRPC, executes them through the server's Spark engine, and streams results back in Arrow batches. Spark 4.3 adds REST endpoints to list and inspect Connect sessions and operations, plus operation IDs for tracing execution across clients and the server (SPARK-57941, SPARK-58547). For local PySpark development on POSIX systems, the experimental `SPARK_LOCAL_CONNECT_REUSE=1` option reuses a persistent Connect server across runs, avoiding repeated startup overhead (SPARK-57787). PySpark's SQLContext compatibility wrapper also works with Spark Connect (SPARK-57021).

The Spark Connect JDBC driver adds ARRAY, MAP, and STRUCT result support and richer result-set handling (SPARK-57380, SPARK-57274), SQL-type metadata and more compatible schema introspection (SPARK-57627), and improved exception handling (SPARK-57933, SPARK-58619). API coverage expands with DataFrame.zip, SCD Type 2 Auto CDC, and nanosecond timestamp types and literals (SPARK-57247, SPARK-58247, SPARK-57160, SPARK-57161).

Separately, community work includes Spark Connect Gateway (SPARK-58455) for session-aware routing and multi-tenant operation, Rust and Rust-backed Python clients (SPARK-58960) for another client implementation path, and a capability-discovery proposal (SPARK-56398) to identify supported server operations before submitting a query. These efforts have separate development timelines and are not bundled 4.3 features.

PySpark: Richer APIs and Better Python Execution

Python workflows often build derived columns in separate steps, combine the results, and move data between pandas, Arrow, and Spark. Spark 4.3 makes more of this work available through DataFrame APIs while improving the execution paths underneath them. DataFrame.zip adds a way to combine compatible column projections without dropping down to RDD APIs, including when using Spark Connect (SPARK-55886, SPARK-57247).

Example: Combine identifiers with a derived value. Create two projections from the same input and bring their columns together:

The result pairs id=1 with doubled=20 and id=2 with doubled=40. `zip` combines projection-only transformations of the same underlying plan; filtering, sorting, joins, or aggregations below those projections must be identical on both sides to preserve row alignment.

The broader execution work targets data transfer, batching, and native computation. Pickle-based UDF input batches can be bounded by bytes rather than only row count, giving better control when rows vary in size (SPARK-57593). Batched pandas-on-Spark projections and additional native NumPy implementations reduce avoidable work (SPARK-58230, SPARK-58234), while pandas-to-Spark conversion adds support for PyArrow ChunkedArray columns (SPARK-46776).

The release also adds Python and Scala bindings for existing vector functions (SPARK-57284), grouped covariance and descriptive statistics (SPARK-40336, SPARK-40337), rolling and expanding median and standard-error operations (SPARK-40341, SPARK-40340), and show_versions diagnostics (SPARK-40373). Traceback locals are opt-in because they can contain sensitive data (SPARK-57271).

Spark SQL: Prepare Better Context for Analytics and AI

Event payloads arrive as nested documents, and closely spaced events can require more precision than microseconds. Preparing these inputs should not require a separate custom-processing layer for every transformation. Spark 4.3 expands native SQL for semi-structured data and temporal analytics, including document manipulation, temporal matching, and an opt-in nanosecond timestamp preview (SPARK-56822).

Example: Distinguish two events within the same microsecond. Enable the preview (feature not stable yet, breaking changes may happen) and compare timestamps one nanosecond apart (SPARK-56969, SPARK-56965, SPARK-57211):

The comparison evaluates to true: the final digit is preserved rather than discarded. The preview supports TIMESTAMP_NTZ(p) and TIMESTAMP_LTZ(p) at precisions from 7 through 9.

This required more than accepting a new SQL type name. Nanosecond support extends through parsing, internal value representations, casts, expressions, Spark Connect, caching, and data-source integration (SPARK-56822). New timestamp_nanos and unix_nanos functions handle numeric conversions (SPARK-57526, SPARK-57527). Storage and connector support remains specific to each data path, so the preview must be evaluated end to end.

Turn changing JSON payloads into usable data

Application events, document exports, and model responses often arrive as nested JSON. Spark 4.3’s JSON_TABLE (SPARK-58366) combines array expansion and typed field extraction, while VARIANT transformations let pipelines update or remove fields without custom document-processing UDFs (SPARK-57804, SPARK-57417). Together, they simplify preparing content for analytics, enrichment, and retrieval pipelines.

Example: Extract relevant document sections.

This produces a typed row that a pipeline can enrich with source metadata before other downstream operations. JSON_TABLE resolves its column paths through a shared traversal of each row item, avoiding a separate parse for every column. Separately, Spark extends VARIANT field-extraction pushdown to supported queries that aggregate, sort, or join. (SPARK-58089).

Follow-on SQL/JSON work (SPARK-58686) extends this direction with JSON_VALUE, JSON_EXISTS, and JSON_QUERY for scalar extraction, presence checks, and fragment retrieval (SPARK-58685, SPARK-58687, SPARK-58723). Explicit missing-value and error handling will help applications process changing schemas more predictably. These are follow-on developments, not additional 4.3 features.

Match events to the information available at the time

Historical analysis and model evaluation need the context available when an event occurred, not information added later. Spark 4.3’s SQL ASOF JOIN expresses (feature in preview, breaking changes may happen) this temporal match directly, replacing hand-written range joins followed by ranking (SPARK-58092).

Example: Reconstruct the features available for a prediction. Here, available_at records when a feature version became available (SPARK-58122, SPARK-58123):

A prediction at 10:03 matches the feature version available at 10:00, not 10:05. A dedicated sort-merge execution path makes this matching an engine operation rather than application-specific query logic. The result is simpler point-in-time enrichment for training data, evaluations, and incident analysis, provided the availability timestamps accurately represent what was known then.

Notable Improvements and Acknowledgements

Spark 4.3 also introduces new APIs and performance improvements for everyday data processing. SQL gains ANSI UNNEST for expanding arrays into rows, along with the `counter_diff` window function for converting cumulative counters into deltas and `jaro_winkler_similarity` for string similarity comparisons. Supported file sources add opt-in support for reading tar, zip, and 7z archives, eliminating a separate extraction step. The engine improves aggregation execution and enables whole-stage code generation for eligible UNION operations by default. An opt-in optimization spreads null join keys across shuffle partitions in eligible outer and anti joins. Nested-data processing gains more pruning of unused fields inside arrays of structs and code generation for more array higher-order functions. Vectorized Parquet reads improve type conversion and dictionary decoding.

Spark 4.3 reflects the strength of the Apache Spark community, with more than 200 contributors helping address over 1,100 Jira tickets. We thank everyone who contributed code, reviews, testing, documentation, and feedback to make this release possible.

Get Started with Spark 4.3

Visit the Apache Spark downloads page for release downloads and documentation, and review the migration guides before upgrading. For Databricks availability, consult the Databricks Runtime release notes.

来源：Databricks：Blog · databricks.com