lesson 230 Reliable Data Pipelines · advanced

Resume after a crash

Recovery is the payoff of checkpoints and idempotent writes: after a crash the run continues from its checkpoint, cleans up anything written after it, and ends with exactly the result a clean run would have produced.

You'll be able to

Resume a loop from a saved offset or a list of finished units, and clear the checkpoint when the run completes
Handle output written after the last checkpoint: make it idempotent, or cut it back to the checkpoint
Test recovery the honest way: crash a run for real, resume it, and compare with a clean run

The idea

A checkpoint is only useful if the restart uses it correctly. Recovery means three things. Continue from the saved position instead of the beginning. Deal with work done *after* the last checkpoint, because a crash can land between writing output and saving the checkpoint. And when the run finishes, remove the checkpoint, or tomorrow's run will think today's is still in progress.

The test of a recovery design is simple to state: crash it, resume it, and the result must equal one clean run. The examples do exactly that, with a real exception standing in for the crash.

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

example-1.py · Crash, then resume

import json
import os

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

def load(path):
    if not os.path.exists(path):
        return {"offset": 0, "counts": {}}
    with open(path, encoding="utf-8") as f:
        return json.load(f)

def count_types(events, path):
    state = load(path)
    for position in range(state["offset"], len(events)):
        kind = events[position]["type"]
        state["counts"][kind] = state["counts"].get(kind, 0) + 1
        state["offset"] = position + 1
        save(path, state)
    os.remove(path)
    return state["counts"]

events = [
    {"id": "48000000581", "type": "PushEvent"},
    {"id": "48000001631", "type": "CreateEvent"},
    {"id": "48000007504", "type": "WatchEvent"},
    {"id": "48000002275"},
    {"id": "48000006895", "type": "PushEvent"},
]
try:
    count_types(events, "types.ckpt")
except KeyError as err:
    print("run 1 crashed on", err, "with checkpoint", load("types.ckpt"))

events[3]["type"] = "IssuesEvent"      # the damaged event is repaired upstream
print("run 2:", count_types(events, "types.ckpt"))
print("clean:", count_types(events, "fresh.ckpt"))
print(os.path.exists("types.ckpt"))

Counting in memory is the easy case: everything after the checkpoint was lost with the process. Output written to a database or a file is harder, because it survives the crash. If the process writes a batch and dies before saving the checkpoint, the restart will write that batch again.

There are two ways out. The first is the one from L03: make the write idempotent, so writing a batch twice is harmless (*at-least-once* delivery plus idempotent writes gives *exactly-once* results). The second, for plain files, is to cut the output back to what the checkpoint describes before continuing (the challenges use that one).

example-2.py · Replay is harmless when writes are idempotent

import sqlite3

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE events (event_id TEXT PRIMARY KEY, type TEXT)")

def load_from(conn, events, offset):
    for event in events[offset:]:
        conn.execute(
            "INSERT INTO events VALUES (?, ?) ON CONFLICT(event_id) DO NOTHING",
            (event["id"], event["type"]),
        )
    conn.commit()

events = [
    {"id": "48000000581", "type": "PushEvent"},
    {"id": "48000001631", "type": "CreateEvent"},
    {"id": "48000007504", "type": "WatchEvent"},
    {"id": "48000002275", "type": "IssuesEvent"},
    {"id": "48000006895", "type": "IssueCommentEvent"},
]

# What the disk holds after the crash: three rows were committed,
# but the process died before the checkpoint moved past the second.
load_from(conn, events[:3], 0)
checkpoint_offset = 2

load_from(conn, events, checkpoint_offset)     # the restart replays event 3
print(conn.execute("SELECT COUNT(*) FROM events").fetchone()[0])
print(conn.execute("SELECT COUNT(DISTINCT event_id) FROM events").fetchone()[0])

When to use it: whenever a job keeps a checkpoint; recovery is the part that has to be right. Mistakes to avoid: starting the loop at 0 instead of the saved offset; appending to an output file without first cutting it back to the checkpoint; never deleting the checkpoint, so the next run skips everything; and testing recovery only by reading the code. Crash it on purpose and compare with a clean run.

Review

Recovery: continue from the checkpoint, deal with output written after it, delete the checkpoint when done.
Output that outlives a crash is replayed on restart: make the write idempotent, or cut the output back to the checkpoint.
Test it for real: crash a run, resume it, and compare with a clean run.
Next module: pipelines read outside data, and outside data can attack. The Secure Python module makes Python code secure against it.

Practise this