lesson 229 Reliable Data Pipelines · advanced

Save progress with a checkpoint

A backfill that crashes after two hours shouldn't start again from zero. A checkpoint is a small file saying how far the run got, written after the work it describes and replaced in one step, so it is never half written.

You'll be able to

Decide what a checkpoint must hold: the position to continue from and the partial results
Write it atomically: a temporary file, then os.replace
Save the checkpoint only after the work it describes is done, and refuse to trust a damaged one

The idea

Long jobs fail partway: a machine restarts, a deploy kills the process, one file is unreadable. A checkpoint is a small record of progress, saved as the job goes, so a restart can continue instead of repeating hours of work.

It holds two kinds of thing: where to continue (the files already done, or an offset into a list) and the partial results that can't be recomputed cheaply (running totals). Not the data itself: the input still exists and can be read again.

Two rules make it trustworthy. First, save it after the work it describes, never before; otherwise a crash in between skips work on restart. Second, write it atomically: write a temporary file, then os.replace(temporary, path). Replacing swaps the name in one step, so anyone reading the checkpoint sees either the old complete file or the new complete file, never half of one (you wrote atomic_write this way in the Decorators & Context Managers module).

Run each example to see what it prints. Running examples doesn't count toward progress.

example-1.py · Save and load a checkpoint

import json
import os

def save_checkpoint(path, state):
    temporary = path + ".tmp"
    with open(temporary, "w", encoding="utf-8") as f:
        json.dump(state, f)
    os.replace(temporary, path)

def load_checkpoint(path):
    if not os.path.exists(path):
        return {"done": [], "pushes": 0}
    with open(path, encoding="utf-8") as f:
        return json.load(f)

pushes_per_hour = {"2025-03-03-00.jsonl": 41, "2025-03-03-01.jsonl": 38, "2025-03-03-02.jsonl": 35}

state = load_checkpoint("backfill.json")
for name, pushes in pushes_per_hour.items():
    state["pushes"] += pushes
    state["done"].append(name)
    save_checkpoint("backfill.json", state)
    print("saved after", name)

print(load_checkpoint("backfill.json"))
print(os.path.exists("backfill.json.tmp"))

Why the detour through a temporary file? A process can be killed in the middle of a write. Writing straight into the checkpoint with open(path, "w") first empties the file, then fills it; stop in between and the only record of two hours' progress is a fragment:

example-2.py · A checkpoint cut off mid-write

import json

state = {"done": ["2025-03-03-00.jsonl", "2025-03-03-01.jsonl"], "pushes": 79}
text = json.dumps(state)
with open("backfill.json", "w", encoding="utf-8") as f:
    f.write(text[:30])          # the process was killed here, part way through the write

with open("backfill.json", encoding="utf-8") as f:
    print(json.load(f))

When the work is a long list rather than a set of files, the position is an offset: how many items are finished. Work in batches and save after each batch; saving after every single item is safe but slow, and saving rarely means more repeated work after a crash.

example-3.py · Batches with an offset

import json
import os

def save_checkpoint(path, state):
    temporary = path + ".tmp"
    with open(temporary, "w", encoding="utf-8") as f:
        json.dump(state, f)
    os.replace(temporary, path)

pushes = [
    {"id": "48000000581", "commits": 2},
    {"id": "48000003472", "commits": 2},
    {"id": "48000000105", "commits": 3},
    {"id": "48000003143", "commits": 3},
    {"id": "48000005159", "commits": 1},
]
BATCH = 2
state = {"offset": 0, "commits": 0}
for start in range(state["offset"], len(pushes), BATCH):
    batch = pushes[start:start + BATCH]
    for push in batch:
        state["commits"] += push["commits"]
    state["offset"] = start + len(batch)
    save_checkpoint("commits.json", state)
    print("checkpoint", state)

When to use it: any run long enough that starting over hurts: backfills, migrations, big exports. Mistakes to avoid: saving before the work is done; writing the checkpoint in place instead of replacing it; storing the whole dataset instead of a position; and treating an unreadable checkpoint as "no checkpoint".

Review

A checkpoint holds where to continue and the partial results, not the data.
Save it after the work it describes; write it to a temporary file and os.replace it.
No checkpoint means a fresh start; a damaged one must stop the run, not restart it.
Next: use the checkpoint to recover from a real crash and finish with the same result as a clean run.

Practise this

Test the checkpoint loader