How to configure Spark streaming checkpoints

Apache Spark Structured Streaming checkpoints store the progress a streaming query needs after a controlled stop, driver failure, or deployment restart. A checkpointed query can remember which input offsets or files were processed and, for stateful queries, the state that belongs to previous micro-batches.

In PySpark, the checkpoint is configured on the streaming writer with checkpointLocation before the query starts. The location must be a directory in an HDFS-compatible filesystem that the driver and executors can reach; a local /tmp path is only suitable for a single-machine check.

Use one checkpoint directory for one logical query and keep the input source, output sink, and output path compatible when the query restarts from that directory. Reusing a checkpoint across unrelated queries or deleting it before a restart makes Spark treat the run as a new stream and can reprocess input.

1. Choose a checkpoint path that belongs only to this streaming query.

   Local check root: /tmp/sg-spark-streaming-checkpoint
   Input path: /tmp/sg-spark-streaming-checkpoint/input
   Output path: /tmp/sg-spark-streaming-checkpoint/output
   Checkpoint path: /tmp/sg-spark-streaming-checkpoint/checkpoint

   For a cluster job, use durable shared storage such as HDFS or another Spark-supported filesystem path that every driver restart can read.

2. Save the checkpoint recovery check as sg_streaming_checkpoint.py.

   sg_streaming_checkpoint.py

   import json
   import shutil
   from pathlib import Path
    
   from pyspark.sql import SparkSession
    
    
   base = Path("/tmp/sg-spark-streaming-checkpoint")
   input_dir = base / "input"
   output_dir = base / "output"
   checkpoint_dir = base / "checkpoint"
    
   shutil.rmtree(base, ignore_errors=True)
   input_dir.mkdir(parents=True)
    
    
   def write_events(path, rows):
       with path.open("w", encoding="utf-8") as handle:
           for row in rows:
               handle.write(json.dumps(row, sort_keys=True) + "\n")
    
    
   write_events(
       input_dir / "events-001.json",
       [
           {"event": "checkout", "region": "MY", "count": 3},
           {"event": "search", "region": "SG", "count": 2},
       ],
   )
    
   spark = (
       SparkSession.builder.master("local[2]")
       .appName("sg-streaming-checkpoint")
       .config("spark.sql.shuffle.partitions", "2")
       .config("spark.ui.showConsoleProgress", "false")
       .getOrCreate()
   )
   spark.sparkContext.setLogLevel("ERROR")
    
   schema = "event STRING, region STRING, count INT"
    
    
   def run_query(label):
       events = spark.readStream.schema(schema).json(str(input_dir))
       query = (
           events.select("event", "region", "count")
           .writeStream.format("json")
           .queryName("sg_streaming_checkpoint")
           .option("path", str(output_dir))
           .option("checkpointLocation", str(checkpoint_dir))
           .trigger(availableNow=True)
           .start()
       )
       query.awaitTermination(60)
       progress = query.lastProgress or {}
       print(f"{label}_query_id={query.id}")
       print(f"{label}_run_id={query.runId}")
       print(f"{label}_input_rows={progress.get('numInputRows')}")
       return str(query.id), str(query.runId)
    
    
   first_query_id, first_run_id = run_query("first")
    
   write_events(
       input_dir / "events-002.json",
       [
           {"event": "checkout", "region": "MY", "count": 4},
       ],
   )
    
   second_query_id, second_run_id = run_query("second")
    
   checkpoint_entries = sorted(
       path.name for path in checkpoint_dir.iterdir() if not path.name.startswith(".")
   )
   print("checkpoint_entries=" + ",".join(checkpoint_entries))
   print(f"query_id_reused={first_query_id == second_query_id}")
   print(f"run_id_changed={first_run_id != second_run_id}")
    
   rows = spark.read.json(str(output_dir)).orderBy("event", "count").collect()
   print(f"output_rows={len(rows)}")
   for row in rows:
       print(f"{row['event']},{row['region']},{row['count']}")
    
   spark.stop()

   The availableNow trigger lets the local check process the files that are already present and terminate. A long-running production query can use the default trigger or a processing-time trigger while keeping the same checkpointLocation option.

3. Run the streaming job with spark-submit.

   $ spark-submit --master 'local[2]' --conf spark.ui.showConsoleProgress=false sg_streaming_checkpoint.py
   ##### snipped #####
   first_query_id=5176356a-8a73-44ad-9f79-e059fc4f129e
   first_run_id=c238b02a-490a-4ed5-aa48-85fa393bad01
   first_input_rows=2
   second_query_id=5176356a-8a73-44ad-9f79-e059fc4f129e
   second_run_id=05d83cd6-9664-47fc-8abe-5ff3ff77c5c0
   second_input_rows=1
   checkpoint_entries=commits,metadata,offsets,sources
   query_id_reused=True
   run_id_changed=True
   output_rows=3
   checkout,MY,3
   checkout,MY,4
   search,SG,2

   query_id_reused=True shows Spark recovered the same logical query from the checkpoint. run_id_changed=True shows the second start was a new execution attempt. second_input_rows=1 shows only the new input file was processed after restart.

   Related: How to submit an Apache Spark job

4. Keep the checkpoint directory with the streaming query while the query must be restartable.

   Required checkpoint entries:
   commits
   metadata
   offsets
   sources

   Do not reuse this directory for a different source, sink, output path, or aggregation shape. Spark documents several query changes as unsupported or undefined when restarting from the same checkpoint.

5. Remove the local proof files after the checkpoint behavior is confirmed.

   $ rm -r /tmp/sg-spark-streaming-checkpoint sg_streaming_checkpoint.py

   Remove only local test paths. Keep production checkpoint directories for streams that still need restart recovery.