How does Structured Streaming achieve exactly-once processing?

The checkpoint records which offsets have been processed, and the Delta sink commits each micro-batch atomically with its batch id. A replayed batch is recognised and skipped, so a restart after failure neither loses nor duplicates rows.

What is a watermark and why do I need one?

A watermark tells the engine how late an event may arrive, which lets it drop state older than that bound. Without one, a streaming aggregation keeps every key forever and the job eventually fails on memory — the most common cause of a stream that dies after days of running.

When should I use foreachBatch?

When the sink operation is not expressible as a streaming write — a `MERGE` upsert, writing to two tables, or calling an external system. Each micro-batch arrives as a normal DataFrame with a batch id you can use to make the operation idempotent.

Can I change a streaming query's code and reuse the checkpoint?

Some changes are safe — filters, added columns, output sinks in some cases. Changing the aggregation keys, the watermark column, or the source is not, and will either fail or produce wrong results. When in doubt, start a new checkpoint and reprocess.

Structured Streaming on Databricks

A streaming query is a batch query the engine runs repeatedly, keeping track of what it has already read. Almost everything that goes wrong in production comes from two things: the checkpoint, and unbounded state.

Reading a stream

from pyspark.sql.functions import col, from_json, current_timestamp
from pyspark.sql.types import StructType, StringType, LongType, DecimalType, TimestampType

schema = (StructType()
    .add("order_id", LongType())
    .add("customer_id", LongType())
    .add("status", StringType())
    .add("amount", DecimalType(10, 2))
    .add("occurred_at", TimestampType()))

raw = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka.internal:9092")
    .option("subscribe", "orders")
    .option("startingOffsets", "latest")
    .option("maxOffsetsPerTrigger", 100000)
    .load())

orders = (raw
    .select(from_json(col("value").cast("string"), schema).alias("j"), col("timestamp").alias("kafka_ts"))
    .select("j.*", "kafka_ts")
    .withColumn("_ingested_at", current_timestamp()))

(orders.writeStream
    .option("checkpointLocation", "/Volumes/bookshop/raw/_checkpoints/orders_stream")
    .trigger(processingTime="30 seconds")
    .toTable("bookshop.bronze.orders_stream"))

{
  "id": "8f2c1a44-8e21-4b0e-9a3c-1d84f0b27a51",
  "runId": "3d9c4b21-77a1-4e02-b8f1-5c2e91a4d883",
  "batchId": 41,
  "numInputRows": 88412,
  "inputRowsPerSecond": 2947.06,
  "processedRowsPerSecond": 4102.88,
  "durationMs": {"addBatch": 18204, "getBatch": 12, "walCommit": 402, "triggerExecution": 21544},
  "sources": [{
    "description": "KafkaV2[Subscribe[orders]]",
    "startOffset": {"orders": {"0": 41209882, "1": 41198204}},
    "endOffset":   {"orders": {"0": 41254102, "1": 41242411}},
    "numInputRows": 88412
  }],
  "sink": {"description": "DeltaSink[bookshop.bronze.orders_stream]", "numOutputRows": 88412}
}

The number to watch is processedRowsPerSecond against inputRowsPerSecond. Processing 4,102/s while 2,947/s arrive means the stream is keeping up with headroom. When input exceeds processed for long, the backlog grows without bound.

maxOffsetsPerTrigger caps each batch. Without it, a stream restarted after an outage tries to process the entire backlog in one batch and typically dies — set it to roughly what a healthy batch handles.

The checkpoint

dbutils.fs.ls("/Volumes/bookshop/raw/_checkpoints/orders_stream/")

[FileInfo(path='.../commits/', name='commits/'),
 FileInfo(path='.../offsets/', name='offsets/'),
 FileInfo(path='.../metadata', name='metadata'),
 FileInfo(path='.../state/', name='state/')]

offsets records what each batch intends to read; commits records what finished. On restart the engine reads the last offset file, sees whether a matching commit exists, and either replays that batch or moves on.

Exactly-once holds because the Delta sink writes the batch id into the commit:

describe history bookshop.bronze.orders_stream limit 2;

version  operation         operationMetrics                                    engineInfo
-------  ----------------  --------------------------------------------------  --------------
     41  STREAMING UPDATE  {numOutputRows: 88412, numAddedFiles: 4}             Databricks-Runtime/16.4
     40  STREAMING UPDATE  {numOutputRows: 91204, numAddedFiles: 4}             Databricks-Runtime/16.4

A replayed batch 41 is recognised and skipped rather than appended twice. Three rules follow:

One checkpoint per query. Two streams sharing one is corruption, not a race.
Deleting the checkpoint reprocesses everything. On an append sink, that is duplicates.
The checkpoint is production state. It belongs in governed storage, not /tmp.

Windows and watermarks

from pyspark.sql.functions import window, sum as _sum, count as _count

revenue = (spark.readStream.table("bookshop.bronze.orders_stream")
    .withWatermark("occurred_at", "10 minutes")
    .groupBy(window("occurred_at", "5 minutes"), "status")
    .agg(_sum("amount").alias("revenue"), _count("*").alias("orders")))

(revenue.writeStream
    .option("checkpointLocation", "/Volumes/bookshop/raw/_checkpoints/revenue_5m")
    .outputMode("append")
    .trigger(processingTime="1 minute")
    .toTable("bookshop.gold.revenue_5m"))

select window.start, status, orders, revenue
from bookshop.gold.revenue_5m
order by window.start desc limit 4;

start                status     orders  revenue
-------------------  ---------  ------  --------
2026-09-09 14:35:00  completed    2841  71204.50
2026-09-09 14:35:00  returned      188   4820.00
2026-09-09 14:30:00  completed    2914  73011.20
2026-09-09 14:30:00  returned      201   5102.44

withWatermark("occurred_at", "10 minutes") is the load-bearing line. It promises that events arrive at most 10 minutes late, which lets the engine finalise a window and discard its state. Without it:

"stateOperators": [{
  "numRowsTotal": 88412004,
  "memoryUsedBytes": 12884901888,
  "numRowsDroppedByWatermark": 0
}]

88 million state rows and 12 GB of memory, growing forever, until the job fails on a day nobody changed anything. With the watermark:

"stateOperators": [{
  "numRowsTotal": 4102,
  "memoryUsedBytes": 8388608,
  "numRowsDroppedByWatermark": 118
}]

4,102 rows and 8 MB, and 118 events arrived too late and were dropped. That trade is the whole point — a longer watermark keeps more late data at the cost of more state.

Output modes interact with this:

Mode	Emits	Needs a watermark
`append`	rows only once finalised	yes, for aggregations
`update`	rows that changed this batch	recommended
`complete`	the entire result each batch	no, but state is unbounded

complete on an unbounded aggregation is the other way to run out of memory.

For very large state, RocksDB keeps it off the JVM heap:

spark.conf.set("spark.sql.streaming.stateStore.providerClass",
               "com.databricks.sql.streaming.state.RocksDBStateStoreProvider")

foreachBatch

A streaming write cannot express a MERGE. foreachBatch hands you each micro-batch as a normal DataFrame:

from delta.tables import DeltaTable

def upsert_batch(batch_df, batch_id):
    # deduplicate within the batch — MERGE rejects multiple matches for one key
    latest = (batch_df
        .withColumn("rn", row_number().over(
            Window.partitionBy("order_id").orderBy(col("occurred_at").desc())))
        .filter("rn = 1").drop("rn"))

    (DeltaTable.forName(spark, "bookshop.silver.orders")
        .alias("t")
        .merge(latest.alias("s"), "t.order_id = s.order_id")
        .whenMatchedUpdateAll(condition="s.occurred_at > t.occurred_at")
        .whenNotMatchedInsertAll()
        .execute())

(spark.readStream.table("bookshop.bronze.orders_stream")
    .writeStream
    .foreachBatch(upsert_batch)
    .option("checkpointLocation", "/Volumes/bookshop/raw/_checkpoints/silver_upsert")
    .trigger(availableNow=True)
    .start())

{"batchId": 12, "numInputRows": 88412, "durationMs": {"addBatch": 42104}}

Two details that matter. The in-batch deduplication is required — a MERGE whose source has two rows for one key fails outright:

UnsupportedOperationException: Cannot perform Merge as multiple source rows matched
and attempted to modify the same target row in the Delta table.

And foreachBatch gives at-least-once, not exactly-once, because your function may be re-executed for the same batch_id. MERGE is naturally idempotent so this is safe; a plain append inside foreachBatch is not. Use batch_id to guard anything that is not:

if spark.sql(f"select count(*) from processed_batches where batch_id = {batch_id}").head()[0]:
    return

Joining a stream to a table

enriched = (spark.readStream.table("bookshop.bronze.orders_stream")
    .join(spark.table("bookshop.silver.customers"), "customer_id", "left"))

A stream-static join re-reads the static side on every micro-batch, so dimension updates are picked up automatically — and a large dimension is re-read every batch, which is worth broadcasting.

Stream-stream joins need watermarks on both sides plus a time bound, or state grows forever:

clicks = spark.readStream.table("bronze.clicks").withWatermark("clicked_at", "30 minutes")
orders = spark.readStream.table("bronze.orders_stream").withWatermark("occurred_at", "30 minutes")

attributed = clicks.join(orders,
    expr("""clicks.session_id = orders.session_id
            and orders.occurred_at between clicks.clicked_at
                                       and clicks.clicked_at + interval 1 hour"""))

Monitoring

class Listener(StreamingQueryListener):
    def onQueryProgress(self, event):
        p = event.progress
        print(f"{p.name} batch={p.batchId} in={p.numInputRows} "
              f"rate={p.processedRowsPerSecond:.0f}/s "
              f"state={p.stateOperators[0].numRowsTotal if p.stateOperators else 0}")

spark.streams.addListener(Listener())

orders_stream batch=41 in=88412 rate=4103/s state=0
revenue_5m    batch=41 in=88412 rate=3902/s state=4102

The four numbers to alert on:

Metric	Warning sign
`inputRowsPerSecond` vs `processedRowsPerSecond`	input consistently higher — falling behind
`numRowsTotal` in `stateOperators`	growing without limit — missing watermark
`durationMs.addBatch`	rising over days — small files or state pressure
`numRowsDroppedByWatermark`	rising — the watermark is too tight for real arrival times

Practice

1. Run a streaming aggregation without a watermark and watch state grow.

batch=10  state=4102004
batch=20  state=8841200
batch=30  state=12488402

Monotonic growth with no ceiling. It runs fine for days, then fails — which is why this is worth reproducing deliberately once rather than discovering in production.

2. Delete a checkpoint and restart the stream.

"startOffset": null,
"endOffset": {"orders": {"0": 41254102}},
"numInputRows": 41254102

startOffset: null means it started from the beginning, and every row is reprocessed. On an append sink that is 41 million duplicates; with a MERGE sink it is merely slow — another argument for idempotent writes.

3. Use foreachBatch with a MERGE and feed it a duplicate key.

UnsupportedOperationException: Cannot perform Merge as multiple source rows matched
and attempted to modify the same target row

Add the row_number() deduplication and it succeeds. Any real event stream has duplicates — a producer retry is enough — so this is a certainty, not an edge case.

4. Compare availableNow with a continuous trigger.

# trigger(availableNow=True)
batch=0 in=88412 ... then the query terminates

# trigger(processingTime="30 seconds")
batch=0 in=88412
batch=1 in=2104
batch=2 in=1988   ... continues

availableNow gives streaming semantics — checkpoints, exactly-once — on a schedule, without paying for an idle cluster between runs. For anything that does not need sub-minute latency, it is the cheaper choice.

Next: cluster policies, SQL warehouses, and controlling what all of this costs.