lesson 228 Reliable Data Pipelines · advanced

Make a load safe to run twice

Pipelines get rerun: a retry after a timeout, a scheduler that fires twice, someone pressing Run again. An idempotent step has the same effect whether it runs once or five times, so a rerun can never double a total.

You'll be able to

Explain idempotency and why reruns make it necessary
Load records keyed by their ID (a dictionary or a primary key with ON CONFLICT) instead of appending
Apply each change once by remembering the IDs already applied, and write outputs in full, sorted

The idea

An operation is idempotent when doing it twice has the same effect as doing it once. *Set order O100282's status to shipped* is idempotent; *add 1 to the shipped count* is not.

Pipelines need this because they are rerun all the time. A job times out after it wrote its data but before it said so, and the scheduler retries it. A message queue delivers the same event twice. A backfill is started again after a fix. If loading appends rows, every rerun adds them again and the star counts double.

The tools: give each record its natural key (an event ID, an order ID) and *upsert* by that key instead of appending; remember which keys or batches were already applied; and write output files in full each run (sorted, so the same input gives the same bytes) instead of appending to them.

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

example-1.py · Loading the same batch twice

batch = [
    {"id": "48000007504", "type": "WatchEvent", "repo": "lumen-labs/pyauth"},
    {"id": "48000004550", "type": "WatchEvent", "repo": "tidewater/easystore"},
    {"id": "48000006650", "type": "WatchEvent", "repo": "quill-data/opencache"},
]

appended = []
by_id = {}
for attempt in [1, 2]:          # the job is retried after a timeout
    for event in batch:
        appended.append(event)
        by_id[event["id"]] = event

print(len(appended), "rows after appending twice")
print(len(by_id), "rows after upserting twice")

A database does the same with a primary key and an upsert clause. INSERT ... ON CONFLICT(event_id) DO NOTHING skips a row whose key is already stored (DO UPDATE SET ... would replace it instead). The load function can then be run as often as the scheduler likes.

example-2.py · A table that ignores replays

import sqlite3

def load_stars(conn, events):
    rows = [(e["id"], e["repo"]) for e in events if e["type"] == "WatchEvent"]
    conn.executemany(
        "INSERT INTO stars (event_id, repo) VALUES (?, ?) ON CONFLICT(event_id) DO NOTHING",
        rows,
    )
    conn.commit()

def stars_per_repo(conn):
    return conn.execute("SELECT repo, COUNT(*) FROM stars GROUP BY repo ORDER BY repo").fetchall()

conn = sqlite3.connect(":memory:")
conn.execute("CREATE TABLE stars (event_id TEXT PRIMARY KEY, repo TEXT)")
hour = [
    {"id": "48000007504", "type": "WatchEvent", "repo": "lumen-labs/pyauth"},
    {"id": "48000000581", "type": "PushEvent", "repo": "tidewater/smartqueue"},
    {"id": "48000004550", "type": "WatchEvent", "repo": "tidewater/easystore"},
    {"id": "48000009120", "type": "WatchEvent", "repo": "lumen-labs/pyauth"},
]
load_stars(conn, hour)
print(stars_per_repo(conn))
load_stars(conn, hour)          # the scheduler ran the job again
print(stars_per_repo(conn))

Not every step has a natural place to store a key. Adding money to a balance, for example: running it twice pays twice. There, keep the IDs of the changes already applied and skip those. And for files: open outputs with "w" and write the whole result, sorted, so a rerun replaces the file with the same bytes. A file opened with "a" gains a second copy of every line.

When to use it: every step of a pipeline that might be retried, which is every step. Mistakes to avoid: += on a running total without remembering what was already added; append-mode outputs; and recording the key *before* the change is made, so a crash in between marks it as done when it wasn't.

Review

Idempotent: running a step again has no further effect. Pipelines are rerun, so every step should be.
Store by natural key (dictionary, primary key with ON CONFLICT) instead of appending.
Remember the IDs of changes already applied, adding each one as it is made; write output files in full and sorted.
Next: a long run that crashes halfway shouldn't start from zero. A checkpoint records how far it got.

Practise this