Stateless vs Stateful

Concepts covered: Structured Streaming

A stateless operator computes a batch's output from that batch's rows alone. A stateful operator also needs what earlier batches contained, so Spark keeps a summary of them between batches: the query's state. Any streaming query can be cut into 2 parts: the operators that run fresh on each batch, and the ones that read and update state.

Operator	Kind	What it keeps between batches
select, withColumn, filter	Stateless	Nothing
Join to a static table	Stateless	Nothing; the static side is read again each batch
union of 2 streams	Stateless	Nothing
groupBy(...).agg(...)	Stateful	1 running row per group
dropDuplicates on a key	Stateful	Every key value it has seen
Join of 2 streams	Stateful	Buffered rows from both sides, covered in a later lesson

Splitting a query

Here is a per-city leaderboard of drivers, built on the same trips stream and checkpointed to /chk/leaderboard/:

leaderboard = (
    trips
    .filter(F.col("status") == "completed")
    .join(cities, "city_id")
    .groupBy("city_name", "driver_id")
    .agg(
        F.count("*").alias("trips"),
        F.sum("fare").alias("earned"),
    )
)

Read it top down. The filter and the join to cities, a 2MB static table under the 10MB broadcast threshold, are stateless: they pipeline inside the first stage of each batch, and the broadcast adds no shuffle. The groupBy is the stateful part. It hash-shuffles the batch's rows on city_name and driver_id, and each task after the shuffle merges its slice of the batch into the running totals it holds for those keys.

Where the state shows up in leaderboard.explain()

Exchange hashpartitioning(city_name, driver_id, 200): the shuffle in front of the state, with 1 state partition per shuffle partition.
StateStoreRestore: reads each key's running totals from before this batch.
StateStoreSave: writes the merged totals back, to be checkpointed with the batch.

The animation plays 1 micro-batch of the same shape as Spark runs it, trimmed to a count per driver: 9 trips in 2 input partitions, a filter that drops the 2 cancelled trips inside the tasks that read them, and a shuffle that sends each driver's rows to 1 of 3 reducer tasks. In the live stream, each of those reducers also owns the state for the drivers that hash to it, which is why a driver must land on the same partition in every batch.

from pyspark.sql import functions as F

counts = (
    trips
    .filter(F.col("status") == "completed")
    .groupBy("driver_id")
    .agg(F.count("*").alias("trips"))
)
counts.collect()

trips
trip_id	driver_id	status	fare
1	d7	completed	14.5
2	d9	completed	22
3	d7	cancelled	0
4	d4	completed	9.75
5	d7	completed	31.2
6	d4	completed	12
7	d9	cancelled	0
8	d2	completed	18.4
9	d7	completed	7.6

That is 1 shuffle and 2 stages per batch, the plan a batch job would get. The streaming version adds a state read before the merge and a state write after it, in every reducer task, on every batch.

What state costs

Stateless part

Work per batch grows with the batch's row count
Holds nothing in executor memory between batches
Restart reads offsets and commits, then carries on
Filters and projections may change between restarts

Stateful part

Work per batch grows with the batch plus the state it touches
Holds state in executor memory and writes it under state/ each batch
Restart reloads state from the checkpoint into the tasks that need it
Grouping keys, aggregates and partition count are fixed by the checkpoint

The partition count is frozen

The number of state partitions is the value of spark.sql.shuffle.partitions at the query's first start, and Spark records it among the settings in each offsets file. Every later restart reuses the recorded value and ignores the one you set, because each key's state was filed under the partition its hash chose on the first day. At the default of 200, a stateful query runs 200 reducer tasks on every batch, even a batch of 12 rows, and each task loads and saves its slice of state. Adaptive query execution will not coalesce them: Spark turns it off for streaming queries and logs a warning if it's enabled. Choose the count from the expected state size before the first start.

State grows with its keys

State size follows the number of distinct keys an operator has seen, and a key stays until something removes it. The leaderboard keeps 1 row per city and driver, which levels off once every active driver has appeared. dropDuplicates(["event_uuid"]) keeps every event id it has ever seen: at 50 million events a day, a month of ids is 1.5 billion keys held across the executors and saved into the checkpoint. Bounding that growth by event time is a later lesson's subject; until then, size any stateful operator over an unbounded key before it ships.

The leaderboard has run for a month against /chk/leaderboard/. Which change, deployed against that same checkpoint, takes effect as intended?

Group by driver_id alone, dropping city_name from the keys
Tighten the filter to completed trips with a fare above 0
Raise spark.sql.shuffle.partitions from 200 to 400 for speed
Add a sum of tips beside the trips and earned aggregates

About this lesson

This section is part of the Structured Streaming: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.