Spark transformWithState in Python: timers, TTL and state that outlives a micro-batch

transformWithState is how you write your own stateful logic in Spark Structured Streaming from Spark 4 on: you group a stream by a key, write a small Python class, and Spark keeps that class's state for each key between micro-batches and wakes it up when a timer you set runs out. It replaces applyInPandasWithState (and mapGroupsWithState in Scala), and both Apache Spark and Databricks now point new work at it.

You need it only when the built-in operators cannot say what you mean. A count per window, a dedupe or a stream join should stay groupBy().agg(), dropDuplicatesWithinWatermark and join; Databricks says the same. You need your own logic when what you emit depends on what has not happened yet: a cart nobody checked out, a sensor that went silent, a session that ended without a logout. That is the example we build here, and every output on this page is what Spark 4.2 printed when we ran it.

The job: flag carts nobody checked out

Events arrive as JSON: a user adds an item or checks out. A cart left alone for 30 minutes after its last add counts as abandoned, and we want one row for it so a reminder email can go out. A checkout should close the cart and emit a row too.

No aggregation can do this, because the abandoned row has to appear when nothing arrives for that user. That is the job of a timer.

The processor

A stateful processor is a class with 4 methods Spark calls for you. init declares the state variables, handleInputRows gets a key's new rows, handleExpiredTimer runs when a timer expires, close cleans up.

from datetime import datetime, timezone
from pyspark.sql import Row
from pyspark.sql.streaming import StatefulProcessor, StatefulProcessorHandle

WAIT_MS = 30 * 60 * 1000  # a cart left alone for 30 minutes counts as abandoned


class CartWatch(StatefulProcessor):
    def init(self, handle: StatefulProcessorHandle) -> None:
        self.handle = handle
        self.items = handle.getListState("items", "item string")
        self.last_seen = handle.getValueState("last_seen", "ts timestamp")

    def handleInputRows(self, key, rows, timerValues):
        user = key[0]
        for r in sorted(rows, key=lambda r: r.ts):
            if r.event == "add":
                self.items.appendValue((r.item,))
                self.last_seen.update((r.ts,))
            elif r.event == "checkout":
                n = len(list(self.items.get())) if self.items.exists() else 0
                self.items.clear()
                self.last_seen.clear()
                for t in self.handle.listTimers():
                    self.handle.deleteTimer(t)
                yield Row(user_id=user, outcome="checked_out", items=n, at=r.ts)
        if self.last_seen.exists():  # cart still open: one timer, 30 minutes after the last add
            for t in self.handle.listTimers():
                self.handle.deleteTimer(t)
            last = self.last_seen.get()[0]
            self.handle.registerTimer(int(last.timestamp() * 1000) + WAIT_MS)

    def handleExpiredTimer(self, key, timerValues, expiredTimerInfo):
        n = len(list(self.items.get())) if self.items.exists() else 0
        at = datetime.fromtimestamp(expiredTimerInfo.getExpiryTimeInMs() / 1000, tz=timezone.utc)
        self.items.clear()
        self.last_seen.clear()
        yield Row(user_id=key[0], outcome="abandoned", items=n, at=at)

    def close(self) -> None:
        pass

A few things in there are easy to get wrong the first time:

• State values are tuples. update((r.ts,)) with the trailing comma, get()[0] to read. Each variable has a name and a schema, written here as a SQL string; in Python the schema is required.
• One state variable per idea. The items are a ListState, because we only append; the last activity time is a ValueState. A MapState would fit if you needed lookups by a second key, such as quantity per item.
• State has no default. Check exists() before get().
• Timers are per key and identified by their time. Moving the deadline means deleting the old timer and registering a new one, which is what the loop over listTimers() does. You cannot register a timer in init.
• The key arrives as a tuple too, so the user is key[0].

The query

events = (spark.readStream
          .schema("user_id string, event string, item string, ts timestamp")
          .json(src)
          .withWatermark("ts", "10 minutes"))

carts = (events.groupBy("user_id")
         .transformWithState(
             statefulProcessor=CartWatch(),
             outputStructType="user_id string, outcome string, items int, at timestamp",
             outputMode="Append",
             timeMode="EventTime",
             eventTimeColumnName="at"))

timeMode="EventTime" means timers fire against the watermark, the time in the data minus the lateness you allow, not against the clock on the wall. eventTimeColumnName names the output column that carries event time, so a later stage downstream still has a watermark to work with.

What Spark printed, file by file

We dropped 3 small JSON files into the source folder, one at a time:

Ana's timer was set for 10:35, 30 minutes after her socks. File 2 contains nothing about Ana, yet she was still not flagged after it, because the watermark had only reached 10:10. Her row appeared only after Cy's lamp pushed the watermark to 10:40. Spark ran an extra batch with no input after each file to apply the new watermark (6 batches for 3 files), and that is where the expired timer fired.

That is the property to design around: an event-time timer fires when other data moves the watermark past it, not when the time passes. If the whole stream goes quiet at night, no cart is flagged until the first event in the morning. When that matters, use processing time.

Pick the time mode for what has to happen

Inside your methods, read the time from timerValues.getCurrentProcessingTimeInMs() or getCurrentWatermarkInMs(), not from datetime.now(). Databricks warns that the system clock makes retries after a task failure unreliable, since a retried batch would see a different time.

Timers act, TTL forgets

Both get rid of old state, but they are not the same tool:

• A timer runs your code. handleExpiredTimer can emit a row, clear state and set the next timer. Our abandoned-cart row exists only because of a timer.
• TTL runs no code. You pass ttlDurationMs when you create a state variable, and Spark drops the value silently once it has not been written for that long. Nothing is emitted.

self.views = handle.getValueState("views", "n long", ttlDurationMs=24 * 60 * 60 * 1000)

TTL is your guard against state that grows without end: users who visit once and never return would otherwise stay in the state store forever. Databricks' docs put it bluntly: without TTL, you have to evict state yourself. The rules differ by type. A ValueState value's TTL restarts each time you write it. In a ListState each element has its own TTL, and only put, which rewrites the whole list, restarts them. In a MapState each key has its own. TTL works only in ProcessingTime mode.

Row or pandas: transformWithState or transformWithStateInPandas

PySpark has 2 versions of the operator, with the same processor methods:

For event logic like ours, with a few rows per key per batch, the row API is the simpler read, and it is the one Databricks real-time mode supports, if latency ever matters.

The setup errors you hit first

We hit both of these on a clean Spark 4.2 install, so you probably will too:

1. [PACKAGE_NOT_INSTALLED] Pandas >= 2.2.0 must be installed, even with the row API. The Python operator runs in a Python worker that talks to the JVM through Arrow and protobuf, so install pandas, pyarrow and protobuf next to pyspark. Without protobuf, the query dies with "TransformWithStateInPySpark driver worker exited unexpectedly", and the real cause (cannot import name 'descriptor' from 'google.protobuf') is only in the log. PySpark 4.2 also warns that it does not fully support pandas 3 yet, so pin pandas<3.
2. [UNSUPPORTED_FEATURE.STORE_BACKEND_NOT_SUPPORTED_FOR_TWS]. The operator runs only on the RocksDB state store, and open-source Spark still defaults to the HDFS-backed one:

spark.conf.set(
    "spark.sql.streaming.stateStore.providerClass",
    "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider",
)

On Databricks you need Databricks Runtime 16.2 or later (16.3 for Python on standard access mode) (16.3 for Python on standard access mode), and RocksDB is already the default from 17.3. Below that, set the line above.

Test the processor without a stream

Spark 4.2's guide documents TwsTester, which runs your processor in memory: feed it rows for a key, look at the state, move the watermark by hand. This is the same CartWatch class, and this is what it printed:

from pyspark.sql import Row
from pyspark.sql.streaming import TwsTester

def t(hhmm):  # 23 Oct 2026, UTC
    return datetime(2026, 10, 23, int(hhmm[:2]), int(hhmm[3:]), tzinfo=timezone.utc)

tester = TwsTester(CartWatch(), timeMode="EventTime",
                   eventTimeExtractor=lambda r: int(r.ts.timestamp() * 1000))
print(tester.test("ana", [Row(user_id="ana", event="add", item="boots", ts=t("10:00")),
                          Row(user_id="ana", event="add", item="socks", ts=t("10:05"))]))
print(tester.peekListState("items", "ana"))
print(tester.setWatermark(int(t("10:30").timestamp() * 1000)))   # 10:30: too early
print(tester.setWatermark(int(t("10:36").timestamp() * 1000)))   # 10:36: timer at 10:35 fires

[]
[('boots',), ('socks',)]
[]
[Row(user_id='ana', outcome='abandoned', items=2, at=datetime.datetime(2026, 10, 23, 10, 35, tzinfo=datetime.timezone.utc))]

In event-time mode the tester needs eventTimeExtractor, a function from a row to milliseconds. It does not model TTL or move the watermark on its own, so keep one end-to-end test that runs a real stream, like the 3-file run above.

Moving off applyInPandasWithState

The old operator gave you one GroupState object per key and one timeout. The new one splits those into named pieces:

To carry over the state an old query has built, read it from its checkpoint with the state data source and pass it in as initialState: Databricks lists migrating an existing workflow as one of the uses of initial state, and the same route seeds a processor from a Delta table.

Once it runs, the same state data source reads the new operator's state one variable at a time (option stateVarName, and readRegisteredTimers for the timers). That is the quickest way to see why a key did or did not fire.

Where to go next

For the ideas underneath (micro-batches, watermarks, output modes), start with streaming and watermarks. What else Spark 4 changed in Python and streaming is in what's new in Apache Spark 4.0 and Spark 4.2's features. If the cart alert has to arrive in milliseconds rather than at the next micro-batch, read Spark real-time mode: on Databricks, real-time mode runs the row-based transformWithState one row at a time, with no event-time timers.