Stateless vs stateful operations in Structured Streaming

• Stateless operations – Process each micro-batch independently using only the rows in that batch, keeping no memory of previous batches; examples are filter, select, map, and withColumn.
• Stateful operations – Produce output that depends on data from earlier batches, so Spark maintains state in a state store across micro-batches; examples are windowed aggregations, stream-stream joins, dropDuplicates, and flatMapGroupsWithState.
• Watermark requirement – Stateful operations need watermarks to bound their state; without one, state grows indefinitely.
• Recovery cost – Stateless queries recover cheaply by replaying from committed offsets; stateful queries must also restore the state store from the checkpoint, which is heavier.
• Design principle – Good streaming design pushes as much work as possible into cheap stateless steps and minimizes the stateful footprint.

What are stateless operations?

Stateless operations transform the rows of the current micro-batch with no dependence on earlier data. A filter that keeps only transactions above a threshold, a select that projects columns, a withColumn that computes a fee, or a map that reshapes a row all produce output purely from the incoming batch.

import static org.apache.spark.sql.functions.*;
import java.math.BigDecimal;

// All stateless -- each batch is independent
Dataset<Row> processed = transactions
    .filter(col("amount").gt(new BigDecimal("10000.00")))  // stateless
    .select("accountId", "amount", "eventTime")            // stateless
    .withColumn("fee",                                      // stateless
        col("amount").multiply(new BigDecimal("0.001")));

processed.writeStream()
    .outputMode("append")  // works because no state
    .format("parquet")
    .option("path", "s3a://nbi-data/high-value")
    .option("checkpointLocation", "s3a://nbi-checkpoints/high-value")
    .start();

What are stateful operations?

Stateful operations produce results that accumulate across batches, so Spark keeps a state store keyed by the grouping or join key. A windowed count must remember the running count for each open window, a stream-stream join must buffer rows from each side, and dropDuplicates must remember which keys have already been seen.

// Stateful -- running count per account (state carried across batches)
Dataset<Row> counts = transactions
    .withWatermark("eventTime", "10 minutes")
    .groupBy("accountId")
    .count();

// Stateful -- deduplication (remembers seen transaction IDs)
Dataset<Row> deduped = transactions
    .withWatermark("eventTime", "10 minutes")
    .dropDuplicates("transactionId", "eventTime");

// Stateful -- stream-stream join (buffers both sides)
Dataset<Row> enriched = orders
    .withWatermark("orderTime", "10 minutes")
    .join(
        payments.withWatermark("payTime", "10 minutes"),
        expr("orderId = payOrderId AND " +
             "payTime >= orderTime AND " +
             "payTime <= orderTime + interval 1 hour")
    );

How is stateless different from stateful?

| Aspect | Stateless | Stateful |
|---|---|---|
| Memory | No state across batches | Keyed state store |
| Operations | filter, select, map, withColumn | groupBy, join, dropDuplicates, flatMapGroupsWithState |
| Watermark | Not needed | Required to bound state |
| Output mode | Append or Update | Depends on query and watermark |
| Recovery | Replay from committed offsets (cheap) | Restore state store from checkpoint (heavy) |
| Scaling | Linear, trivial | State size drives cost |
| Checkpoint compat | Any query change is safe | Changing stateful logic can break checkpoint |

// Stateless recovery: just replay from last committed offsets
// Stateful recovery: replay + restore state store

Why do stateful operations need watermarks?

Because stateful operations hold state indefinitely in principle, they need watermarks to bound it. A windowed aggregation keeps state for each window and relies on the watermark to finalize and evict closed windows. A stream-stream join uses watermarks on both sides plus time constraints to know when buffered rows can be dropped.

// Without watermark -- state grows forever
transactions
    .groupBy(window(col("eventTime"), "5 minutes"), col("accountId"))
    .count();  // Every 5-min window's state kept indefinitely

// With watermark -- state is bounded
transactions
    .withWatermark("eventTime", "10 minutes")
    .groupBy(window(col("eventTime"), "5 minutes"), col("accountId"))
    .count();  // Windows closing before watermark have state evicted

How does recovery differ?

A stateless query recovers by replaying from checkpointed offsets, which is cheap and simple. A stateful query also restores its state store from the checkpoint, which is heavier and introduces compatibility constraints: changing the query's stateful logic can make an existing checkpoint incompatible, forcing a reset.

// Stateless: restart reads last committed offset, replays batch
// Time to recover: seconds

// Stateful: restart reads last committed offset AND
// restores the entire state store (per-account counts, window states)
// Time to recover: depends on state size (can be minutes for large state)

What is the RocksDB state store?

For large state, the default in-memory state store can cause executor memory pressure and garbage-collection pauses. The RocksDB state store keeps state off-heap on local disk, trading slightly higher per-access latency for the ability to handle much larger state without JVM heap pressure.

// Configure RocksDB state store for large state
spark.conf().set(
    "spark.sql.streaming.stateStore.providerClass",
    "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider"
);

// Now stateful operations store state on local SSD via RocksDB
// Handles hundreds of millions of keys without heap pressure

What are common mistakes?

The most common mistake is not realizing that an operation is stateful. A groupBy without window is still stateful and keeps per-key state across batches. Another mistake is using dropDuplicates without a watermark, causing Spark to remember every key forever.

// WRONG -- groupBy is stateful even without a window
transactions.groupBy("accountId").count();
// State for every account is kept forever without a watermark

// WRONG -- dropDuplicates without watermark
transactions.dropDuplicates("transactionId");
// Remembers every transaction ID forever -- unbounded state

// CORRECT -- add watermark to bound state
transactions
    .withWatermark("eventTime", "10 minutes")
    .dropDuplicates("transactionId", "eventTime");
// Only remembers IDs within the watermark window

What should a senior engineer know?

Good design pushes as much work as possible into cheap stateless steps and keeps the stateful footprint minimal. Pre-filter before an aggregation, choose tight but adequate watermarks, and use RocksDB when state is large. Knowing which operations are stateful is the first step to reasoning about a streaming job's cost and reliability.

// Senior pattern: stateless pre-filter before stateful aggregation
Dataset<Row> optimized = transactions
    .filter(col("status").equalTo("COMPLETED"))  // stateless pre-filter
    .filter(col("amount").gt(new BigDecimal("0")))  // stateless
    .withWatermark("eventTime", "10 minutes")
    .groupBy(window(col("eventTime"), "5 minutes"), col("accountId"))
    .agg(sum("amount").as("total"));  // stateful, but fewer rows