Asynchronous processing with transformWithState is now generally available

Asynchronous processing for the Python row-based transformWithState API is now generally available (GA). Built on asyncio, this feature executes state operations and user processing logic concurrently, reducing idle time and boosting streaming throughput.

Use cases

Imagine you maintain a stateful streaming pipeline that groups records by key, updates running aggregates, or enriches data via external REST endpoints. Synchronous processing blocks worker threads during state retrieval and network calls.

With asynchronous processing, you can:

• Run state updates across grouping keys concurrently without needing third-party async streaming libraries.
• Trigger parallel external HTTP requests via asyncio.gather within your stateful processing logic.
• Batch inter-process communication automatically to maximize overall query throughput.

How to use

To switch from synchronous logic, implement AsyncStatefulProcessor instead of StatefulProcessor, define the processing methods using async def, and consume state read/write operations using await.

from pyspark.sql import Row
from pyspark.sql.streaming import AsyncStatefulProcessor, AsyncStatefulProcessorHandle
from pyspark.sql.types import StructType, StructField, LongType

value_schema = StructType([StructField("count", LongType(), True)])

class AsyncCountProcessor(AsyncStatefulProcessor):
  async def init(self, handle: AsyncStatefulProcessorHandle) -> None:
    self.count = handle.getValueState("count", value_schema)

  async def handleInputRows(self, key, rows, timerValues):
    total = (await self.count.get() or (0,))[0]
    for _ in rows:
      total += 1
    await self.count.update((total,))
    yield Row(action=key[0], count=total)

  async def close(self) -> None:
    pass

What happens in this code:

1. State variable instantiation via getValueState inside init stays synchronous.
2. The handler method handleInputRows is declared as an async coroutine (async def).
3. Reading (self.count.get()) and updating (self.count.update()) state values use the await keyword.

To run the query, pass the processor to transformWithState as usual:

q = (
  events.groupBy("action")
    .transformWithState(
      statefulProcessor=AsyncCountProcessor(),
      outputStructType=output_schema,
      outputMode="Update",
      timeMode="None",
    )
    .writeStream.format("memory")
    .queryName("async_counts")
    .trigger(availableNow=True)
    .start()
)
q.awaitTermination()

What happens in this code:

1. The transformWithState query syntax remains identical to synchronous queries.
2. Because synchronous and asynchronous processors share the same internal state structure, queries can switch implementations using the same checkpoint directory.

Limitations and requirements

• Requires Databricks Runtime 19 or higher, or serverless compute environment version 6 or higher.
• In Lakeflow pipelines, the pipeline channel must be set to PREVIEW.
• Supported only for the Python row-based transformWithState API. Not supported for transformWithStateInPandas or the Scala API.
• State operations that yield collections (such as AsyncListState.get or AsyncMapState.keys) return async iterators that must be consumed using async for.