The Ultimate Guide to Apache Spark 4.3

A release about time (mostly)

Spark 4.3.0 came out on October 30, including the more in depth Databricks release blog, which means my feeds are about to fill with listicles that paste the release notes into a numbered list and call it a day. I wanted to do something a little more in depth by reviewing each of the major features in this release as closely as possible, so that people can get a better idea of the expected behaviours and what it's like to actually run the changes. Funnily, I expect a blog like this to have a shorter shelf life than usual as these releases are currently en track for a quarterly cadence, meaning Spark is about to get much more fully featured in the next year as it moves towards 5.0.

Reading the release notes end to end, I kept noticing that the big items all circle the same thing from different angles. Real-time Mode is about how quickly one record gets through a stateful query; SCD Type 2 is about how long a row's history survives; ASOF JOIN is about which version of a fact was true at a given moment; and nanosecond timestamps are about how finely Spark can tell two moments apart in the first place. The remaining features (Project Feather, the Spark Connect observability work, DataFrame.zip, Python UDF transpilation) are about a different clock, which is yours: how long you wait between writing a query and seeing it run.

Most of Spark 4.3 is about time, both the data's and yours: the scheduler learned to pipeline shuffles, so stateful streaming queries now run at millisecond latency; Auto CDC learned to keep history; SQL learned point-in-time joins and nanosecond precision; and the rest of the release shortens the loop between writing Spark code and finding out whether it works.

Every example here is a runnable script in spark-demos/demos/16_spark_43_guide, and every clip below is that script running against Spark 4.3.0 (pip install pyspark==4.3.0) on my machine. The Real-time Mode clips come from the Real-time Mode demo, which is a teaser for one of my upcoming Spark Streaming blog posts ;)

Below is a rundown of the new features and how to enable or take advantage of them.

Stateful Real-time Mode

Real-time Mode landed in 4.1 as a new trigger for Structured Streaming: instead of slicing a stream into micro-batches and waiting for each one to be planned, scheduled and committed, the query runs long-lived tasks that process records as they arrive and checkpoint on a much longer interval. The catch in 4.1 and 4.2 was that it only ran stateless queries (filters, projections, stream-static joins), because anything stateful needs a shuffle, and a classic shuffle is a blocking barrier where every map task has to finish writing before a single reduce task can read.

A new PipelinedShuffleDependency (SPARK-58185) routes data through a new StreamingShuffleManager (SPARK-57141), which pushes records to downstream tasks over Netty while the upstream tasks are still running. The DAGScheduler now schedules a pipelined group of stages together, fails them together, and refuses to start a group without enough task slots (SPARK-58263, SPARK-58398). This admission check prevents deadlock on a busy cluster: a consumer stage started without its producer would sit idle while holding cores. With that machinery in place, Real-time Mode supports streaming aggregation (SPARK-58635), dropDuplicates (SPARK-58508) and Scala/Java transformWithState (SPARK-57228), and is no longer marked @Experimental.

The query itself doesn't change shape. In this deduplication and windowed-count step from the bookshop demo, only the trigger and output mode are specific to Real-time Mode:

trending = (clicks .withWatermark("ts", "10 seconds") .dropDuplicates(["event_id"]) # resent clicks counted once  .groupBy(F.window("ts", "10 seconds"), "aisle") .count()) (trending.writeStream.format("kafka") .outputMode("update") .trigger(realTime="30 seconds") # the whole change  .start()) 

Measured end to end on my machine (from the click's own timestamp to the broker's append time on the output topic, 200 clicks a second for 45 seconds), the same query looks like this in each mode:

The 50 ms median comes from a timer. The streaming shuffle flushes when its buffer fills or spark.shuffle.streaming.networkBufferMaxWaitTimeMs expires; the default is 50 ms, and a lightly loaded query does not fill the buffer first. Reducing the wait to 1 ms brought the median down to 2 ms at the cost of more small network writes, a tradeoff to tune per workload rather than a useful global default. (The shorter run in the clip measured a 67 ms median and 96 ms p99 at the default, then 2 ms and 3 ms at 1 ms; the table contains the longer benchmark, and this amount of variation is normal for a single-machine test.) Databricks reports p99 latencies of 136 ms for dropDuplicates, 139 ms for a window aggregation and 141 ms for transformWithState on its benchmark workloads (SPARK-58884). Those results are not guarantees, and neither are mine, which come from one desktop with Kafka running in Docker.

For a stateful example, the recommender keeps a taste vector per shopper, built from book embeddings, in transformWithState and rescores the catalog on every click. The recommendations therefore change a few milliseconds after each click:

Real-time Mode still has several limits. transformWithState supports only Scala and Java, requires the RocksDB state store, and fires a processing-time timer on an idle key only when another row reaches its partition or the Real-time batch ends. ASOF JOIN is not supported. Pipelined stages also require cores concurrently, so the cluster must be sized for the whole pipeline rather than its widest stage.

SCD Type 2 in Auto CDC

Auto CDC arrived in 4.2 as the Declarative Pipelines way to turn a change feed into current state: give it keys, a sequencing column and a delete rule, and it keeps a target table in sync with MERGE, handling out-of-order and replayed events for you. In 4.2 that meant SCD Type 1, where the target only ever holds the latest version of each row. 4.3 adds SCD Type 2 (SPARK-58321), in Python (stored_as_scd_type=2, track_history_column_list), in SQL (STORED AS SCD TYPE 2, TRACK HISTORY ON) and through Spark Connect, so the target keeps every version, each bounded by __START_AT and __END_AT.

I reused the ghost kitchen from that post, this time tracking its menu. Each dish change arrives on a Kafka topic as a JSON event, as it would from a CDC tool. The pipeline has a Python source that reads the topic,

@dp.table def menu_changes(): return (spark.readStream.format("kafka") .option("kafka.bootstrap.servers", "localhost:29292") .option("subscribe", "menu-changes") .option("startingOffsets", "earliest") .load() .select(from_json(col("value").cast("string"), CHANGE).alias("e")) .select("e.*")) 

and this one SQL statement:

-- Keep every version of every dish, not just the latest one. CREATE STREAMING TABLE menu FLOW AUTO CDC FROM STREAM(menu_changes) KEYS (dish_id) SEQUENCE BY seq APPLY AS DELETE WHEN op = 'DELETE' COLUMNS * EXCEPT (op) STORED AS SCD TYPE 2 TRACK HISTORY ON (price, station); 

TRACK HISTORY ON controls which changes open a new version. Changes to unlisted columns update the current version in place, which prevents corrections such as typo fixes from creating history entries. Publishing change events in batches and running the pipeline after each batch shows both cases:

After the third run, the table contains the full history of each dish:

The typo fix at sequence 3 did not open a third row for the tacos because dish is not tracked; it updated the current version instead. Removing khao soi closed its window at sequence 2 rather than deleting it, preserving the previous menu state. __START_AT and __END_AT contain the sequencing value, which is an integer here but could instead be a commit timestamp.

As with Type 1, the target must support DSv2 row-level operations because Auto CDC applies each micro-batch with MERGE. Spark's built-in file formats, including Parquet, fail at startup with AUTOCDC_TARGET_DOES_NOT_SUPPORT_MERGE; Apache Iceberg is the usual choice. Iceberg has not published a Spark 4.3 runtime yet (Maven Central currently stops at iceberg-spark-runtime-4.1), so I used an Iceberg runtime built from a branch with Spark 4.3 support. The Iceberg runtime, Kafka connector jars and spark.sql.extensions are static configurations and must be passed to spark-pipelines run, which accepts spark-submit arguments. Putting them in spark-pipeline.yml makes the CLI apply them to a live session and fail with CANNOT_MODIFY_CONFIG.

Project Feather: faster on small data

Spark has always been a slightly absurd tool for a 500-row file because its planning assumes the data is large: a groupBy gets a partial aggregate, a hash shuffle and a final aggregate, while an orderBy adds a range shuffle. Each exchange creates a stage boundary and another scheduling round trip. That overhead disappears into the work on a cluster; on a laptop, in a test suite, or in an agent running hundreds of small queries, it becomes the work. Project Feather (SPARK-56978) addresses that overhead, and 4.3 includes its first two pieces, both opt-in.

Shuffle-free single-task execution (SPARK-57851, spark.sql.optimizer.singleTaskExecution.enabled) notices when a query fragment reads something that fits in one partition (by default a single non-empty small file, or a local relation of up to 1,000 rows) and plans the scans, aggregations, distincts, windows, sorts and limits above it as one task with no exchanges at all. The same aggregation, with the flag off and then on:

def run(single_task, runs=20): spark.conf.set("spark.sql.optimizer.singleTaskExecution.enabled", single_task) t = time.perf_counter() for _ in range(runs): # a fresh DataFrame each time, like a notebook cell or a test  df = (spark.read.parquet(small) .groupBy("store").count().orderBy("store")) df.collect() return df, (time.perf_counter() - t) / runs * 1000 

Both exchanges (and the shuffle query stages AQE wraps around them) are gone, and the query takes about 60% of the time, warm, on a session that's already paid for JVM startup. The loop builds a new DataFrame every time on purpose, because that's what a notebook cell, a test or an agent does. My first version collected the same DataFrame twenty times, and from the second collect on AQE reuses the shuffle output it already has, skipping the very stages Feather removes, so the two settings came out within a few milliseconds of each other. I wrote up the full benchmark (more queries, cold starts, and a test suite that went from 12.6 s to 5.2 s) in Benchmark Project Feather on a Fresh DataFrame.

The second piece is Arrow-native Dataset caching (SPARK-57268): set spark.sql.cache.serializer=org.apache.spark.sql.execution.columnar.ArrowCachedBatchSerializer when the application starts (it's static) and df.cache() stores Arrow batches instead of Spark's own columnar format, which pairs nicely with everything Arrow already does between the JVM and Python. Compression is available through spark.sql.execution.arrow.compression.codec, but it defaults to none, so out of the box the Arrow cache is bigger than the default one. Broader in-process local repartitioning was deferred to the next release (SPARK-57399).

Spark Connect: operation IDs and REST endpoints

In my idiot's guide to Connect I complained that debugging a Connect app means correlating a client-side error with a server you can't see, and 4.3 fixes the two halves of that. First, the Python and Scala clients now generate an operation ID before every ExecutePlan request (SPARK-58547) and hand it back on success (df.executionInfo.operation_id) and, more usefully, on failure (e.operation_id on the exception), because a request that fails before its first response never learned a server-generated ID. The server publishes the same ID as a Spark local property, so listeners and executor tasks can see it too.

remote = SparkSession.builder.remote("sc://localhost:15043") with remote.getOrCreate() as spark: # new in 4.3: sessions are context managers  ok = spark.sql("SELECT count(*) AS n FROM range(1000000)") ok.collect() print("succeeded, operation", ok.executionInfo.operation_id) try: spark.sql("SELECT revenue FROM range(10)").collect() except Exception as e: # the client knows the ID even on failure  print("failed, operation", e.operation_id) 

Second, the server's sessions and operations are now in the monitoring REST API (SPARK-57941), next to the SQL and streaming endpoints, where before they only lived in the Connect tab of the UI:

curl -s $API/connect/sessions | jq curl -s $API/connect/operations | jq '.[] | select(.operationId == env.OP)' 

Take the ID the client printed for the failure, and the server's record of that same operation has its state, timings, the SQL execution it ran and the full error:

Because the Connect listener writes these events to the event log, the same endpoints work in the History Server after the application exits. Other Connect changes include the session context manager above (SPARK-58159), a SQLContext wrapper for existing code (SPARK-57021), an experimental SPARK_LOCAL_CONNECT_REUSE=1 setting that keeps a local Connect server alive between runs (SPARK-57787), and broader support in the Connect JDBC driver for complex types, DatabaseMetaData keys and SQLSTATE class 08 connection errors. The INVALID_HANDLE.SESSION_* errors changed SQLSTATE from HY000 to 08003, so retry logic that matches SQLSTATE may need updating. Connect Gateway and the Rust-backed Python client are concurrent projects, not Spark 4.3 features.

PySpark: DataFrame.zip and UDF transpilation

DataFrame.zip (SPARK-55886, with Connect support in SPARK-57247) places two column projections of the same DataFrame side by side, row for row, without a join key. It replaces joins built on monotonically_increasing_id(), whose rows are not guaranteed to align. Both sides must originate from the same plan, with matching filters, sorts, joins and aggregations beneath the projections.

The experimental Python UDF transpiler (SPARK-55206) is the one I've been waiting for since the UDF showdown, where the conclusion was that the fastest Python UDF is the one that never runs Python. With spark.sql.experimental.optimizer.transpilePyUDFs=true and ANSI mode on, PySpark reads simple UDF bodies when you define them and lowers them to Catalyst expressions, so len(dish) becomes Spark's own length and the BatchEvalPython step (and the Python worker, and the serialization) drops out of the plan:

spark = (SparkSession.builder .config("spark.sql.ansi.enabled", "true") # transpiling needs ANSI  .getOrCreate()) @udf("int", useArrow=False) def name_length(dish: str) -> int: return len(dish) 

The restrictions are not all documented together, so I read pyspark/sql/udf.py to collect them. Transpilation applies only to non-Arrow batched UDFs (hence useArrow=False), classic PySpark, ANSI mode, and function bodies that PySpark can recover from source, excluding an inline lambda passed directly to udf(). With ANSI mode disabled, PySpark warns and falls back because the generated expressions use ANSI semantics. The transpiler also preserves Python behavior where Spark differs: a transpiled len raises on NULL as CPython would, represented in the full plan by a raise_error branch. Parameter type hints keep the generated plan small. Other PySpark changes bound UDF input batches by bytes, implement more pandas API on Spark functions natively, remove Python 3.10 support, and raise the minimum NumPy version to 1.23.2.

Spark SQL: ASOF JOIN, JSON_TABLE, VARIANT edits and nanoseconds

ASOF JOIN (SPARK-58092) is the point-in-time join: for each row on the left, the single closest row on the right that satisfies a MATCH_CONDITION comparison, optionally within an equality ON or USING key. In my Rust client post I had to pull a Spark result into DuckDB to do exactly this, so it's nice to have it native, with a dedicated sort-merge operator. The canonical ML case is matching each prediction to the feature version that had actually been published when the prediction was made:

SELECT p.made_at, p.score, f.version AS features_used, f.published_at FROM predictions p ASOF JOIN feature_versions f MATCH_CONDITION (p.made_at >= f.published_at) -- latest one out by then 

The 10:03 prediction gets v1 from 10:00 and never v2 from 10:05, which didn't exist yet (using it would be leakage), and the 10:07 prediction gets v2 alone, where a plain >= join would return both versions and leave you to pick the latest yourself, which is exactly where the leakage bugs usually creep in. It's a preview: spark.sql.join.asofJoin.enabled is off by default (and internal, so it won't show up in SET -v), only INNER and LEFT OUTER are supported, and in streaming only stream-static works, with the stream on the left.

JSON_TABLE (SPARK-58366) is the SQL-standard table function for expanding a JSON array into typed rows. It also supports FOR ORDINALITY and EXISTS columns:

And VARIANT values can finally be edited in place with variant_set, variant_insert, variant_delete, variant_array_append and variant_strip_nulls (plus try_ versions), instead of being parsed into a struct and serialized back:

Nanosecond timestamps (SPARK-56876) add TIMESTAMP_NTZ(p) and TIMESTAMP_LTZ(p) for precisions 7 to 9, with unix_nanos, timestamp_nanos, current_timestamp(p), datetime functions, interval arithmetic, and Parquet, ORC, JSON, CSV, Avro, JDBC, Arrow and Connect support. Spark's TIMESTAMP has been microseconds since forever, which means two market events a nanosecond apart (or two rows from a Parquet file written by something that does use nanoseconds) silently became the same instant:

Nanosecond timestamps remain a preview behind spark.sql.timestampNanosTypes.enabled, and the Databricks release post warns that they may change incompatibly. Typed literals such as TIMESTAMP_NTZ(9) '...' do not parse yet, so values must be cast from strings. I would not use the type for production work yet: while building this demo, I found a query that returns an incorrect count when a nanosecond column appears beside other DISTINCT aggregates in the same SELECT; I am reducing the reproduction before filing it. The TIME type also gained nanosecond precision and broader connector support, but remains disabled behind an internal flag.

Other SQL additions include ANSI UNNEST in FROM, with WITH ORDINALITY; counter_diff, a window function that converts a running counter such as requests served or bytes sent into per-interval deltas and returns NULL after a reset; and jaro_w...