Data Engineering · Episode 01 · 1:26
Idempotent pipelines: why one rerun doubled the revenue
A job crashed after loading Oct 7 and the naive rerun appended the same 8 orders again: $4,100 instead of $2,050. Deleting that day's rows and inserting fresh ones in one transaction makes every rerun land on $2,050.
What you’ll needPython 3.10+, DuckDB
What you’ll need
- python.org/downloads →
Python 3.10 or newerRuns the code. Check yours with python3 --version. - code.visualstudio.com →
VS Code or CursorEither works; so does any editor you like. - git-scm.com/downloads →
gitGets the code from GitHub. 
DuckDB 1.0+SQL on files and tables, no database serverpip install "duckdb>=1.0"
Set it up
- 1Get the code
git clone https://github.com/DayanEbrar0X/data-anatomy.ai.git cd data-anatomy.ai
- 2Make a virtual environmentIt keeps this project's packages separate from the rest of your computer.
python3 -m venv .venv source .venv/bin/activate
- 3Install the packagesrequirements.txt installs every lesson's packages. This one needs only: DuckDB.
pip install -r requirements.txt # or just this lesson: pip install "duckdb>=1.0"
- 4Run it
cd data-engineering/15-idempotent-pipelines python3 src/load.py
Project structure15-idempotent-pipelines/
Project structure
data-engineering/15-idempotent-pipelines/Open on GitHub →
- data/
- history.csv6 orders for Oct 6, already in the warehouse ($1,730)
- orders.csvtonight's 8 orders for Oct 7 ($2,050)
- short/
- job.pythe warehouse, the two-step job and the report
- rerun.pythe 9-line version from the short
- src/
- job.pythe warehouse, the two-step job and the report
- load.pymainthe code from the video
- README.mdthe idea, how to run it, a line-by-line walkthrough, things to try
data-engineering/15-idempotent-pipelines/src/load.pyOpen on GitHub →
from job import warehouse, run_job, report
def naive(db, day, rows):
db.executemany("INSERT INTO revenue "
"VALUES (?, ?, ?)", rows)
def idempotent(db, day, rows):
db.begin() # all or nothing
db.execute("DELETE FROM revenue "
"WHERE day = ?", [day])
naive(db, day, rows)
db.commit()
for load in (naive, idempotent):
db = warehouse() # already holds Oct 6
run_job(db, load, crash=True)
run_job(db, load) # the rerun
report(db, load.__name__)Transcript
Your job failed halfway. You reran it. Revenue just doubled. The fix has a name.
Idempotent: run it once or five times, same result. Here's the setup. The warehouse already holds Oct 6. Tonight's job loads eight orders for Oct 7, then publishes a report.
Step one finishes. Step two crashes. So the scheduler reruns everything. The naive load is one insert.
It appends whatever it gets. It never checks whether those rows are already there. So the rerun appends the same eight orders again. Sixteen rows.
The idempotent load treats the day as a partition. Open a transaction, so it's all or nothing. Delete every row for that day, then insert the day fresh, and commit. Rerun it, and the day is rewritten.
Never added to. Now the test. Each load gets a fresh warehouse, a crashed run, then the rerun. Naive: Oct 7 shows $4,100.
Sixteen rows. Idempotent: $2,050 in eight rows. The real number. And Oct 6?
$1,730, untouched both times. Schedulers retry. Backfills rerun old days. Reruns are normal.
So build every load to be safe to run again. Delete and insert by partition, or merge on a key. One gotcha: delete and insert must share one transaction, or a crash leaves the day empty. Fail halfway, rerun, same revenue.
That's idempotent.
Read the lesson on GitHub →- #dataengineering
- #duckdb
- #python
- #etl
