Checkpointing: a consistent cut

How does a job take a consistent snapshot while records flow, and what do aligned and unaligned barriers cost?

To recover without replaying from the beginning, you need a consistent cut: a snapshot in which state reflects exactly some prefix of the input, with the source offsets recorded inside the same snapshot. Nothing is counted twice, nothing is skipped.

Small example. The source has read up to offset 1,000. The count operator has counted records 0–999 exactly. If the snapshot stores "count = 1,000" and "offset = 1,000", then after a restore, reading resumes at 1,000 and the count is right. If the snapshot stored "count = 1,000" but "offset = 950", records 950–999 would be counted twice. If it stored "offset = 1,050", records 1,000–1,049 would be skipped.

The hard part is getting such a cut while records keep flowing between operators on different machines. The method is Chandy–Lamport snapshots (a classical algorithm for taking a consistent snapshot of a distributed system without stopping it), applied to a dataflow graph:

1. The engine injects a marker (a barrier) into the stream at the sources, at the point that matches the offsets it records.
2. Barriers flow with the records through the graph.
3. Each operator snapshots its state when the barrier reaches it.
4. Records in flight between operators are accounted for deterministically (see below).

Aligned barriers

An operator with multiple inputs waits until the barrier has arrived on all of them. While it waits, it buffers the records from channels whose barrier arrived early (those channels "ran ahead"), and does not process them. Then it snapshots.

Which records are in the snapshot? All records before the barrier on every channel have been processed by this operator and are reflected in its state. Records after the barrier on early channels are held in the buffer and not yet processed, so they are not reflected, and they will be processed after the snapshot. So there is no in-flight data to store in the snapshot. The state and the source offsets already describe one exact prefix.

The cost. The alignment duration equals the skew between channels (how much later the slowest barrier arrives). And under backpressure the barrier travels at the speed of the data, so checkpoints slow down precisely when the job is struggling. That is a death spiral:

• load slows checkpoints;
• slow checkpoints mean the recovery point does not advance;
• the eventual recovery is more expensive than it would have been.

This was Ana's 9-minute checkpoints.

Unaligned barriers

The operator snapshots immediately when the first barrier arrives, and captures the in-flight buffer contents into the checkpoint as well: the records that are still in channels ahead of and behind the barrier. Checkpoint duration is decoupled from load. You pay in checkpoint size and write bandwidth.

This is the conserved-cost law exactly: the cost was relocated from latency into storage and network, not removed.