What's new in Spark Structured Streaming in Spark 4.x?

Summary

• Three Spark 4.x feature releases have shipped (4.0.0 in May 2025, 4.1.0 in December 2025, 4.2.0 in July 2026), and each one changed Structured Streaming.
• Spark 4.0 introduced transformWithState, the new arbitrary stateful operator, plus a State Data Source that reads a query's state from its checkpoint (still marked experimental).
• Spark 4.1 introduced Real-time Mode (selected with Trigger.RealTime, for stateless Scala queries), a row-based Python transformWithState, and AQE support for stateless streaming queries.
• Spark 4.2 brought the Real-time Mode trigger to PySpark, stable names for streaming sources so a query can add, remove or reorder sources without breaking its checkpoint, and stream-stream join updates.
• Python streaming data sources arrived in 4.0, and 4.2 gave their readers admission control and Trigger.AvailableNow support.

What's new in Spark Structured Streaming in Spark 4.x?

As of October 2026, the Spark 4.x line has three feature releases, and Structured Streaming changed in each one. Spark 4.0.0 (May 23, 2025) introduced transformWithState for stateful logic and made state queryable. Spark 4.1.0 (December 16, 2025) introduced Real-time Mode, initially for Scala. Spark 4.2.0 (July 14, 2026) brought that mode to PySpark and let you name streaming sources so a query can evolve without a new checkpoint. If you're starting a new stateful pipeline on 4.x, build it on transformWithState: the Spark docs encourage it for complex stateful applications.

Release-by-release changes

From the official Apache Spark release notes:

Release	Notable Structured Streaming changes
4.0.0 (May 2025)	Arbitrary State API v2 (transformWithState) for Scala and Java, and transformWithStateInPandas for Python; State Data Source reader; Python streaming data sources; maxBytesPerTrigger for the file source
4.1.0 (Dec 2025)	Real-time Mode (Scala stateless support); row-based Python transformWithState; stream-stream joins on virtual column families; AQE and changeable shuffle partitions for stateless queries; RocksDB memory integrated with the unified memory manager; automatic state store snapshot repair
4.2.0 (Jul 2026)	Real-time Mode trigger in PySpark; named sources (DataStreamReader.name() and IDENTIFIED BY); sink naming backed by a V3 commit log; stream-stream non-outer join in Update mode; join state format V4; an error on inconsistent checkpoint metadata

transformWithState, the new stateful operator (4.0)

transformWithState supersedes mapGroupsWithState and flatMapGroupsWithState in Scala and applyInPandasWithState in Python; the docs call it their next-generation replacement. You define a StatefulProcessor class with typed state variables (value, list and map state), optional TTL-based eviction, and timers that are checkpointed with the query. Python has two forms: transformWithStateInPandas, which uses the pandas interface, and transformWithState, which uses the Row interface and was added in 4.1. Schema evolution inside a state variable works when spark.sql.streaming.stateStore.encodingFormat is set to avro. The example uses timeMode="None", which rules out timers and TTL.

from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle
from pyspark.sql.types import StructField, StructType, TimestampType

class DownTimeDetector(StatefulProcessor):
    def init(self, handle: StatefulProcessorHandle) -> None:
        state_schema = StructType([StructField("value", TimestampType(), True)])
        self.handle = handle
        self.last_seen = handle.getValueState("last_seen", state_schema)

    def handleInputRows(self, key, rows, timerValues):
        ...  # read and update self.last_seen, then yield output Rows

    def close(self) -> None:
        pass

q = (df.groupBy("key")
     .transformWithState(statefulProcessor=DownTimeDetector(),
                         outputStructType=output_schema,
                         outputMode="Update", timeMode="None"))

Reading state with the State Data Source (4.0, experimental)

The State Data Source reads a streaming query's state from its checkpoint with a batch query. format("statestore") returns state rows and format("state-metadata") returns operator metadata. Options include batchId to read an earlier batch, readChangeFeed with changeStartBatchId to see how state changed across micro-batches, and stateVarName for transformWithState variables. The docs mark the source experimental, so its options and output may change.

state = spark.read.format("statestore").load("/checkpoints/orders")
changes = (spark.read.format("statestore")
           .option("readChangeFeed", True)
           .option("changeStartBatchId", 5)
           .load("/checkpoints/orders"))

Real-time Mode (4.1 for Scala, 4.2 for PySpark)

Real-time Mode arrived in Spark 4.1.0 as a Scala feature for stateless queries, chosen with Trigger.RealTime. The trigger takes a batch duration, and Trigger.RealTime() with no argument defaults to 5 minutes. Spark 4.2.0 added the PySpark form, trigger(realTime="5 seconds"); the whole trigger() API is marked evolving. Neither release lists stateful support, so treat Real-time Mode as stateless for now. It's a separate trigger from continuous processing (Trigger.Continuous), which the docs still mark experimental.

Python sources, naming and state store changes

• Python streaming data sources (4.0). Implement DataSourceStreamReader or SimpleDataSourceStreamReader and DataSourceStreamWriter in Python. In 4.2, readers gained admission control (implement getDefaultReadLimit() and have latestOffset(start, limit) honor the limit) and Trigger.AvailableNow support.
• Source naming (4.2). DataStreamReader.name() and IDENTIFIED BY give sources stable names, so you can add, remove and reorder sources without breaking checkpoints. The PySpark docstring marks name() experimental. Sink naming is in the release notes too, but the 4.2.0 DataStreamWriter API reference shows no public sink name() method.
• State store hardening (4.1 and 4.2). RocksDB memory usage now integrates with the unified memory manager, file-level checksums verify RocksDB state files, snapshots repair automatically, and row checksums detect corruption.

How Databricks approaches Structured Streaming 4.x features

Databricks Runtime 16.2 and later include transformWithState, and from Databricks Runtime 17.3 the default state store provider is RocksDB. Databricks documents real-time mode as a Structured Streaming trigger type for operational workloads, with support for Scala, Java and Python; its reference page lists the supported sources, sinks and operators. For declarative pipelines, Lakeflow Pipelines runs streaming tables and materialized views in triggered or continuous mode.

FAQs

Does transformWithState replace mapGroupsWithState?

The Spark docs call it the next-generation replacement. Moving over means rewriting your logic as a StatefulProcessor class, so plan it as a migration, not a drop-in swap.

How does Real-time Mode relate to continuous processing?

They're different triggers. Trigger.Continuous runs the experimental continuous processing mode, while Trigger.RealTime selects the newer Real-time Mode from Spark 4.1.0.

Can I inspect the state of a streaming query?

Yes. Run a batch query with the State Data Source against the query's checkpoint location, as shown above.

Which 4.x release should I target for streaming?

Target 4.2.0 if you can. It's the release that added the PySpark Real-time Mode trigger, named streaming sources, and admission control for Python streaming readers.