Why does my backfill reprocess everything?

Summary

• A streaming query knows what it already processed only through its checkpoint. Delete the checkpoint or point the query at a new location and, in Databricks's words, "the next run of the query begins fresh".
• Start options such as Kafka startingOffsets and Delta startingVersion apply only when a new query starts. Once a checkpoint records progress, they are ignored.
• A pipeline full refresh clears checkpoints and recomputes from the source, so a backfill that "reprocesses everything" is often a full refresh.
• To make a replay safe, write idempotently, for example with a keyed MERGE.

Why does my backfill reprocess everything?

A streaming query knows what it already processed only through its checkpoint. The Spark docs say the checkpoint saves the range of offsets processed in each trigger and the running aggregates, so a query can "continue where it left off". When the checkpoint is gone, or the query points at a new one, nothing records that progress. Databricks puts it plainly: "When you delete the files in a checkpoint directory or change to a new checkpoint location, the next run of the query begins fresh."

What makes a stream start over?

Symptom	Likely cause	Fix
A restart reads the whole source	The checkpoint was deleted, or the query uses a new checkpointLocation	Reuse the original location, and give each query its own
startingOffsets or startingVersion seem ignored	They apply only when a new query starts, and a recorded checkpoint wins	Set them for the first run, or use a new checkpoint for an intentional replay
A new stream on a Delta table reads the whole table	The query first processes the table's initial snapshot	Set startingVersion or startingTimestamp when you start the query
A new file-source stream reads every existing file	maxFileAge does not limit the first batch: "all files will be considered valid"	Filter on a timestamp column, or keep old files out of the input path
Old rows reappear after an upstream UPDATE or MERGE	ignoreChanges re-processes rewritten files, and "unchanged rows may still be emitted"	Use skipChangeCommits, or make consumers handle duplicates
Aggregates or deduplication restart from zero	Their state lives in the checkpoint	Keep the checkpoint; stateful operations cannot change between restarts

How do I backfill only a range?

For Kafka, the Spark docs say a batch query can cover "a defined range of offsets":

df = (spark.read.format("kafka")
    .option("kafka.bootstrap.servers", "host1:port1,host2:port2")
    .option("subscribe", "topic1")
    .option("startingOffsets", """{"topic1":{"0":23,"1":-2}}""")
    .option("endingOffsets", """{"topic1":{"0":50,"1":-1}}""")
    .load())

In the JSON, -2 means earliest and -1 means latest. For a Delta source, set startingVersion or startingTimestamp when you start the new query. The Delta docs add that the source schema is always the table's latest schema, so check for incompatible changes after the version you pick.

How do I make a replay safe?

A replay writes rows again, so make the write idempotent. By default foreachBatch "provides only at-least-once write guarantees", and the Delta docs say the merge inside it must be idempotent because a restart can apply the same batch more than once. This keyed MERGE from the Delta docs updates existing keys and inserts new ones:

MERGE INTO target_table t
USING source_table s
ON t.key = s.key
WHEN MATCHED THEN UPDATE SET *
WHEN NOT MATCHED THEN INSERT *

A matched key takes the source's value, so replaying old rows can overwrite newer ones. An insert-only merge, which omits the WHEN MATCHED clause, adds only keys that are not already present. If you use the txnAppId and txnVersion writer options, a new checkpoint needs a different txnAppId, otherwise "writes from the restarted query will be ignored".

What does a Spark Declarative Pipelines full refresh do?

Spark Declarative Pipelines (SDP) arrived in Spark 4.1.0. In the Spark 4.2 docs, spark-pipelines run performs a default incremental update. --full-refresh DATASETS "clears all existing data and checkpoints for the specified datasets and recomputes them from scratch", --full-refresh-all resets the whole graph, and --refresh DATASETS updates "without clearing existing data". For sinks, "full refresh updates reset checkpoints but do not clean previously computed results", so an external sink can receive the replayed rows again.

spark-pipelines run --full-refresh orders,customers

How does Databricks approach backfills?

The Databricks docs say Lakeflow Pipelines "extend and are interoperable with Spark Declarative Pipelines". A default refresh of a streaming table processes new records. A full refresh clears the table's data and flow checkpoints and reprocesses all records from the source, and Databricks recommends it only when necessary, warning that it "can result in dropped records if input data is no longer available". To block it on a table, set the table property pipelines.reset.allowed to false.
For a backfill, add an append flow with once=True. It must return a batch DataFrame, runs one time, and runs again after a full refresh. The docs' example also gives the incremental flow a cutoff with the Auto Loader modifiedAfter option so it does not ingest the whole path. To reprocess selected flows without clearing ingested data, start an update through the REST API with reset_checkpoint_selection.

FAQs

Can I rewind a running stream by changing startingVersion?

No. The Delta docs say these options are ignored once progress is recorded in the checkpoint, and the Kafka docs say "resuming will always pick up from where the query left off".

Which query changes need a new checkpoint?

Per the Spark docs, changing the number or type of input sources is not allowed between restarts, changes to subscribed topics or files are generally not allowed, and stateful operations cannot change. Rate limits such as maxOffsetsPerTrigger can change.