Chapter 5 · Pipelines and AI
ETL and ELT: Building a Reliable Data Pipeline
- Page 17 of 22
- 20 min read
Every dashboard, every trained model and every RAG index is fed by a data pipeline: code that regularly pulls data from where it is created, reshapes it, and puts it where it is used. When a pipeline is unreliable, nobody notices at first. Then the revenue chart shows a dip that never happened, or a model is retrained on half a day of duplicated rows. A good pipeline is boring: it can be run again after any failure, it never double-counts, and it tells you loudly when something is wrong.
This page builds one from the pieces of this chapter. It extracts from the shop database, from a courier "API" (a local JSON file stands in for it) and from a CSV of targets, transforms and validates the rows, loads them into a separate warehouse database, and is idempotent: run it twice, and the second run changes nothing.
What you will learn
- What a pipeline is, ETL vs ELT, and full vs incremental loads with a watermark (and why a lookback window is needed).
- Idempotent loading with upserts, and proving it by running twice.
- Validation and reconciliation (row counts, sums), logging, and retries with backoff for flaky sources.
- Scheduling (cron, an Airflow DAG), monitoring and alerts, and testing a pipeline.
- CDC: what change data capture is and when you need it.
ETL and ELT
Both move data from sources to a target (a warehouse, a lake, a feature store). They differ in where the transformation happens:
ETL: sources ──extract──► [ transform in Python / Spark ] ──load──► warehouse (clean tables only)
ELT: sources ──extract──► warehouse (raw tables) ──transform with SQL inside the warehouse──► clean tables| ETL | ELT | |
|---|---|---|
| Transform runs in | the pipeline code (Python, Spark) | the warehouse, as SQL (often managed with dbt) |
| Raw data kept? | usually not | yes, so you can re-transform when the logic changes |
| Good for | cleaning that SQL is bad at (parsing, APIs, ML scoring), sensitive data that must be masked before it lands | large volumes in cloud warehouses (BigQuery, Snowflake) where SQL compute is cheap and scales |
Real pipelines mix them, and so will ours: Python joins the API data to the order lines (ETL), then SQL inside the warehouse builds the monthly summary (ELT).
The sources
The shop database is the first source. The other two are written here so the page is self-contained: the courier API's response, saved as JSON, and a monthly targets file from the sales team:
import json
deliveries = [
{"order_id": 1, "courier": "Pathao", "delivered_on": "2026-01-14"},
{"order_id": 2, "courier": "RedX", "delivered_on": "2026-01-23"},
{"order_id": 3, "courier": "Pathao", "delivered_on": "2026-02-05"},
{"order_id": 4, "courier": "Steadfast", "delivered_on": "2026-02-16"},
{"order_id": 6, "courier": "RedX", "delivered_on": "2026-03-11"},
{"order_id": 7, "courier": "Pathao", "delivered_on": "2026-03-17"},
{"order_id": 8, "courier": "Steadfast", "delivered_on": "2026-03-30"},
{"order_id": 9, "courier": "RedX", "delivered_on": "2026-04-06"},
{"order_id": 10, "courier": "Pathao", "delivered_on": "2026-04-21"},
{"order_id": 11, "courier": "Steadfast", "delivered_on": "2026-05-08"},
{"order_id": 12, "courier": "RedX", "delivered_on": None},
{"order_id": 14, "courier": "Pathao", "delivered_on": "2026-06-12"},
]
with open("courier_api.json", "w", encoding="utf-8") as f:
json.dump({"deliveries": deliveries}, f, indent=1)
with open("targets.csv", "w", encoding="utf-8") as f:
f.write("month,target\n2026-01,8000\n2026-02,8000\n2026-03,20000\n"
"2026-04,20000\n2026-05,15000\n2026-06,20000\n")
print("sources ready")sources readyBuilding the pipeline
The target: a warehouse with state
The target is a separate SQLite file, warehouse.db. Besides the data tables it has two bookkeeping tables: etl_state remembers how far the last run got (the watermark), and etl_runs records every run for monitoring. The primary key of the fact table, one row per order line, is what makes idempotent loading possible:
import csv
import logging
import sqlite3
import sys
import time
from datetime import date, timedelta
log = logging.getLogger("etl")
WAREHOUSE_DDL = """
CREATE TABLE IF NOT EXISTS fact_order_lines (
order_id INTEGER NOT NULL,
product_id INTEGER NOT NULL,
order_date TEXT NOT NULL,
month TEXT NOT NULL,
category TEXT NOT NULL,
status TEXT NOT NULL,
quantity INTEGER NOT NULL,
revenue INTEGER NOT NULL,
courier TEXT,
delivery_days INTEGER,
PRIMARY KEY (order_id, product_id)
);
CREATE TABLE IF NOT EXISTS monthly_sales (
month TEXT PRIMARY KEY, orders INTEGER NOT NULL, revenue INTEGER NOT NULL,
target INTEGER, pct_of_target REAL
);
CREATE TABLE IF NOT EXISTS etl_state (key TEXT PRIMARY KEY, value TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS etl_runs (
run_id INTEGER PRIMARY KEY, mode TEXT, since TEXT, extracted INTEGER,
inserted INTEGER, updated INTEGER, status TEXT
);
"""Extract
Three extractors, one per source. The SQL extractor pushes the joins and the filter down to the source database and returns plain dictionaries. The API extractor is wrapped in retries, because networks fail; the stand-in class fails on purpose for its first calls, the way a real API sometimes times out:
def extract_order_lines(src, since):
"""Order lines of orders placed on or after `since` (YYYY-MM-DD)."""
cur = src.execute("""
SELECT oi.order_id, oi.product_id, o.order_date, o.status,
COALESCE(c.name, 'Uncategorised') AS category,
oi.quantity, oi.quantity * oi.unit_price AS revenue
FROM order_items oi
JOIN orders o ON o.order_id = oi.order_id
JOIN products p ON p.product_id = oi.product_id
LEFT JOIN categories c ON c.category_id = p.category_id
WHERE o.order_date >= ?
ORDER BY oi.order_id, oi.product_id""", (since,))
cols = [d[0] for d in cur.description]
return [dict(zip(cols, row)) for row in cur]
class CourierAPI:
"""Stand-in for an HTTP API. Real code: requests.get(url, timeout=10).json()."""
def __init__(self, path, fail_first=0):
self.path, self.fail_first = path, fail_first
def get_deliveries(self):
if self.fail_first > 0:
self.fail_first -= 1
raise ConnectionError("courier API timed out")
with open(self.path, encoding="utf-8") as f:
return json.load(f)["deliveries"]
def with_retries(fn, attempts=4, base_delay=0.01):
"""Call fn(); on a transient error wait 1x, 2x, 4x ... base_delay and try again."""
for attempt in range(1, attempts + 1):
try:
return fn()
except ConnectionError as exc: # transient: worth retrying
if attempt == attempts:
raise
delay = base_delay * 2 ** (attempt - 1)
log.warning("attempt %d failed (%s), retrying in %.2fs", attempt, exc, delay)
time.sleep(delay)
def extract_targets(path):
with open(path, newline="", encoding="utf-8") as f:
return [{"month": r["month"], "target": int(r["target"])} for r in csv.DictReader(f)]Only ConnectionError is retried. A bad API key or malformed data fails the same way every time; retrying it only delays the alert. Two notes for real code. With requests, catch requests.exceptions.ConnectionError and requests.exceptions.Timeout: they are not subclasses of Python's built-in ConnectionError. And add jitter, a random part of the delay (for example delay * random.uniform(0.5, 1.5)), so that a hundred workers that failed together do not all retry in the same instant and knock the API over again. Libraries such as tenacity package all of this. (The delays here are tiny and fixed so the page's output stays the same on every run.)
Transform and validate
The transform adds the month and joins each line to its delivery (how many days it took, and with which courier). Validation then checks the rows against rules before anything is written. If a rule fails, the run stops and the warehouse is untouched:
def transform(lines, deliveries):
by_order = {d["order_id"]: d for d in deliveries}
rows = []
for line in lines:
d = by_order.get(line["order_id"], {})
days = None
if d.get("delivered_on"):
days = (date.fromisoformat(d["delivered_on"]) - date.fromisoformat(line["order_date"])).days
rows.append({**line, "month": line["order_date"][:7],
"courier": d.get("courier"), "delivery_days": days})
return rows
def validate(rows):
problems = []
for r in rows:
if r["quantity"] <= 0 or r["revenue"] <= 0:
problems.append(f"order {r['order_id']}: non-positive quantity or revenue")
if r["delivery_days"] is not None and not 0 <= r["delivery_days"] <= 30:
problems.append(f"order {r['order_id']}: delivery took {r['delivery_days']} days")
if problems:
raise ValueError("validation failed: " + "; ".join(problems))Load: idempotent upserts
The load is where idempotency is won or lost. A plain INSERT run twice either fails on the primary key or, without a key, duplicates everything. An upsert inserts new lines and updates existing ones (the ON CONFLICT … DO UPDATE syntax is in INSERT, UPDATE, DELETE and Upserts, Safely). An insert that skips rows already there, like the NOT EXISTS load in Importing and Exporting: CSV, JSON, Excel and Parquet, is idempotent too, but it never picks up a change to a row it loaded before; here orders change status, so we need the update. The WHERE … IS NOT … clause updates a row only if one of the columns that can change (status, quantity, revenue, courier, delivery days) is different, so an unchanged row is not even rewritten, and counting changes tells you exactly what the run did. (IS NOT rather than <>, so a NULL courier compares correctly.)
UPSERT_FACT = """
INSERT INTO fact_order_lines (order_id, product_id, order_date, month, category, status,
quantity, revenue, courier, delivery_days)
VALUES (:order_id, :product_id, :order_date, :month, :category, :status,
:quantity, :revenue, :courier, :delivery_days)
ON CONFLICT (order_id, product_id) DO UPDATE SET
order_date = excluded.order_date, month = excluded.month, category = excluded.category,
status = excluded.status, quantity = excluded.quantity, revenue = excluded.revenue,
courier = excluded.courier, delivery_days = excluded.delivery_days
WHERE (fact_order_lines.status, fact_order_lines.quantity, fact_order_lines.revenue,
fact_order_lines.courier, fact_order_lines.delivery_days)
IS NOT (excluded.status, excluded.quantity, excluded.revenue,
excluded.courier, excluded.delivery_days)
"""
BUILD_MONTHLY = """
INSERT INTO monthly_sales (month, orders, revenue, target, pct_of_target)
SELECT f.month, COUNT(DISTINCT f.order_id), SUM(f.revenue), t.target,
ROUND(100.0 * SUM(f.revenue) / t.target, 1)
FROM fact_order_lines f LEFT JOIN stg_targets t ON t.month = f.month
WHERE f.status <> 'cancelled'
GROUP BY f.month
ON CONFLICT (month) DO UPDATE SET
orders = excluded.orders, revenue = excluded.revenue,
target = excluded.target, pct_of_target = excluded.pct_of_target
WHERE (monthly_sales.orders, monthly_sales.revenue, monthly_sales.target)
IS NOT (excluded.orders, excluded.revenue, excluded.target)
"""
def count(con, table):
return con.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
def load(wh, rows, targets):
"""Upsert fact rows, then rebuild the monthly summary with SQL (the ELT step)."""
before, changes = count(wh, "fact_order_lines"), wh.total_changes
wh.executemany(UPSERT_FACT, rows)
inserted = count(wh, "fact_order_lines") - before
updated = wh.total_changes - changes - inserted
wh.execute("CREATE TEMP TABLE IF NOT EXISTS stg_targets (month TEXT PRIMARY KEY, target INTEGER)")
wh.execute("DELETE FROM stg_targets")
wh.executemany("INSERT INTO stg_targets VALUES (:month, :target)", targets)
changes = wh.total_changes
wh.execute(BUILD_MONTHLY)
return inserted, updated, wh.total_changes - changesReconcile
Validation checks rows; reconciliation checks totals: does the target agree with the source? Here every order line in the shop must be in the warehouse, with the same total revenue. It is the cheapest test that catches a whole class of bugs: a join that drops rows, a filter that is too strict, a load that half-failed.
def reconcile(src, wh):
source = src.execute("SELECT COUNT(*), SUM(quantity * unit_price) FROM order_items").fetchone()
target = wh.execute("SELECT COUNT(*), SUM(revenue) FROM fact_order_lines").fetchone()
if source != target:
raise RuntimeError(f"reconciliation failed: source {source} vs warehouse {target}")
return targetRun: full or incremental, with a watermark
A full load extracts everything every time: simple and always correct, but slow when the source has years of data. An incremental load extracts only what is new since the last run, remembered as a watermark (here the latest order date loaded). The catch: rows that changed but are older than the watermark are missed. Order 13 is pending today; when it ships next week, its date does not change. So the extractor goes back a lookback window before the watermark. Re-reading a few days twice is harmless, precisely because the load is idempotent.
Note also the comparison: order_date >= ?, not >. A watermark is the largest value seen so far, and more rows with exactly that value can still arrive (another order on the same day, or the same second for a timestamp). With > they would be skipped forever; with >= the boundary is read again, which the upsert makes harmless.
LOOKBACK_DAYS = 14
def run_pipeline(api, src_path="shop.db", wh_path="warehouse.db", full=False):
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s",
stream=sys.stdout, force=True)
src, wh = sqlite3.connect(src_path), sqlite3.connect(wh_path)
mode = None
try:
wh.executescript(WAREHOUSE_DDL)
state = wh.execute("SELECT value FROM etl_state WHERE key = 'orders_watermark'").fetchone()
mode = "full" if full or state is None else "incremental"
since = "0001-01-01" if mode == "full" else \
(date.fromisoformat(state[0]) - timedelta(days=LOOKBACK_DAYS)).isoformat()
lines = extract_order_lines(src, since)
deliveries = with_retries(api.get_deliveries)
targets = extract_targets("targets.csv")
log.info("%s extract since %s: %d order lines, %d deliveries, %d targets",
mode, since, len(lines), len(deliveries), len(targets))
rows = transform(lines, deliveries)
validate(rows)
with wh: # one transaction: all of it or none of it
inserted, updated, monthly = load(wh, rows, targets)
if rows:
wh.execute("INSERT INTO etl_state VALUES ('orders_watermark', ?) "
"ON CONFLICT (key) DO UPDATE SET value = excluded.value",
(max(r["order_date"] for r in rows),))
wh.execute("INSERT INTO etl_runs (mode, since, extracted, inserted, updated, status) "
"VALUES (?, ?, ?, ?, ?, 'ok')", (mode, since, len(rows), inserted, updated))
n, revenue = reconcile(src, wh)
log.info("load: %d inserted, %d updated, %d monthly rows changed; warehouse %d lines, %d taka",
inserted, updated, monthly, n, revenue)
return inserted, updated, monthly
except Exception:
log.exception("run failed")
with wh:
wh.execute("INSERT INTO etl_runs (mode, status) VALUES (?, 'failed')", (mode,))
raise
finally:
src.close()
wh.close()Notice the transaction around the load. If the process dies halfway, the warehouse keeps the previous complete state and the watermark does not move, so the next run simply does the work again.
Running it twice
The first run finds no watermark and does a full load. The courier API times out twice before answering:
run_pipeline(CourierAPI("courier_api.json", fail_first=2))WARNING etl: attempt 1 failed (courier API timed out), retrying in 0.01s
WARNING etl: attempt 2 failed (courier API timed out), retrying in 0.02s
INFO etl: full extract since 0001-01-01: 20 order lines, 12 deliveries, 6 targets
INFO etl: load: 20 inserted, 0 updated, 6 monthly rows changed; warehouse 20 lines, 103550 takaNow the test that matters. Take a fingerprint of the warehouse tables, run the pipeline again, and compare:
import hashlib
def fingerprint(path="warehouse.db"):
con = sqlite3.connect(path)
rows = con.execute("SELECT * FROM fact_order_lines ORDER BY order_id, product_id").fetchall()
rows += con.execute("SELECT * FROM monthly_sales ORDER BY month").fetchall()
con.close()
return hashlib.sha256(repr(rows).encode()).hexdigest()[:12]
before = fingerprint()
run_pipeline(CourierAPI("courier_api.json"))
print("unchanged:", fingerprint() == before)INFO etl: incremental extract since 2026-05-26: 3 order lines, 12 deliveries, 6 targets
INFO etl: load: 0 inserted, 0 updated, 0 monthly rows changed; warehouse 20 lines, 103550 taka
unchanged: TrueThe second run was incremental: it went back 14 days from the watermark, re-read the 3 lines of orders 13 and 14, found nothing different, and wrote nothing. Rerunning after a crash, a retry by the scheduler, or a person pressing "run" twice is now safe. The result in the warehouse:
wh = sqlite3.connect("warehouse.db")
for row in wh.execute("SELECT * FROM monthly_sales ORDER BY month"):
print(row)
print(wh.execute("""SELECT courier, COUNT(DISTINCT order_id), ROUND(AVG(delivery_days), 1)
FROM fact_order_lines WHERE courier IS NOT NULL
GROUP BY courier ORDER BY courier""").fetchall())
wh.close()('2026-01', 2, 6600, 8000, 82.5)
('2026-02', 2, 7000, 8000, 87.5)
('2026-03', 3, 21400, 20000, 107.0)
('2026-04', 2, 23450, 20000, 117.3)
('2026-05', 2, 8500, 15000, 56.7)
('2026-06', 2, 18100, 20000, 90.5)
[('Pathao', 5, 2.3), ('RedX', 4, 3.5), ('Steadfast', 3, 2.0)]When the source changes
A week later, a new order arrives and pending order 13 ships. The shop database changes like this:
INSERT INTO orders (order_id, customer_id, order_date, status) VALUES (15, 8, '2026-06-20', 'pending');
INSERT INTO order_items (order_id, product_id, quantity, unit_price) VALUES (15, 1, 1, 1200), (15, 4, 2, 900);
UPDATE orders SET status = 'shipped' WHERE order_id = 13;
SELECT order_id, order_date, status FROM orders WHERE order_id >= 13 ORDER BY order_id;+----------+------------+-----------+
| order_id | order_date | status |
+----------+------------+-----------+
| 13 | 2026-06-02 | shipped |
| 14 | 2026-06-09 | delivered |
| 15 | 2026-06-20 | pending |
+----------+------------+-----------+Order 13 is dated 2026-06-02, a week before the watermark (2026-06-09). An extractor that read only rows newer than the watermark would never see its change:
src = sqlite3.connect("shop.db")
newer_only = extract_order_lines(src, "2026-06-10")
print("without lookback:", sorted({r["order_id"] for r in newer_only}))
src.close()
run_pipeline(CourierAPI("courier_api.json"))without lookback: [15]
INFO etl: incremental extract since 2026-05-26: 5 order lines, 12 deliveries, 6 targets
INFO etl: load: 2 inserted, 1 updated, 1 monthly rows changed; warehouse 22 lines, 106550 takaWith the lookback window the run picked up both changes: two new lines for order 15 and the status update of order 13, and the June summary was refreshed. Reconciliation still balances.
A lookback window is a heuristic: an order updated 20 days after it was placed would still be missed. Reliable incremental loads need the source to say what changed: an
updated_atcolumn maintained on every write (a trigger can do it, see Views, Stored Procedures, Functions and Triggers) used as the watermark, or change data capture (CDC), which reads the database's own change log (the MySQL binlog, the PostgreSQL WAL) with a tool such as Debezium and streams every insert, update and delete. CDC catches deletes too: a deleted row is simply not there for a watermark query to find, unless the source only marks rows as deleted (a "soft delete" flag) instead of removing them.
Running it in production
Scheduling
The simplest scheduler is cron. This line runs the pipeline at 02:15 every night and appends its log to a file. The five fields are minute, hour, day of month, month and day of week; * means "every". cron uses the server's time zone, which on cloud servers is often UTC, not Bangladesh time:
# m h dom mon dow command
15 2 * * * cd /srv/shop-etl && .venv/bin/python run_etl.py >> logs/etl.log 2>&1When pipelines depend on each other (load orders, then rebuild features, then retrain the model), use an orchestrator: Apache Airflow, Dagster or Prefect. They run tasks in order, retry failed ones, keep history and alert on failure. An Airflow DAG (a graph of tasks) for this pipeline looks like this. It is written for Airflow 3, with the Airflow 2 import paths in the comments; it needs an Airflow installation, so it is not run here:
from datetime import datetime, timedelta
from airflow.sdk import DAG # Airflow 2: from airflow import DAG
from airflow.providers.standard.operators.python import PythonOperator
# Airflow 2: from airflow.operators.python import PythonOperator
with DAG(
dag_id="shop_etl",
schedule="15 2 * * *", # same as the cron line
start_date=datetime(2026, 6, 1),
catchup=False,
default_args={"retries": 3, "retry_delay": timedelta(minutes=5)},
) as dag:
load_sales = PythonOperator(task_id="load_sales", python_callable=run_etl_main)
build_features = PythonOperator(task_id="build_features", python_callable=build_features_main)
load_sales >> build_features # features run only after a successful loadscheduletakes the same cron expression. It replacedschedule_intervalin Airflow 2.4, and Airflow 3 accepts onlyschedule. Airflow reads it in UTC unlessstart_datecarries a time zone, so this DAG runs at 02:15 UTC, 08:15 in Dhaka.catchup=Falsestops Airflow from running once for every past interval sincestart_datewhen the DAG is switched on (in Airflow 3 that is already the default).retriesreruns a failed task. That is only safe because the task is idempotent: a retry after a half-finished run must not double-count.- Airflow also has the TaskFlow style, where plain functions decorated with
@dagand@taskbecome the DAG and its tasks.
Monitoring and alerts
A pipeline that fails silently is worse than one that fails loudly. The etl_runs table is a ready-made monitor. Add checks for the failures that do not raise errors: a run that loaded nothing for days, or data that is too old (freshness):
def health_check(wh_path, today, max_age_days=2):
"""Return a list of alerts; an empty list means healthy."""
wh = sqlite3.connect(wh_path)
alerts = []
last = wh.execute("SELECT status FROM etl_runs ORDER BY run_id DESC LIMIT 1").fetchone()
if last is None or last[0] != "ok":
alerts.append("last run did not succeed")
newest = wh.execute("SELECT MAX(order_date) FROM fact_order_lines").fetchone()[0]
age = (today - date.fromisoformat(newest)).days
if age > max_age_days:
alerts.append(f"newest order is {age} days old")
wh.close()
return alerts
print(sqlite3.connect("warehouse.db").execute(
"SELECT run_id, mode, extracted, inserted, updated, status FROM etl_runs ORDER BY run_id").fetchall())
print(health_check("warehouse.db", today=date(2026, 6, 21)))
print(health_check("warehouse.db", today=date(2026, 6, 30)))[(1, 'full', 20, 20, 0, 'ok'), (2, 'incremental', 3, 0, 0, 'ok'), (3, 'incremental', 5, 2, 1, 'ok')]
[]
['newest order is 10 days old']In production the list goes to Slack, email or a pager instead of print. today is a parameter rather than date.today() so the check can be tested; the real caller passes today's date.
Testing a pipeline
Pipelines deserve tests like any code. Test the transform with tiny hand-made inputs, and test idempotency end to end on a throwaway copy of the data:
import shutil
def test_transform_computes_delivery_days():
lines = [{"order_id": 1, "product_id": 1, "order_date": "2026-01-12", "status": "delivered",
"category": "Books", "quantity": 1, "revenue": 1200}]
rows = transform(lines, [{"order_id": 1, "courier": "Pathao", "delivered_on": "2026-01-14"}])
assert rows[0]["delivery_days"] == 2 and rows[0]["month"] == "2026-01"
def test_pipeline_is_idempotent():
shutil.copy("shop.db", "test_shop.db")
logging.disable(logging.CRITICAL) # keep the test quiet
try:
first = run_pipeline(CourierAPI("courier_api.json"), "test_shop.db", "test_wh.db", full=True)
snapshot = fingerprint("test_wh.db")
second = run_pipeline(CourierAPI("courier_api.json"), "test_shop.db", "test_wh.db", full=True)
finally:
logging.disable(logging.NOTSET)
assert first[0] > 0 and second == (0, 0, 0)
assert fingerprint("test_wh.db") == snapshot
test_transform_computes_delivery_days()
test_pipeline_is_idempotent()
print("pipeline tests passed")pipeline tests passedEven a full reload changes nothing the second time. That is the property to protect: if a future change breaks it, this test fails before production does.
Where this shows up
- Analytics: nightly ELT from the app database into BigQuery or Snowflake, with dbt models building the reporting tables.
- ML: feature pipelines that recompute customer features daily; idempotency means a re-run never trains on duplicated rows.
- RAG: a document pipeline that extracts new and changed pages, chunks and embeds them, and upserts them into a vector table keyed by document and chunk id.
Common mistakes
- Plain INSERTs in the load. Any re-run duplicates data. Give the target a natural key and upsert, or delete-and-reload a whole partition (a day, a month) inside one transaction.
- A watermark with no lookback. Late updates to older rows are missed forever. Use a lookback window, an
updated_atcolumn, or CDC. - Moving the watermark before the load commits. If the load then fails, the next run starts after data that was never loaded. Update state in the same transaction as the data, as
run_pipelinedoes. - Retrying everything. Retrying a validation error or a wrong password just delays the alert. Retry only transient errors (timeouts, connection resets, HTTP 429/503), with backoff, jitter and a limit.
- No reconciliation. A join that silently drops 3% of rows passes every row-level check. Compare counts and sums between source and target on every run.
Try it yourself
- Easy: query
warehouse.dbfor revenue per category fromfact_order_lines, excluding cancelled orders, highest first. - Medium: make reconciliation stricter: also compare revenue per
statusbetween source and warehouse, and raise if any status differs. Run your function on the current data. - Hard: change order 15's quantity of product 4 from 2 to 3 in
shop.db, run the pipeline, and show that exactly one fact row was updated and the June row ofmonthly_saleschanged. Then run it once more and show nothing changed.
Answers
# 1. Easy
wh = sqlite3.connect("warehouse.db")
print(wh.execute("""SELECT category, SUM(revenue) FROM fact_order_lines
WHERE status <> 'cancelled' GROUP BY category ORDER BY 2 DESC""").fetchall())
wh.close()
# 2. Medium
def reconcile_by_status(src_path="shop.db", wh_path="warehouse.db"):
src, wh = sqlite3.connect(src_path), sqlite3.connect(wh_path)
source = dict(src.execute("""SELECT o.status, SUM(oi.quantity * oi.unit_price)
FROM orders o JOIN order_items oi ON oi.order_id = o.order_id
GROUP BY o.status""").fetchall())
target = dict(wh.execute("SELECT status, SUM(revenue) FROM fact_order_lines GROUP BY status").fetchall())
src.close()
wh.close()
if source != target:
raise RuntimeError(f"status totals differ: {source} vs {target}")
return target
print(reconcile_by_status())
# 3. Hard
src = sqlite3.connect("shop.db")
with src:
src.execute("UPDATE order_items SET quantity = 3 WHERE order_id = 15 AND product_id = 4")
src.close()
print(run_pipeline(CourierAPI("courier_api.json")))
print(sqlite3.connect("warehouse.db").execute(
"SELECT * FROM monthly_sales WHERE month = '2026-06'").fetchone())
print(run_pipeline(CourierAPI("courier_api.json")))Summary
- A pipeline extracts, transforms and loads. ETL transforms before loading; ELT loads raw data and transforms with SQL in the warehouse. Most real pipelines mix both.
- Incremental loads use a watermark plus a lookback window (or
updated_at, or CDC) so late changes are not missed. - Idempotency comes from keys and upserts, with state updated in the same transaction as the data. Prove it: run twice and compare.
- Validate rows before loading, reconcile totals after, log every run, and retry only transient errors with backoff.
- Schedule with cron or an orchestrator (Airflow), monitor runs and freshness, alert on problems, and test transforms and idempotency.
Next: Preparing ML-Ready Data with SQL and pandas takes data that a pipeline like this delivers and turns it into a training set: features and a target, point-in-time correctness to avoid leakage, and a reproducible dataset that feeds a scikit-learn model.