Operating Structured Streaming Checkpoints on Databricks: Restartability, Exactly-once, and Failure Recovery

Structured Streaming checkpoints are the foundation that persists a streaming query's progress, enabling restart after failures and Exactly-once processing. Get the design wrong and you will hit duplicates, data loss, or OOMs from runaway state. This article covers checkpoint internals, design rules, failure recovery procedures, and the conditions for Exactly-once from a practical operations standpoint.

Inside the Checkpoint Directory

The checkpoint directory contains four subdirectories. Each one manages a different aspect of the streaming query's state.

/mnt/checkpoints/orders-pipeline/
├── offsets/
│   ├── 0    # Batch 0: {"kafka-topic": {"0": 100, "1": 150}}
│   ├── 1    # Batch 1: {"kafka-topic": {"0": 200, "1": 300}}
│   └── 2
├── commits/
│   ├── 0    # Batch 0: sink write committed
│   └── 1    # Batch 1: sink write committed
├── sources/
│   └── 0/
└── state/
    └── 0/
        └── delta/  # State store snapshots

Recovery on restart works as follows. If a batch has an entry in offsets/ but no matching entry in commits/, it is treated as 'read but not committed' and gets reprocessed. Batches with a commit entry are skipped, which prevents double writes.

Design Rule: One Checkpoint per Query

The single most important checkpoint rule is: assign one unique checkpoint path to one streaming query.

• Sharing a path across multiple queries causes offsets files to collide, leading to data loss and duplicates
• Even within the same query, reusing an old checkpoint after a logic change can fail due to state incompatibility
• Recommended naming follows a hierarchy like /mnt/checkpoints/{pipeline-name}/{query-name}/

# Give every query its own checkpointLocation
orders_query = (
    orders_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/mnt/checkpoints/orders-pipeline/orders-to-silver")
    .toTable("silver.orders")
)

payments_query = (
    payments_df.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/mnt/checkpoints/orders-pipeline/payments-to-silver")
    .toTable("silver.payments")
)

Choosing Where to Store Checkpoints

Use cloud storage or UC Volumes in production. Restrict DBFS checkpoints to development and validation.

Conditions for Achieving Exactly-once

Structured Streaming's Exactly-once semantics only hold when both source replayability and sink idempotency are in place.

When the sink is a Delta table, Delta's transaction log and the Structured Streaming checkpoint cooperate to deliver Exactly-once. When you write to an external system through foreachBatch, you need to manage idempotency keys using the batch ID.

def process_batch(batch_df, batch_id):
    # Example of idempotent processing using the batch ID
    batch_df.write \
        .format("delta") \
        .mode("append") \
        .saveAsTable("silver.events")

(source_df.writeStream
    .foreachBatch(process_batch)
    .option("checkpointLocation", "/mnt/checkpoints/events-pipeline/events-to-silver")
    .start()
)

Failure Recovery Procedures

Here are the recovery patterns for streaming jobs that have stopped due to a failure.

Pattern 1: Plain restart (the common case)

Restart the job with the same checkpointLocation. The query resumes from the batch after the last committed one. This is enough when nothing in the code has changed.

Pattern 2: Recovery with a logic change

• Adding or removing output columns: keep the checkpoint if mergeSchema covers the change
• Changes to stateful operations (different watermark window, new aggregation key): reset the checkpoint and rebuild the sink table as well
• Source changes (e.g. adding a topic): reset the checkpoint and start fresh

Pattern 3: Recovering from a corrupted checkpoint

# 1. Move the corrupted checkpoint aside
dbutils.fs.mv(
    "/mnt/checkpoints/orders-pipeline/orders-to-silver",
    "/mnt/checkpoints/_archived/orders-to-silver-corrupted-20260327"
)

# 2. Check the last timestamp written to the sink table
last_ts = spark.sql("""
    SELECT MAX(event_timestamp) FROM silver.orders
""").collect()[0][0]

# 3. Restart with a new checkpoint (use MERGE to prevent duplicates)
(source_df.writeStream
    .foreachBatch(lambda df, id: merge_deduplicate(df, id))
    .option("checkpointLocation", "/mnt/checkpoints/orders-pipeline/orders-to-silver")
    .start()
)

State Management and Watermarks

Windowed aggregations and stream-stream joins accumulate state in the checkpoint's state/ directory. Without a watermark, state grows without bound and eventually triggers an OOM.

from pyspark.sql.functions import window, col

(orders_df
    .withWatermark("event_time", "10 minutes")
    .groupBy(window("event_time", "5 minutes"), "product_id")
    .agg({"amount": "sum"})
    .writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "/mnt/checkpoints/orders-pipeline/windowed-agg")
    .toTable("gold.orders_5min_agg")
)

• A watermark is a declaration that 'no more data older than this will arrive'
• State past the watermark is purged automatically, keeping memory and checkpoint size under control
• If the window is too short, late data is dropped; if too long, state balloons — it is a trade-off

Operational Checklist

• Put checkpointLocation on cloud storage or UC Volumes (DBFS for development only)
• Standardize a naming convention with one checkpoint path per query
• Set a watermark on every stateful operation
• Monitor checkpoint directory size periodically to catch abnormal growth
• Document failure recovery procedures in a runbook, including how to handle the sink table when resetting a checkpoint
• Configure auto-recovery via the job retry policy (Databricks Jobs' Retries setting)

Check Your Understanding

Frequently Asked Questions