Checkpointing

Concepts covered: Structured Streaming

A streaming query is meant to run for weeks, and the processes running it will die along the way: a node gets reclaimed, a deploy restarts the driver, an executor runs out of memory. The checkpoint is how the next run learns where the last one stopped. It is a folder on durable storage, named by the checkpointLocation option on the writer, that the query updates on every micro-batch.

trips = (
    spark.readStream
    .schema(trip_schema)
    .json("/landing/trip_events/")
)

completed = (
    trips
    .filter(F.col("status") == "completed")
    .select("trip_id", "driver_id", "city_id", "fare")
)

query = (
    completed.writeStream
    .foreachBatch(write_payout_lines)
    .option("checkpointLocation", "/chk/payout_lines/")
    .trigger(processingTime="1 minute")
    .start()
)

Inside /chk/payout_lines/, each micro-batch leaves files named by its batch number. Here is what each entry means, using batch 41 as the example:

Entry	Written	Holds
metadata	Once, at the first start	The query's id, which ties every later run to this folder
offsets/41	Before batch 41 runs	Where batch 41 ends in each source, plus settings frozen for the batch
sources/0/41	While batch 41 is planned	The 3 new files batch 41 reads; the file source keeps this list
commits/41	After the sink accepts batch 41	A marker that batch 41 is finished
state/	Every batch, stateful queries only	Values carried between batches, such as running counts

An offset is a source's position marker. For a partitioned message log it is a record number per partition; for the file source it is an entry number in the file list it keeps under sources/. Batch 41 starts where offsets/40 ended and stops at offsets/41, so the pair pins down exactly which rows belong to the batch: here, 3 files holding 634 trip events.

One micro-batch in order. The driver records the batch's range in offsets/41 before any task starts, and writes commits/41 only after the sink's write returns.

trip_events files (source, file source)
Plan batch 41 (transform, driver)
offsets/41 (storage, checkpoint)
Run batch 41 (transform, executors)
payout_lines (consumer, sink)
commits/41 (storage, checkpoint)

trip_events files -> Plan batch 41
Plan batch 41 -> offsets/41
offsets/41 -> Run batch 41
Run batch 41 -> payout_lines
payout_lines -> commits/41

That order is the design. offsets/41 is a promise about what batch 41 will read, made before the work starts; commits/41 is the receipt, written after. A crash can land between them, and Spark's restart logic exists to handle that gap.

What a restart reads

When a query starts with a checkpointLocation that already has content, Spark finds the highest batch number in offsets/ and in commits/ and compares them before reading any new data.

Latest offsets	Latest commits	What Spark does first
41	41	Plans batch 42, starting where batch 41 ended
41	40	Runs batch 41 again over the range stored in offsets/41
none	none	Treats the query as new and asks each source for its starting position

The middle row decides correctness. Spark never plans a fresh range for a half-finished batch: it re-reads the same 3 files, rebuilds the same 634 rows and passes them to the sink again, under the same batch id. Whether the sink is harmed by that depends on the sink, which the next section takes up.

A stream with no checkpoint

Without a checkpoint, a restarted query has no record of its past and begins as a new query. What that costs depends on the source:

File source: lists every file in the landing folder and processes all of them, so every trip already paid shows up in the sink again.
Message log read from its latest position: skips every record produced while the query was down, so those trips are never paid.
Message log read from its earliest position: replays everything the log still retains, duplicating it downstream.

Any state the query had built is gone as well. Spark protects most sinks from this: the file sink and foreachBatch refuse to start without a checkpointLocation, while the console sink, meant for debugging, gets a temporary folder that is deleted when the query stops.

Do

Put the checkpoint on storage that outlives the cluster, such as an object store path
Give each query its own checkpoint folder and keep it for the life of the query
Reuse the same checkpointLocation when you redeploy the same query

Avoid

Point 2 queries at one checkpoint folder; they overwrite each other's offsets
Put it on the driver's local disk; a driver restarted elsewhere cannot see it
Delete the folder to unstick a failing stream; that resets its position too

The payout query dies after offsets/41 is written and after the sink has received part of batch 41, but before commits/41 exists. It restarts with the same checkpointLocation. What does Spark run first?

Batch 42, over only the files that landed after the crash
Batch 41 again, over the 3 files recorded in the checkpoint
Batch 41 again, over every file in the folder at restart time
Nothing, until an operator picks a new starting position

About this lesson

This section is part of the Structured Streaming: Intermediate lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.