Structured Streaming: micro-batches, state & exactly-once

The unbounded table and the micro-batch loop

Structured Streaming models a stream as an input table that only grows: every arriving record is a new row. You write an ordinary DataFrame query against that table, and the engine keeps the result table up to date without ever materializing the input. Each trigger reads only the data that arrived since the previous batch, combines it with the intermediate state the query kept, writes what changed to a sink, and discards the source rows.

The driver owns the loop. For micro-batch N it asks each source for its latest available offsets, writes the chosen range to the checkpoint, and plans the batch as a normal Spark SQL execution (an IncrementalExecution in the source), so Catalyst, shuffles, stages, and tasks all apply per batch. Executor tasks read exactly that range, run the operators, load and save state-store versions for stateful operators, and write the sink output. When the batch finishes, the driver records it as committed and plans N+1.

Stateful queries also run no-data batches: batches with zero input rows whose only job is to advance the watermark, emit finished results, and evict state. In the watermark runs behind this lesson, a no-data batch followed each data batch that raised the watermark — for example batch 1, with 0 rows, running at a watermark of 11:57.

The word “checkpoint” means something different here than in cache and recovery: a streaming checkpoint is one query's progress log plus its state, not lineage truncation for a batch DataFrame.

Check the sources: Structured Streaming guide: input and result tables · MicroBatchExecution source

Triggers decide when a batch starts

A trigger controls timing, not what a batch contains. Every micro-batch trigger still plans, runs, and commits batches the same way.

Continuous processing is a different engine, not a faster trigger. The guide quotes latencies around 1 ms against about 100 ms at best for micro-batches, but the part interviewers probe is the guarantee: continuous mode is at-least-once. It supports only map-like operations (projections and selections; SQL functions except aggregations, current_timestamp(), and current_date()), the Kafka and rate sources, and the Kafka, memory, and console sinks. It needs a core for every source partition, does not retry failed tasks automatically, and cannot use foreachBatch. Locally, a rate-source filter into a memory sink ran under a continuous trigger; an aggregation failed with Continuous processing does not support Aggregate operations, and a file source was rejected the same way.

Check the sources: Guide: triggers · PySpark DataStreamWriter.trigger · Guide: continuous processing

Exactly-once: replayable sources, the checkpoint, idempotent sinks

End-to-end exactly-once is not one atomic commit. It comes from replaying a batch deterministically after a failure into a sink that tolerates the replay. The write order in MicroBatchExecution is:

1. The driver writes offsets/N: the write-ahead record of the batch's source offsets, plus the batch watermark, batch timestamp, and a few pinned settings.
2. Tasks process that range; stateful operators write the next state version under state/.
3. The sink writes the batch output.
4. The driver writes commits/N.

On restart Spark compares the latest offsets entry with the latest commit: a committed batch moves the query to N+1, an uncommitted one is re-run over the same offsets. After two batches of a stateless file-to-Parquet query, the checkpoint held metadata, offsets/0, offsets/1, commits/0, commits/1, sources/0/0, and sources/0/1 — the last two recording which input file each batch took. Stateful queries add state/. A restart on that checkpoint with no new files read zero rows; adding one file read exactly one.

Replay only helps when both ends cooperate:

• The source must be replayable. The file source (files moved into the directory atomically) and Kafka qualify. The Kafka source keeps its offsets in the checkpoint and does not commit them to Kafka, and data that has aged out of the topic cannot be replayed (failOnDataLoss). The socket source is not fault-tolerant. Kafka behavior here is documentation-reviewed only.
• The sink must be idempotent or transactional. The file sink is documented as exactly-once; the Kafka and foreach sinks are at-least-once; console and memory are not fault-tolerant; foreachBatch is as safe as your function. Delta table writes are covered in Delta Lake.

foreachBatch passes each batch as a DataFrame plus its batchId; a replayed batch carries the same id and, for a deterministic query, the same rows. In a crash test after the write but before commits/1, a plain Parquet append duplicated the replayed row, while a Delta write keyed by the batch id skipped it:

def write_batch(batch_df, batch_id):
    (batch_df.write.format("delta").mode("append")
        .option("txnAppId", "orders-to-silver")   # stable for this checkpoint
        .option("txnVersion", batch_id)           # a replayed batch is skipped
        .save(target_path))

query = (orders.writeStream.foreachBatch(write_batch)
         .option("checkpointLocation", checkpoint_path)
         .trigger(availableNow=True).start())
query.awaitTermination()

If the function writes the batch more than once, persist it first: without that, a two-row batch reported numInputRows of 6 because it was computed three times.

Check the sources: MicroBatchExecution: offset and commit logs · Guide: fault-tolerance semantics · Guide: sources and sinks · Guide: foreachBatch · Kafka integration guide · Delta Lake: idempotent writes in foreachBatch

State, watermarks, and output modes

Streaming aggregations, dropDuplicates, stream-stream joins, and mapGroupsWithState/flatMapGroupsWithState keep state between batches in a versioned state store: by default in executor JVM memory backed by checkpoint files, or with the RocksDB provider in native memory and local disk. State shrinks only when something tells the engine a key can no longer change. That is the watermark's job.

How the watermark moves. At the end of a batch the engine takes the maximum event time it has seen and subtracts the withWatermark delay; the next batch uses that value. It follows event time, not the wall clock, so if data stops the watermark stops. With several watermarked inputs the global watermark is the minimum by default (spark.sql.streaming.multipleWatermarkPolicy).

When late data is dropped. The documented guarantee runs one way: a row within the delay is never dropped, and an older row may or may not be. For a windowed aggregation Spark's test is the window's end, not the row's timestamp. With a 10-minute delay and 5-minute windows:

• Rows at 12:01, 12:02, and 12:07 set the watermark to 11:57. A later batch with 12:20 and a late 12:03 still counted 12:03 (the 12:00 window became 3), then the watermark moved to 12:10.
• A 12:04 row arriving next was dropped: its window ended at 12:05, not after 12:10. numRowsDroppedByWatermark reported 1.
• In a separate run with the watermark at 12:12, a 12:11:30 row was still counted because its window ends at 12:15, while a 12:09 row was dropped.

Watermarks evict aggregation state only in append or update mode, only when the aggregation groups by the watermarked column or a window on it, and only when withWatermark comes before the aggregation.

For deduplication where a retried event carries a new timestamp, Spark 3.5 adds dropDuplicatesWithinWatermark: it keys on the id alone and keeps state only while copies can still fall within the watermark delay.

Check the sources: Guide: late data and watermarking · statefulOperators source: window-end eviction · Guide: output modes · Guide: RocksDB state store

What you may change on the same checkpoint

A restart on the same checkpoint reuses the stored offsets, commits, and state, so the new query must still fit them. The guide's “Recovery Semantics after Changes in a Streaming Query” sorts changes into allowed (possible, with semantics that depend on the query) and not allowed (the restarted query is likely to fail with unpredictable errors).

• Allowed: adding or removing filters; projections with the same output schema; adding, removing, or changing rate limits such as maxOffsetsPerTrigger; changing a Kafka output topic; changing foreach writer code or the code inside a state-mapping function.
• Not allowed: changing the number or type of sources; generally, changing subscribed topics or files; changing a file sink's output path; changing any stateful operation's schema — grouping keys or aggregates, deduplication columns, stream-stream join columns or join type, user state schema or timeout type.
• Fixed at the first run: spark.sql.shuffle.partitions (state is hash-partitioned by key), the state store provider class, and the multiple-watermark policy. Each offsets/N records them, and on restart Spark overwrites the session value with the recorded one.

Do not treat a successful restart as proof that a change was safe. Spark compares the stored state schema with the new one by field count, data types, and nullability compatibility, not by field names. On one checkpoint, changing count() to sum("amount") failed with StateSchemaNotCompatible, and so did adding a second grouping key. Swapping groupBy("city") for groupBy("country") — still one string key — ran without error, and a row with country “Delhi” came out with count 2 because it reused the old city key's state. Adding a filter in front of the same aggregation worked as documented.

Check the sources: Guide: recovery semantics after changes · Guide: configurations fixed after the first run · OffsetSeq source: pinned configurations

Reading lastProgress

query.lastProgress returns the most recent StreamingQueryProgress (a dictionary in Python), and query.recentProgress keeps the last few. Read several entries, not one.

• id stays the same across restarts from one checkpoint — locally it matched the value in metadata. runId identifies one start of the query, and batchId is the micro-batch number.
• numInputRows, inputRowsPerSecond (the rate data is arriving), and processedRowsPerSecond (the rate Spark is processing it).
• durationMs: latestOffset and getBatch for source discovery, queryPlanning, walCommit and commitOffsets for the offset and commit logs, addBatch for running the batch and writing the sink, and triggerExecution for the whole trigger.
• eventTime: watermark, plus max, min, and avg event time when the batch has rows.
• stateOperators: numRowsTotal, numRowsUpdated, numRowsRemoved, numRowsDroppedByWatermark, memoryUsedBytes, and provider-specific customMetrics.
• sources: startOffset, endOffset, latestOffset, and per-source rates.

A query that is falling behind shows processedRowsPerSecond staying below inputRowsPerSecond, or triggerExecution longer than the trigger interval. Then find where the time goes: a large addBatch points at processing or the sink, a growing numRowsTotal points at state, and slow walCommit or commitOffsets points at checkpoint storage. Two cautions from local runs: the file source reported latestOffset as null, and a continuous-mode query reported no progress at all.

Check the sources: Guide: reading metrics interactively · StreamingQueryProgress source

Try it on a small dataset

Run this in a PySpark notebook with an existing spark session. Create a local session if needed. The example was run for review; this website does not execute Spark.

import json, os, tempfile
from pyspark.sql import functions as F

spark.conf.set("spark.sql.session.timeZone", "UTC")
base = tempfile.mkdtemp()
events_dir, checkpoint = os.path.join(base, "events"), os.path.join(base, "checkpoint")
os.makedirs(events_dir)

def add_file(name, times):
    tmp = os.path.join(base, name)
    with open(tmp, "w") as f:
        f.writelines(json.dumps({"ts": t}) + "\n" for t in times)
    os.rename(tmp, os.path.join(events_dir, name))  # atomic move into the source

counts = (
    spark.readStream.schema("ts TIMESTAMP").json(events_dir)
    .withWatermark("ts", "10 minutes")
    .groupBy(F.window("ts", "5 minutes"))
    .count()
)

def show_batch(batch_df, batch_id):
    rows = batch_df.select(F.date_format("window.start", "HH:mm").alias("start"), "count").collect()
    print("batch", batch_id, sorted((r.start, r["count"]) for r in rows))

def run_until_caught_up():
    query = (counts.writeStream.outputMode("update").foreachBatch(show_batch)
             .option("checkpointLocation", checkpoint)
             .trigger(availableNow=True).start())
    query.awaitTermination()
    for p in query.recentProgress:
        print("  progress", p["batchId"], "rows", p["numInputRows"],
              "watermark", p["eventTime"]["watermark"][11:16],
              "dropped", p["stateOperators"][0]["numRowsDroppedByWatermark"])

add_file("1.json", ["2026-09-15T12:01:00Z", "2026-09-15T12:02:00Z", "2026-09-15T12:07:00Z"])
run_until_caught_up()
add_file("2.json", ["2026-09-15T12:20:00Z", "2026-09-15T12:03:00Z"])
run_until_caught_up()
add_file("3.json", ["2026-09-15T12:04:00Z"])
run_until_caught_up()

Ran on local PySpark 3.5.7 (local[2], JDK 17, Delta Lake 3.2.1 for the foreachBatch block) on 2026-09-15; output above is observed.

A query crashes after its sink wrote micro-batch 7 but before commits/7 existed. What happens on restart, and what must be true for the output to stay exactly-once?

Say your answer out loud before opening the notes.