Learn Apache Spark - Checkpointing & Fault Tolerance

#Introduction

Production streaming applications run for days or even months without stopping. During that time, many things can happen: a node dies, the network drops, Kafka becomes unresponsive. Episode 16 covers how Spark stays reliable through all of it — via checkpointing and fault tolerance.

Without a recovery mechanism, a single small failure can force a streaming job to restart from scratch, losing already-processed data or producing duplicates. Understanding how Spark stores progress and state is the key to building streaming pipelines you can trust.

This episode covers four things: checkpointing in Structured Streaming, fault tolerance and recovery behavior, state management for streaming jobs, and best practices for long-running streaming applications.

#Checkpointing in Structured Streaming

#What Is Stored in a Checkpoint

A checkpoint is the location where Spark periodically stores processing metadata. Its contents include:

• Offset log: the last position of data read from each source — this is what prevents data loss on restart.
• State metadata: the definition of the running query and its configuration.
• State data: for stateful operators, state snapshots are also stored.

Setting the checkpoint location

query = df.writeStream \
    .format("parquet") \
    .option("path", "data/hasil") \
    .option("checkpointLocation", "data/checkpoint") \
    .outputMode("append") \
    .start()

checkpointLocation must be stored on durable storage (HDFS, S3, or a persistent Kubernetes volume) — not on a worker's local disk that can be lost. This metadata is how Spark knows where to resume from.

#Placing Checkpoints Correctly

An important rule: a checkpoint location must not be shared by two different queries, and it must not be reused by a query whose structure has changed drastically. If the logic changes significantly, use a new checkpoint location.

#Fault Tolerance and Recovery

#From Failure to Recovery

When a streaming application is force-stopped or a node dies, the recovery flow goes like this:

1. A new driver is started again (by the cluster manager or Kubernetes).
2. Spark reads metadata from the checkpoint location.
3. The last offset becomes the starting point for re-reading data from the source.
4. The last state is restored, and processing continues from there.

restart driver → read checkpoint → resume from last offset → reprocess

#Exactly-Once vs At-Least-Once

Delivery semantics depend on the sink:

• Kafka: with the Kafka sink and checkpoints, Spark provides exactly-once — data is neither lost nor duplicated.
• File sink (Parquet): provides at-least-once; if the job fails mid-write, a batch can be rewritten.

spark-submit --conf spark.sql.streaming.schemaInference=true \
  --conf spark.sql.streaming.fileSink.log.deletion=true job.py

spark.sql.streaming.fileSink.log.deletion is one of the file sink metadata cleanup configurations. For stricter duplicate guarantees in batch processing, you can use Delta Lake, which provides idempotent transactions.

#State Management for Streaming Jobs

#State from Windows and Aggregations

Every stateful aggregation (window, groupBy) stores intermediate results in memory. Watermarks help clean up state that's no longer relevant. If state grows without control, the application can slow down or run out of memory.

#The State API: mapGroupsWithState

For custom stateful logic — such as session tracking — Spark provides the mapGroupsWithState and flatMapGroupsWithState APIs:

flatMapGroupsWithState in Scala

import org.apache.spark.sql.streaming.GroupState
 
def updateState(userId: String,
                events: Iterator[Event],
                state: GroupState[UserState]): Iterator[Output] = {
  val current = state.getOption.getOrElse(UserState.empty)
  val updated = current.update(events)
  state.update(updated)
  Iterator(Output(userId, updated.summary))
}

GroupState[UserState] stores per-key state that is persisted periodically to the checkpoint. You must define a timeout so the state of inactive keys can be removed — otherwise state will only grow.

#Cleaning Up State with a Timeout

Setting a state timeout

state.setTimeoutDuration("30 minutes")

state.setTimeoutDuration("30 minutes") automatically triggers the timeout callback for keys that haven't received events for 30 minutes — the standard pattern for sessionization.

#Best Practices for Long-Running Streaming

#Durable Design

Several practices keep streaming applications healthy for months:

• Checkpoint on persistent storage: not local disk, which can be lost along with the pod.
• Monitor lag and state size: alert when Kafka lag grows or state crosses a threshold.
• Clean restarts: make sure shutdown uses query.stop() and the driver is given time to finish writing the last batch.
• Don't change the query structure carelessly once a checkpoint exists.

#Handling Restarts and Upgrades

When upgrading code or libraries:

1. submit a new query with a new checkpoint (test mode)
2. verify no duplicate data and the sink is normal
3. cutover: stop the old query, run the new one

This pattern avoids a failure in the middle of the night caused by changes that were never tested against the old checkpoint format.

#Conclusion

Episode 16 rounds out the reliability aspect: checkpoints store offsets, state, and metadata so jobs can continue after failure; recovery restores position and state automatically; state management with mapGroupsWithState controls custom logic; and best practices keep streaming applications healthy over the long term.

Key takeaways:

• The checkpoint location must be on persistent, consistent storage.
• The offset log in the checkpoint prevents data loss and duplication.
• Exactly-once semantics depend on the sink — Kafka supports it, file sinks are at-least-once.
• Custom state needs an explicit timeout so it doesn't grow.
• Upgrading query structure should use a new checkpoint.

In the next episode, episode 17, we'll discuss custom extensions and UDFs — writing UDFs, UDAFs, and UDTs in Scala, Java, and Python, custom serialization with Kryo, extending Spark with external libraries, and packaging libraries for reuse across many jobs.