Moving data reliably
By the end: You can build a pipeline that survives being run twice.
- Batch and streaming
- Idempotency and backfills
- Schema change
- Data quality checks
New to data pipelines? A pipeline is code that runs on a schedule to move data from where it is produced to where it gets used, reshaping it on the way: pull yesterday's records from a database, clean them, write them somewhere a model or a dashboard can read. Nobody starts it by hand. It runs nightly or hourly, whether or not anyone is watching. Why it matters
A pipeline is a job that moves data from where it is produced to where it is used, on a schedule, without anyone watching. The hard part is not moving it once but surviving everything that goes wrong afterwards: the run that fails halfway, the run that fires twice, the upstream column that quietly changes type. This section is about building one that can be re-run safely, which is the property that separates a pipeline from a script.
What went wrong on Monday What the pipeline needed
the job died after writing half the rows replace a partition, never append
a bug was found in last quarter's numbers a date parameter, not "yesterday"
upstream added a column a schema check that tolerates additions
upstream turned an integer into a string a schema check that fails on types
the export ran but produced 40 rows a volume check against recent history
an event arrived three days late a trailing reprocess windowPipelines fail partially#
The defining problem is that a pipeline stops halfway. Some data was written, some was not, and nothing recorded where it got to. Nothing in the target says which rows are from the failed run, because rows do not carry that. Every design decision follows from asking what happens on the second run.
Make reruns safe#
An idempotent pipeline produces the same result whether it runs once or five times. That is a property of how you write, not a claim you make in a docstring, so here it is in code. The target is a list of rows tagged with the day they belong to, and the difference between the two loaders is one line.
def load_append(table, day, rows):
return table + [(day, r) for r in rows]
def load_replace(table, day, rows):
kept = [row for row in table if row[0] != day] # drop the day first
return kept + [(day, r) for r in rows]
rows = ["a", "b", "c"]
t = []
t = load_append(t, "2026-03-01", rows)
t = load_append(t, "2026-03-01", rows) # the retry after a timeout
print(len(t)) # 6: the day is counted twice
t = []
t = load_replace(t, "2026-03-01", rows)
t = load_replace(t, "2026-03-01", rows) # the same retry
print(len(t)) # 3: identical to one runThat is only half of it. Delete-then-insert as two visible steps leaves a window in which the day exists but is empty, and a crash inside that window is worse than the duplicate it was meant to prevent: nothing errors, and a reader sees a real zero. Build the new state first, then publish it with one operation that either happens or does not.
def build(table, day, rows, fail_after=None):
"""Compute the whole new state. Publishing is the caller's single assignment."""
staged = [row for row in table if row[0] != day]
for i, r in enumerate(rows):
if fail_after is not None and i == fail_after:
raise RuntimeError("source timed out")
staged.append((day, r))
return staged
table = build([], "2026-03-01", ["a", "b", "c"])
try:
table = build(table, "2026-03-02", ["d", "e", "f"], fail_after=1)
except RuntimeError:
pass # the assignment never ran
print(len(table)) # 3: day two published nothing at all
print(sorted({d for d, _ in table})) # ['2026-03-01']Backfills#
At some point you will reprocess two years of history, usually because a transform was wrong. The loop over dates is the easy part. What actually bites is everything around it.
- The job must take a date. A pipeline that reads the clock cannot be backfilled at all, and rewriting it to take one is the first day of the work.
- Direction matters. Run forwards if a day's output depends on the day before it; run backwards from today if the recent months are the ones people are looking at.
- Throughput. Seven hundred days launched at once will saturate the source database and page whoever owns it. Cap the concurrency and run it in chunks.
- Cost. Seven hundred times the daily cost, paid in an afternoon rather than over two years. Work it out before you start, not from the invoice.
- The table is live while you rewrite it. For hours, readers see some days on the old logic and some on the new. Either say so, or backfill into a copy and swap at the end.
- It will be interrupted. Keep a ledger of finished days so the resume costs minutes rather than starting again.
import datetime as dt
def daterange(start, end):
day = start
while day <= end:
yield day
day += dt.timedelta(days=1)
def backfill(run, start, end, ledger, limit=None):
"""Resumable. Takes the same run(day) the scheduler calls every morning."""
done = set(ledger.read())
processed = 0
for day in daterange(start, end):
if day.isoformat() in done:
continue
run(day) # publishes the day, atomically
ledger.mark(day.isoformat()) # only now is it done
processed += 1
if limit is not None and processed >= limit:
break # one chunk per invocation
return processed
class MemoryLedger:
def __init__(self): self.done = []
def read(self): return list(self.done)
def mark(self, key): self.done.append(key)
seen, ledger = [], MemoryLedger()
backfill(seen.append, dt.date(2026, 3, 1), dt.date(2026, 3, 10), ledger, limit=4)
backfill(seen.append, dt.date(2026, 3, 1), dt.date(2026, 3, 10), ledger)
print(len(seen), len(set(seen))) # 10 10: nothing ran twiceSchemas change#
Three things can happen to an upstream schema and only one of them is safe. A column is added, which nothing downstream reads and nothing breaks. A column is removed or renamed, which breaks every consumer that reads it, loudly and immediately. Or a column keeps its name and changes type, which breaks nothing at load time and quietly corrupts everything computed from it. The third is the one worth writing code against.
EXPECTED = {"order_id": "int", "user_id": "int", "total": "float", "country": "str"}
def kind_of(value):
return {bool: "bool", int: "int", float: "float", str: "str"}.get(type(value), "other")
def diff_schema(expected, row):
observed = {k: kind_of(v) for k, v in row.items() if v is not None}
added = [k for k in observed if k not in expected]
missing = [k for k in expected if k not in row]
changed = [
(k, expected[k], observed[k])
for k in observed
if k in expected and observed[k] != expected[k]
and not (expected[k] == "float" and observed[k] == "int") # widening is safe
]
return added, missing, changed
row = {"order_id": 7, "user_id": 31, "total": "19.99", "country": "DE", "channel": "app"}
added, missing, changed = diff_schema(EXPECTED, row)
print(added) # ['channel'] log it, carry on
print(missing) # [] a removal would be here
print(changed) # [('total', 'float', 'str')] stop the runThe policy that follows is dull and works. Additions are allowed without ceremony. Type changes stop the run. A rename is not one change, it is three separated in time, and doing all three in one release is an outage.
week 0 producer writes both country and country_code nothing breaks
week 1 consumers move to country_code, one at a time country still written
week 3 producer stops writing country nothing reads it
schema version bumped, old version deletedQuality checks in the pipeline#
Two kinds, and most pipelines only have the first. Absolute checks say what must always be true of a batch. They catch broken data.
def check(df, expected_rows):
assert len(df) > 0, "empty batch"
assert df.id.is_unique, "duplicate ids"
assert df.total.notna().all(), "missing totals"
assert (df.total >= 0).all(), "negative totals"
# Volume check catches upstream failures that produce valid but tiny data
assert len(df) > 0.5 * expected_rows, f"only {len(df)} rows, expected {expected_rows}"Relative checks compare today against what the last few runs looked like. They catch data that is not broken, just wrong: a source that silently started filtering, a partner that changed a currency, a region that stopped reporting. Compare against the median of recent runs rather than the mean, or one bad day raises the bar for the whole week after it.
def check_relative(today, history, tolerance=0.5):
"""history: row counts from the last successful runs, most recent last."""
if len(history) < 7:
return # not enough to judge yet
typical = sorted(history)[len(history) // 2]
if today < typical * tolerance or today > typical / tolerance:
raise ValueError(f"{today} rows against a typical {typical}")
recent = [4000, 4200, 3900, 4100, 4050, 4300, 3950]
check_relative(4100, recent) # fine
try:
check_relative(40, recent) # the export that finished early
except ValueError as e:
print(e) # 40 rows against a typical 4050Fail the pipeline, do not write bad data
Writing suspect data and flagging it means downstream consumers use it before anyone reads the flag. Stopping is disruptive and recoverable; silent corruption spreads and is not. The check belongs between the transform and the publish, on the staged result, so a failure leaves yesterday's good data in place.
Batch or streaming#
Batch accumulates data and processes a period of it. Streaming processes each event as it arrives. The difference people quote is latency, but the difference that decides the design is what you can redo. A batch job's input for a period is a fixed set of files that still exist, so a rerun is exact. A streaming job's input is a position in a log, redoing means replaying from an offset, and everything the job already emitted has already been acted on downstream.
Batch Streaming
unit of work a period: an hour, a day one event
rerun reprocess the period rewind to an offset, replay
correct on retry free, if you replace only if every write is idempotent
late data lands in a later run needs a watermark and a policy
cost when idle nothing the cluster runs all night
debugging rerun it locally on files reproduce a moment in a stream
people to operate one a rotaSo the question is whether freshness changes a decision. Fraud, alerting and live personalisation, yes. Reports, training data and nearly all analytics, no. Ask whether anyone would act differently if this data were an hour old, and if the answer is no, run it hourly. A five-minute batch gets most of the benefit of streaming while keeping batch's recovery story, which is why micro-batch is where a lot of supposedly streaming systems end up.
Where people go wrong
Appending instead of replacing. The first rerun after a failure silently doubles a day of data, and the error surfaces weeks later as an inexplicable spike in a dashboard.
If this comes up in an interview
How do you make a data pipeline safe to rerun?
Idempotency: write to a partition keyed by the period being processed and replace it rather than appending, so a rerun produces the same result. Parameterise by date range so backfills use the same code path.
How do you handle an upstream schema change?
Validate incoming data against an expected schema and fail loudly rather than writing whatever arrived. Adding nullable columns is safe; removals and type changes need a versioned migration with both supported during the transition.
What quality checks would you put in a pipeline?
Row count against expectation, uniqueness of keys, null checks on required fields, range checks on numerics, and a volume check. The volume check is what catches an upstream failure that still produces structurally valid but nearly empty data.
Build this: Build a pipeline that processes a date range, make it idempotent, add quality checks, then run it twice on the same date and confirm the output is identical.
This is one section of Data pipelines, 18 hours across 4 sections. Read it in full to get the whole topic in order.