Spark Structured Streaming checkpoint recovery runbook: restart, change or reset a stream safely

#When to use this runbook

• A streaming job fails on restart right after a deploy, with errors about sources, state schema or offsets.
• Someone asks "can we just delete the checkpoint?"
• You need to move a stream to a new checkpoint location, a new cluster or a new workspace.

The checkpoint is where the stream's progress lives. Deleting it is the most common cause of reprocessed data and duplicate rows in streaming pipelines. Work through the steps in order.

#Step 1: Stop the stream and confirm only one is running

Two queries on the same checkpoint corrupt each other's progress (with a Delta sink you will see ConcurrentTransactionException).

for q in spark.streams.active:
    print(q.id, q.runId, q.name, q.status)

In Databricks Jobs, set Maximum concurrent runs to 1 for every streaming job. Cancel the run before you touch the checkpoint:

databricks jobs list-runs --job-id 123456 --active-only
databricks jobs cancel-run 987654321

#Step 2: Look inside the checkpoint

databricks fs ls dbfs:/Volumes/main/ops/checkpoints/orders_bronze/
# or on ADLS directly
az storage fs file list --account-name mylake --file-system checkpoints \
  --path orders_bronze --auth-mode login --output table

Read the latest offset file to see where the stream is:

databricks fs cat dbfs:/Volumes/main/ops/checkpoints/orders_bronze/offsets/4182

For a Delta source the JSON includes reservoirVersion (the source table version). For Kafka it is a map of topic to partition offsets, e.g. {"orders":{"0":1849302,"1":1850117}}. Write these down. You need them if you start a new checkpoint.

If the highest offsets/ number is one ahead of the highest commits/ number, the last batch was planned but not committed. On restart Spark reruns it. That is normal.

#Step 3: Decide whether the change is checkpoint-compatible

Match the failed restart to the rules from the Spark guide ("Recovery semantics after changes in a streaming query"):

If the change is in the "Yes" rows and the restart still fails, roll back the deploy and read the full stack trace. It is likely a source problem (Step 6), not a checkpoint problem.

If the change is in a "No" row, you need a new checkpoint. Go to Step 4.

#Step 4: Start a new checkpoint at the right position

Caution

Never delete the old checkpoint. Point the query at a new path and keep the old one until the new stream has run cleanly for a few days. It is your only record of what was processed.

Starting options such as startingVersion and startingOffsets are only read when a query starts with an empty checkpoint. On an existing checkpoint they are ignored, which surprises people who change them and redeploy.

#Delta source

Start from the version after the one recorded in Step 2:

(spark.readStream
    .option("startingVersion", 52871)        # last reservoirVersion + 1
    .table("main.raw.orders")
  .writeStream
    .option("checkpointLocation", "/Volumes/main/ops/checkpoints/orders_bronze_v2")
    .toTable("main.bronze.orders"))

startingTimestamp works too, but a version is exact.

#Kafka source

starting = '{"orders":{"0":1849302,"1":1850117}}'

(spark.readStream.format("kafka")
    .option("kafka.bootstrap.servers", brokers)
    .option("subscribe", "orders")
    .option("startingOffsets", starting)
  ...)

Check that those offsets still exist on the brokers. If retention has removed them, the stream fails unless failOnDataLoss is false, which silently skips the missing data.

#Auto Loader

The file-tracking state lives in the checkpoint. A new checkpoint re-ingests every file in the path unless you set cloudFiles.includeExistingFiles to false, which skips files already present when the stream starts. Files that arrived between the old stream stopping and the new one starting are then skipped too. Backfill them deliberately.

#Make the overlap harmless

Starting a new checkpoint usually reprocesses the last batch or two. Make the sink idempotent so this doesn't create duplicates:

def upsert(batch_df, batch_id):
    batch_df.createOrReplaceTempView("updates")
    batch_df.sparkSession.sql("""
        MERGE INTO main.bronze.orders t
        USING updates s ON s.order_id = t.order_id
        WHEN MATCHED THEN UPDATE SET *
        WHEN NOT MATCHED THEN INSERT *
    """)

(stream.writeStream
    .foreachBatch(upsert)
    .option("checkpointLocation", "/Volumes/main/ops/checkpoints/orders_bronze_v2")
    .start())

For plain Delta appends inside foreachBatch, the txnAppId and txnVersion writer options let Delta skip a batch it has already committed.

#Step 5: Stateful queries need a rebuild plan

A new checkpoint means empty state. Windowed aggregations and dropDuplicates start from zero, so:

• Aggregates for open windows will be incomplete. Start the new stream far enough back (earlier startingVersion) to cover the longest window or watermark, and MERGE into the sink so recomputed windows overwrite old rows.
• Dedup state is lost. Rows already seen can come through again. Dedup again at the sink with a MERGE on the key.

#Step 6: Source-side failures that look like checkpoint problems

Exact message text varies by Spark and Databricks Runtime version; check your build's output.

#Step 7: Validate after restart

q = spark.streams.get("<query id>")
print(q.lastProgress["batchId"], q.lastProgress["numInputRows"], q.lastProgress["sources"])

Then compare source and sink counts for the restart window and check the sink for duplicate keys:

SELECT order_id, count(*) AS c
FROM main.bronze.orders
WHERE ingest_date >= '2026-10-22'
GROUP BY order_id HAVING count(*) > 1;

#Prevent it next time

• One checkpoint path per query, versioned in the path (orders_bronze_v1, _v2), stored on a Unity Catalog volume or external location, not on the driver's disk.
• Review every streaming code change against the table in Step 3 before merging.
• Make every sink idempotent (MERGE or txnAppId/txnVersion) so a reset costs time, not data quality.
• Keep source retention (Delta log and VACUUM, Kafka topic retention) longer than your worst-case outage.
• Alert when a streaming job has been stopped longer than a set period, not only when it fails.