Checkpointing in Structured Streaming

• Checkpoint location – Configured with option("checkpointLocation", path) on durable storage like S3 or HDFS; each streaming query needs its own distinct path.
• Offset tracking – The checkpoint stores the source offsets that have been read (write-ahead) and the offsets that have been fully committed, enabling exactly-once replay on failure.
• State persistence – For stateful queries, the contents of the state store (running counts, window states, join buffers) are versioned per batch and written to the checkpoint.
• Recovery – On restart, Spark reads the checkpoint, resumes from the last committed offsets, and restores state, so no data is skipped or double-counted.
• Incompatible changes – Changing the aggregation keys, state schema, or number of stateful operators can make an existing checkpoint unusable, requiring a fresh checkpoint path.

What is checkpointing in Structured Streaming?

Checkpointing is the mechanism that persists a streaming query's progress and state to reliable storage so the query can recover exactly where it left off after a failure or restart. You configure it with the checkpointLocation option, and the checkpoint directory stores offsets, commits, state, and query metadata.

Dataset<Row> counts = transactions
    .withWatermark("eventTime", "10 minutes")
    .groupBy("accountId")
    .count();

counts.writeStream()
    .outputMode("update")
    .format("console")
    .option("checkpointLocation", "s3a://nbi-checkpoints/account-counts")
    .start()
    .awaitTermination();

Why must the checkpoint be on durable storage?

The checkpoint must survive the loss of any single executor or the driver, so it must be on reliable, distributed storage such as HDFS or S3, not local disk. Every streaming query needs its own distinct path; sharing one path between two queries corrupts both.

// CORRECT -- separate paths on durable storage
countQuery.writeStream()
    .option("checkpointLocation", "s3a://nbi-checkpoints/count-query")
    .start();

sumQuery.writeStream()
    .option("checkpointLocation", "s3a://nbi-checkpoints/sum-query")
    .start();

// WRONG -- sharing a checkpoint path
countQuery.writeStream()
    .option("checkpointLocation", "s3a://nbi-checkpoints/shared")
    .start();
sumQuery.writeStream()
    .option("checkpointLocation", "s3a://nbi-checkpoints/shared") // CORRUPT!
    .start();

How does offset tracking give exactly-once?

Before processing a batch, Spark writes the offset range to the offsets log (write-ahead). After the batch is fully processed and written to the sink, Spark writes to the commits log. On restart, Spark checks the last offsets and commits entries: if a batch was started but not committed, it reprocesses that batch, and because the sink is idempotent or transactional, reprocessing does not duplicate output.

Checkpoint directory structure:
  s3a://nbi-checkpoints/account-counts/
    ├── offsets/          # Write-ahead offset ranges per batch
    │   ├── 0             # Batch 0: offsets [0, 1042]
    │   ├── 1             # Batch 1: offsets [1043, 2150]
    │   └── 2             # Batch 2: offsets [2151, 3200]
    ├── commits/          # Fully committed batch markers
    │   ├── 0
    │   └── 1             # Batch 2 not committed -- will be replayed
    ├── state/            # State store snapshots
    │   └── 0/            # Partition 0
    │       ├── 1.delta
    │       └── 2.delta
    └── metadata          # Query ID, schema, plan

How does state recovery work?

For stateful queries, the state store (running counts, window aggregates, join buffers) is versioned per batch and written into the checkpoint. On restart Spark restores the state as of the last committed batch, so running counts continue from where they were rather than resetting to zero.

// Before crash:
//   Batch 5 committed: accountId=ACC-001 count=47
//   Batch 6 started but NOT committed

// After restart:
//   State restored to Batch 5: ACC-001 count=47
//   Batch 6 replayed from offsets [5201, 6100]
//   ACC-001 count correctly becomes 52 (not reset to 0)

What changes are compatible with an existing checkpoint?

Some changes are safe across restarts, while others break the checkpoint.

| Change Type | Compatible? | Notes |
|---|---|---|
| Adjust trigger interval | Yes | Does not affect state or plan |
| Change non-stateful filter value | Yes | Stateless operators are flexible |
| Add a new non-stateful column | Yes | Does not affect state schema |
| Change aggregation keys | No | State is keyed differently |
| Add or remove stateful operators | No | State schema mismatch |
| Change output mode (with state change) | No | Plan incompatibility |
| Change watermark delay | Usually yes | Does not change state schema |

// Safe change: adjust trigger from 10s to 30s
counts.writeStream()
    .trigger(Trigger.ProcessingTime("30 seconds")) // was 10 seconds
    .option("checkpointLocation", "s3a://nbi-checkpoints/account-counts")
    .start(); // Resumes from existing checkpoint

// Unsafe change: add branchCode to groupBy
transactions.groupBy("accountId", "branchCode").count() // NEW key
    .writeStream()
    .option("checkpointLocation", "s3a://nbi-checkpoints/account-counts")
    .start(); // FAILS -- state schema mismatch

What happens with incompatible changes?

When you make an incompatible change, you must start from a fresh checkpoint path. This means the state resets and you either reprocess from the beginning or accept a gap.

// Breaking change: new aggregation key
// Step 1: Stop old query
// Step 2: Start new query with a NEW checkpoint path
transactions.groupBy("accountId", "branchCode").count()
    .writeStream()
    .option("checkpointLocation",
        "s3a://nbi-checkpoints/account-branch-counts-v2") // NEW path
    .option("startingOffsets", "earliest") // Reprocess if needed
    .start();

What about checkpoint write latency?

For large state, the per-batch checkpoint write can dominate batch duration. The RocksDB state store supports incremental checkpointing, uploading only changed state files, which keeps checkpoint cost proportional to the delta rather than the total state.

// Configure RocksDB for incremental checkpoints
spark.conf().set(
    "spark.sql.streaming.stateStore.providerClass",
    "org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider"
);
// Now checkpoint writes are incremental -- only changed state files
// Large state (100M keys) checkpoints in seconds instead of minutes

What are common mistakes?

The most common mistake is changing stateful query logic and restarting from the same checkpoint, expecting it to work. Another mistake is using local disk for the checkpoint path, which loses the checkpoint when the driver node is replaced.

// WRONG -- local disk checkpoint (lost on node failure)
counts.writeStream()
    .option("checkpointLocation", "/tmp/checkpoints/counts") // LOCAL!
    .start();

// CORRECT -- durable storage
counts.writeStream()
    .option("checkpointLocation", "s3a://nbi-checkpoints/counts")
    .start();

What should a senior engineer know?

Teams should version their checkpoint paths and plan migrations deliberately. For a breaking change, run the new query against a new checkpoint while backfilling, rather than pointing it at the old one and hoping. Monitor checkpoint write latency as a metric, because for large state it can dominate batch duration. Treat the checkpoint as durable, query-specific, and sensitive to logic changes.