ETLData engineering

How to make an ETL pipeline safe to rerun

Vishnujan Narayanan July 2026

A rerun should repair data, not corrupt it. Four ways that rule breaks, and a six-point checklist.

Rerunning a pipeline should fill a gap, correct a bad row, or change nothing. Never duplicate, never delete, never move the finish line backwards. That is the whole property, and it is easy to state and easy to break in ways that leave the data looking completely normal.

The standard patterns for getting it are well established: give every record a deterministic identifier derived from its content, upsert rather than append, or overwrite whole partitions instead of adding to them, and keep a watermark so the next run knows where to resume. I would use all of those, and I want to talk about what happens underneath them, because I broke this property four times while nominally following that advice.

Watermarks: data-derived or stored

The second run is the hard one. It has to decide what it already has, and there are two places to get that from: a separate store of progress, or the data itself.

Reading it from the data — find the newest row, ask for everything after that — is the better default for anything small enough not to need an orchestrator. There is no progress store to drift out of step with reality, because reality is the file. An interrupted run needs no cleanup; the next one reads what landed and carries on. Airflow's model of a run per data interval is the same idea with the bookkeeping made explicit, and if you are already running a scheduler you should use its version rather than inferring one.

The cost of inferring it is a single sentence that explains three of the four failures below: every row you write is an instruction to the next run. Write a bad row and you have not only stored bad data, you have told the next run where to start. So the checkpoint has to be computed from validated rows, and "valid" has to be defined in code rather than assumed.

Empty rows as the resume point

Plenty of sources return a record all-or-nothing. Ask for a period that has not settled, or get quietly throttled, and you get a row with a date on it and no values in it. Write that like any other row and then take the newest date as your resume point, and the empty row is the newest date. The pipeline concludes it is up to date and never asks for the real periods behind it again.

# Before: any row can become the resume point, including one
# that carries a date and no values at all.
start = pd.read_csv(path)["Date"].max()

# After: only rows that actually carry a value count.
df = pd.read_csv(path)
start = df[df["Close"].notna()]["Date"].max()

Two policies follow from that one condition. A response where every row is empty is a failed fetch rather than data, so it never reaches the file at all. And a pass at startup deletes the empty rows earlier runs wrote, because otherwise the damage from before you knew about this sits there permanently. Rows dated in the future need identical treatment — one of those convinces the pipeline it has data through 2027 and it stops fetching entirely.

This is the argument for a schema check at the boundary rather than trusting the source. pandera will declare what a valid row looks like and reject the load; I hand-rolled the equivalent and would not do that again.

Storage paths derived from a network call

I named per-item files using a label looked up at runtime. That lookup goes over the network and is rate-limited, so it returns the full name when the source is healthy, the bare identifier when it is not, and something else again when the label changes upstream.

Three answers meant three filenames for one item, and since the pipeline works out what it has by reading the file, a new filename looked exactly like an item with no history — so it downloaded the entire archive again into it. I ended up with items spread across three files, each holding a different partial copy of the same series.

Paths are now resolved from the stable identifier only, by looking for an existing file for that identifier and reusing whatever it is already called. The general form: anything you derive from a network call will eventually come back different, and if that value decides where data is stored, one bad minute at the source splits your dataset permanently.

Deduplication by merge, and row inflation

Reruns deliberately re-read a few periods, because records arrive late and a run that starts exactly where the last one stopped steps over them. So the pipeline constantly sees rows it already holds and needs to recognise them.

I implemented that as a merge — joining new rows against stored rows to see which matched. That works until the stored data itself contains duplicate keys, at which point each new row matches several stored rows and you get one output row per match. The check meant to remove duplicates was multiplying them, and it did that quietly.

def deduplicate(new_df, existing_df):
    new_keys = _composite_key(new_df, DEDUP_KEYS)
    existing_keys = set(_composite_key(existing_df, DEDUP_KEYS))
    return new_df[(~new_keys.isin(existing_keys)).values]

An anti-join cannot inflate anything because it never joins; it asks yes or no about each incoming row. If you would rather keep the merge, pandas takes validate="one_to_one", which raises on exactly this instead of performing it. I did not know that argument existed for an embarrassingly long time.

Two things underneath matter as much as the anti-join. The key has to be built from columns that exist — name three that do not and every row gets the same empty key, which collapses whole days of records into one row with no error anywhere. And the same record has to produce the same key every time, which it will not by default: the same quantity arrives as 629221 and 629221.0, the same date as 15-Jan-2026 and 2026-01-15. Canonicalise each field before building the key.

All of which is a hand-rolled version of what a database does with a unique constraint and INSERT ... ON CONFLICT DO NOTHING. If your target can be a database rather than a file, most of this section stops being your problem, and the reason my pipelines write CSVs is history rather than judgment.

Adaptive backoff as a data-quality control

All of the above assumes the source keeps answering properly. Hit a rate limit and you often do not get a clean refusal — you get a partial or empty response, which is how bad rows get into the file in the first place. Backing off is a data quality control here, not politeness.

A fixed delay is the wrong instrument: too slow on a good day, too aggressive on a bad one. What I run instead starts with five requests in flight, drops one whenever the source returns an error, and adds one back after fifteen consecutive successes, so it finds the source's tolerance and re-finds it when that changes. Individual requests get three attempts with a widening wait and a hard ten-second timeout. Whether a given refusal is even worth retrying is which refusals are worth retrying at all that I got wrong for a while.

Atomic writes: temp file and rename

This one is not from my own failures, and I include it because it is the thing I would have got wrong next. Writing a file in place is not a single operation. A crash partway leaves a truncated file, and the next run reads that as the complete dataset and carries on from it.

The pattern is to write a temporary file in the same directory and rename it over the original, since rename is the one filesystem operation that either happens or does not. The LWN article on ext4 and data loss is the clearest explanation of why the obvious version is unsafe and where fsync belongs. In a warehouse the equivalent is writing to a staging table and swapping, or overwriting a whole partition rather than appending to it — the partition-overwrite pattern is the cleanest form of idempotency available and it is worth reaching for whenever the data is naturally partitioned by time.

Source-side corrections: the unhandled case

Everything above makes reruns recover from my mistakes. None of it helps when the source changes a value it already gave me.

If a record is corrected upstream, the duplicate check sees a key it already holds, concludes it has that record, and throws the correction away. Converging on a stable dataset and accepting corrections pull against each other, and using set membership picks one without ever deciding to. For prices it mostly works out because adjustments get recalculated anyway. For filings and anything compliance-adjacent it is a real hole.

The proper answer is to stop overwriting history and start versioning it: key on identity, and let a changed value become a new row with its own validity window. That is a slowly changing dimension, and dbt snapshots implement it if you are in a warehouse. I read about that after the fact, and it is what I would build if this dataset mattered to anyone but me.

The gap I still have not closed is telling "the source had nothing new" apart from "the source refused to answer", since both produce zero rows. The fix for that and for the corrections problem is probably the same thing — re-fetch data I already have on a schedule and diff it against what I stored — which is the next thing I want to build and which I expect will be unpleasant reading.

References and further reading

Jepsen tests databases by deliberately breaking them and then checking whether the data survived. The reports are worth reading even if you never touch distributed systems, because they are the clearest demonstration of the pattern behind this whole post: a system reporting success while quietly losing or duplicating writes.

If you are building this on a scheduler rather than by hand, Airflow's best practices state the same rule as a requirement: every task must be idempotent and re-runnable for its data interval. Its run and backfill model is the vocabulary for the windowing described above.

Doing each piece properly

Postgres INSERT ... ON CONFLICT — an upsert against a unique key is the database's version of the anti-join here, and it cannot inflate rows the way a careless merge can.
dbt incremental models — the same "only process what is new, and be safe to run twice" problem, with an explicit unique key and a full-refresh escape hatch.
dbt snapshots — what to do when the source rewrites its own history, which is the case my resume logic still cannot see.
Stripe idempotency keys — the clearest short explanation of idempotency from the API side: how a client retries safely when it does not know whether the first attempt landed.
pandera — declare what a valid row looks like and reject a load that does not match. The tidy version of the hand-written validity checks above; I am not using it yet, and probably should be.
Ext4 and data loss and Files are hard — required reading if your checkpoint is a file, which mine is.
tenacity — retries and backoff you do not have to write, for the "going too fast" section.

Common data ingestion bugs — a year of real pipeline failures sorted by cause, which is where these rules came from.
How to scrape a site that paginates by date — a walkthrough of pulling two decades out of an endpoint that only answers a few months at a time.
How to test a data pipeline — how to test resume behaviour: stop a run part-way and check that the next one repairs it.
What a 429 really means — how to tell a rate limit that clears by itself from one that never will, and what each means for a retry.