A checkpoint is what makes a query the same query after a restart. Delete it, or point at a new location, and the next run begins fresh.
| Contents | Purpose |
|---|---|
| Offsets | Source positions processed in each micro-batch |
| Commits | Which micro-batches reached the sink, enabling exactly-once semantics |
| State | Stateful operator metadata, state schema and state store contents |
| Metadata | The unique query ID; configuration is kept in the offset log |
(df.writeStream
.option("checkpointLocation", "/Volumes/main/sales/chk/orders")
.toTable("main.sales.orders"))Every query needs its own location, and multiple queries must never share one. Sinks such as display() and the memory sink generate a temporary location if it is omitted, and those give no fault tolerance or consistency guarantees. Figure 4.1 shows how the checkpoint fits with state and the watermark.
What exactly-once covers
The checkpoint records offsets and commits, and a micro-batch must contain the same data on re-execution. When a query restarts, the micro-batch already planned runs first, even if settings have changed; new settings apply from the first newly planned batch. An out-of-memory failure on an oversized batch therefore cannot be fixed by tightening a rate limit alone: scale up the compute, or, in rare cases, restart with a new checkpoint.
| Boundary | Guarantee | What to add |
|---|---|---|
| Delta source to Delta sink | Exactly-once | Nothing |
| Streaming tables in Lakeflow Pipelines | Exactly-once by default; offsets and output commit together | Nothing, for managed flows |
foreachBatch or custom external writes |
At-least-once | Idempotent writes |
| Kafka as a sink | At-least-once; a retried batch can duplicate messages | Deduplicate at the consumer, for example by event ID |
| Custom Python data source | Depends on the implementation’s offset handling | Deduplicate downstream if offsets are not tracked |
Changes allowed across a restart
| Change | Status | Note |
|---|---|---|
| Add or remove filters | Allowed | |
| Change rate limits or trigger interval | Allowed | Applies from the next planned batch |
| Projection with the same output schema | Allowed | Different schema only if the sink allows it |
| Number or type of input sources | Not allowed by default | Source evolution relaxes this |
| Kafka topics or Auto Loader paths | Generally needs a new checkpoint | Results are unpredictable |
| File sink output directory | Not allowed | |
| File sink to Kafka sink | Allowed | Kafka sees only new data |
| Kafka sink to file sink | Not allowed | |
| Grouping keys or aggregates | Not allowed | State schema must remain the same |
| Join type or equi-join columns | Not allowed | |
| Watermark threshold or aggregation columns in a pipeline | Requires a full refresh | Existing state is incompatible |
Source evolution, on Runtime 18.2 and above, gives sources stable names so that they can be reordered, added or removed. It needs a fresh checkpoint and cannot be turned off for that checkpoint afterwards.
spark.conf.set(
"spark.sql.streaming.queryEvolution.enableSourceEvolution",
"true")
orders_eu = (spark.readStream
.name("orders_eu")
.table("main.sales.orders_eu"))Once it is on, every source must be named, or the query raises UNNAMED_STREAMING_SOURCES_WITH_ENFORCEMENT. A removed name cannot be reused, and a renamed source starts again from the beginning.
Compute changes matter too. dropDuplicates() and dropDuplicatesWithinWatermark() can fail a state schema compatibility check when the access mode changes. Dedicated to no isolation, and standard to serverless, are allowed; avoid other combinations.
Common trap: Setting a lower
maxFilesPerTriggerwill shrink a failed micro-batch on restart - the planned micro-batch has to complete first with the same data, because micro-batches are idempotent. Admission controls apply only after that batch. Scale up, or as a last resort use a new checkpoint.