Streaming Checkpoints and foreachBatch for Exactly-Once Recovery

1.What a checkpoint stores

A streaming query recovers from failure because it records its progress. Checkpoints and write-ahead logs work together to give Structured Streaming its processing guarantees. The checkpoint holds everything that identifies the query, including its state and which records it has processed. That is also why deleting the checkpoint files, or pointing the query at a new location, makes the next run start from scratch.

Set checkpointLocation before you start the query, and give every query its own location. Queries must never share one. Some sinks, such as display() in notebooks and the memory sink, create a temporary checkpoint if you leave it out. Those temporary locations provide no fault-tolerance or consistency guarantees and might not be cleaned up properly, so Databricks recommends always setting one.

(df.writeStream
  .option("checkpointLocation", "/Volumes/catalog/schema/volume/path")
  .toTable("catalog.schema.table")
)

Sources1

2.Which query changes a checkpoint survives

Structured Streaming automatically checkpoints state data to fault-tolerant storage and restores it after a restart. This only works if the restarted query still matches what the checkpoint describes. Changes to filters, rate limits and trigger intervals are generally safe. Changes to the input sources, the subscribed Kafka topics or Auto Loader paths, the type of stateful operation, the state schema, or the sink type generally need a new checkpoint. Sources are identified by their position in the query plan, which is why adding or reordering them breaks recovery unless source naming is turned on.

Other settings are fixed when the checkpoint is created. The number of shuffle partitions is one: changing spark.sql.shuffle.partitions has no effect on a query that already has a checkpoint. Databricks Runtime 18 LTS and above adds on-demand repartitioning for stateful queries. The state management scheme cannot change between restarts either, so switching state store means starting from a new checkpoint. Queries that use dropDuplicates() or dropDuplicatesWithinWatermark() can also fail a state schema compatibility check when you change the compute access mode. Do not change the access mode for those queries.

Sources12

3.foreachBatch: arbitrary sinks with exactly-once you build yourself

writeStream.foreachBatch(...) runs batch code on the output of each micro-batch. Your function receives two arguments: the micro-batch's output DataFrame and the micro-batch's unique ID. It is how you write to sinks that have no streaming writer, and Delta Lake MERGE in Structured Streaming requires it. Delta sinks don't support update output mode, so running MERGE INTO inside foreachBatch is the standard way to write streaming aggregates to a Delta table with update semantics. foreachBatch relies on micro-batch execution, so it does not work in continuous processing mode.

Because a retried batch arrives with the same batchId, you can use it to make the writes idempotent. For Delta table writes, set the txnAppId and txnVersion write options and bind txnVersion to batchId. For custom destinations such as OLTP databases, queues or APIs, you have to build idempotency yourself: running a given batchId again must produce the same result. Your function also has to cope with an empty DataFrame. A Delta source can produce one, for example after an OPTIMIZE with no files to process or when file pruning removes every record.

def process_batch(output_df, batch_id):
  # Process valid DataFrames only
  if not output_df.isEmpty():
    # business logic
    pass

streamingDF.writeStream.foreachBatch(process_batch).start()

With a stateful operator upstream, such as dropDuplicatesWithinWatermark, each call must consume the entire DataFrame or restart the query. If it doesn't, the query fails on the next batch. A function that only calls batch_df.show(2) consumes just part of the batch, so it has to drain the rest explicitly:

def partial_func(batch_df, batch_id):
  batch_df.show(2)
  batch_df. ? (do_nothing) # silently consume the rest of the batch

For errors, Databricks recommends letting the query fail fast and leaving retries to the orchestration layer, such as Lakeflow Jobs or Airflow. Retry loops inside your function risk losing data. A try/except that swallows the error might also leave the batch partly consumed, so either re-raise the error or drain the rest of the batch. If you write to several sinks, prefer separate streaming writers: writing them all from one foreachBatch serializes the writes and adds latency.

Sources3

Exam traps

Each one states something that sounds right. Open it to see what is actually true.

1. 1.Because the query has a checkpoint, writes made in foreachBatch are exactly-once.Why is that wrong?

   foreachBatch guarantees only at-least-once writes. You get exactly-once only by using batchId to make the writes idempotent, for example with txnAppId/txnVersion on Delta.

   Covered in foreachBatch: arbitrary sinks with exactly-once you build yourself

2. 2.You can add a grouping key to a running streaming aggregation and restart it from the same checkpoint.Why is that wrong?

   The checkpointed state has a fixed schema. Adding, removing or changing a stateful operation is not allowed between restarts, so this change needs a new checkpoint.

   Covered in Which query changes a checkpoint survives

3. 3.Two streaming queries writing to the same table can share one checkpoint location.Why is that wrong?

   A checkpoint identifies a single query, through its query ID, offsets and commits. Every query needs its own location.

   Covered in What a checkpoint stores