A daily pipeline you can rerun safely

A scheduled job that loads yesterday's data will fail one morning, and someone will run it again. This section is about writing it so the rerun is harmless: partition the output by date, compute into a temporary location, check the result, and replace a whole day at once instead of appending. Get that property and every failure has the same fix, which is to run the day again, and backfilling a bug fix across three months is the same code in a loop rather than a special path.

The shape#

extract day D from source  ->  transform  ->  check  ->  replace partition D in the target
                                                     |
                                                     fail: publish nothing, alert

In code#

import duckdb, datetime as dt, pathlib, shutil

def run(day: dt.date):
    src = f"raw/calls/date={day}/*.json"
    out = pathlib.Path(f"warehouse/daily_usage/date={day}")
    tmp = out.with_name(out.name + ".tmp")

    con = duckdb.connect()
    df = con.sql(f"""
        SELECT user_id, model, sum(input_tokens) AS input_tokens,
               sum(output_tokens) AS output_tokens, count(*) AS calls
        FROM read_json_auto('{src}')
        WHERE status = 'ok'
        GROUP BY ALL
    """).df()

    check(df, day)                                  # raises on bad data
    tmp.mkdir(parents=True, exist_ok=True)
    df.to_parquet(tmp / "part.parquet")
    if out.exists():
        shutil.rmtree(out)
    tmp.rename(out)                                 # publish: replace the whole day

def check(df, day):
    assert len(df) > 0, f"{day}: no rows"
    assert (df["input_tokens"] >= 0).all(), f"{day}: negative tokens"
    assert df["user_id"].notna().all(), f"{day}: missing user ids"

Because each run replaces exactly one day's partition, running it twice gives the same result as running it once. That property, idempotency, is what makes failures boring: rerun the day. Three details in that code are doing the work and are worth naming, because leaving any one of them out gives you a pipeline that looks idempotent and is not.

• The check runs on the staged result, before anything is published. Check after publishing and you have already corrupted the day you are about to complain about.
• The write goes to a temporary directory with a different name, so a half-written file is never at the path readers use.
• The last step is a rename, which the filesystem performs as one operation. Writing directly into the output directory and deleting the extra files afterwards is two operations, and a crash between them leaves a day that is part old and part new.

That third point has a limit worth knowing. The rmtree then rename above is itself two operations, so a crash in the gap leaves the day missing entirely rather than mixed. Missing is recoverable and mixed is not, which is why it is written in that order, but if you need the swap to be genuinely atomic then write to a new versioned directory and move a pointer, which is what table formats such as Iceberg and Delta do for you.

Proving it, rather than believing it#

Idempotency is a claim about behaviour under failure, so the test has to inject a failure. This runs without a warehouse: a dict stands in for the target, and the load is killed at a chosen row.

def load(target, day, rows, die_at=None):
    """Stage the whole day, then publish with one assignment."""
    staged = []
    for i, row in enumerate(rows):
        if die_at is not None and i == die_at:
            raise RuntimeError("killed mid-run")
        staged.append(row)
    target[day] = staged                       # the publish
    return target

target, rows = {}, [1, 2, 3, 4]

for die_at in (None, 0, 2, 3):                 # clean run, then three crash points
    try:
        load(target, "2026-03-01", rows, die_at=die_at)
    except RuntimeError:
        pass

print(target)                                  # {'2026-03-01': [1, 2, 3, 4]}
print(sum(target["2026-03-01"]))               # 10, after four runs and three crashes

When the target is a table, not files#

Same property, different mechanism. Delete the day's rows and insert the new ones inside one transaction, so no reader ever sees the gap and a failure rolls the whole thing back.

BEGIN;
DELETE FROM daily_usage WHERE usage_date = :day;
INSERT INTO daily_usage
SELECT :day, user_id, model, sum(input_tokens), sum(output_tokens), count(*)
FROM raw_calls
WHERE call_date = :day AND status = 'ok'
GROUP BY user_id, model;
COMMIT;

The alternative is a merge on a natural key, which upserts row by row rather than replacing a period. It is the right shape when rows update individually, for instance a customer record that changes. It is the wrong shape for a daily aggregate, because a row that disappears from the source is never deleted by a merge: it simply keeps its old value forever, and nobody notices until someone reconciles the totals.

Data that keeps arriving#

Most sources are not finished with a day when the day ends. Payments settle, a partner exports late, a mobile client syncs when it next has signal. If the job only ever runs yesterday, those records land in a partition that was already published and are never counted.

def daily(today: dt.date, lookback: int = 7):
    """Reprocess a trailing window, not one day. Safe only because run() replaces."""
    for offset in range(1, lookback + 1):
        run(today - dt.timedelta(days=offset))

This is the concrete payoff of idempotency. Without it, reprocessing a week every night would multiply a week of data by seven; with it, the trailing window is free of consequences and the only question is compute. It is also why the cost of the property is worth paying up front, before you know you need the window.

Backfills#

def backfill(start: dt.date, end: dt.date):
    day = start
    while day <= end:
        run(day)
        day += dt.timedelta(days=1)

That loop is correct and will not survive contact with two years of history. Add three things before you start one: a ledger of days already finished so an interruption resumes, a concurrency cap so the source database is not saturated, and a number for what it will cost, obtained by timing one day and multiplying. The last is the one people skip, and the invoice arrives a month later attached to nobody's name.

Scheduling#

An orchestrator such as Airflow, Dagster or Prefect runs run(yesterday) every morning, retries on failure, runs backfills in parallel, and shows which days succeeded. Cron works for one job; an orchestrator earns its keep when jobs depend on each other.

Two settings decide whether it wakes you at three in the morning. Retries should be bounded and spaced, because the usual cause of a failure is an upstream system that is down and will not be up in ten seconds. And a job needs an alert on not having run at all, not only on having failed: a scheduler that stopped firing produces no failures, no red squares, and a table that quietly stops growing.