PART 6 · PYTHON FOR DATA ENGINEERING · CHAPTER 33

Build an ETL pipeline: idempotent, logged, configurable

FreeLesson 70 of 78

One file, three pure-ish functions and a main(). Run it twice for the same day and the warehouse does not change: that is the property interviewers probe.

ORDERS_DAILY.PYPYTHON 3.12

import argparse, csv, logging, sqlite3, sys

logging.basicConfig(level=logging.INFO, stream=sys.stdout,
                    format="%(levelname)s %(name)s: %(message)s")
log = logging.getLogger("orders_daily")

def extract(path, day):                       # E: stream one day
    with open(path, newline="") as f:
        for r in csv.DictReader(f):
            if r["order_date"] == day:
                yield r

def transform(rows):                          # T: pure, testable
    for r in rows:
        yield (int(r["order_id"]), int(r["customer_id"]), r["order_date"],
               r["status"].strip().lower(), round(float(r["amount"]), 2))

def load(con, day, rows):                     # L: idempotent
    with con:                                 # one transaction
        con.execute("DELETE FROM fact_orders WHERE order_date = ?", (day,))
        cur = con.executemany("INSERT INTO fact_orders VALUES (?,?,?,?,?)", rows)
    return cur.rowcount

def main(argv=None):
    p = argparse.ArgumentParser(description="Load one day of orders")
    p.add_argument("--date", required=True)
    p.add_argument("--src", default="orders.csv")
    args = p.parse_args(argv)
    con = sqlite3.connect("warehouse.db")
    con.execute("""CREATE TABLE IF NOT EXISTS fact_orders (order_id INT PRIMARY KEY,
                   customer_id INT, order_date TEXT, status TEXT, amount REAL)""")
    n = load(con, args.date, transform(extract(args.src, args.date)))
    total = con.execute("SELECT COUNT(*) FROM fact_orders").fetchone()[0]
    log.info("date=%s loaded=%d table_rows=%d", args.date, n, total)
    return 0

if __name__ == "__main__":
    main(["--date", "2025-04-20"])
    main(["--date", "2025-04-20"])            # rerun the same day
    main(["--date", "2025-04-28"])
    try:
        main([])                              # missing --date
    except SystemExit as e:
        log.error("bad arguments, exit code %s", e.code)

Output

INFO orders_daily: date=2025-04-20 loaded=1 table_rows=1
INFO orders_daily: date=2025-04-20 loaded=1 table_rows=1
INFO orders_daily: date=2025-04-28 loaded=1 table_rows=2
ERROR orders_daily: bad arguments, exit code 2

Decision in the code	Why it matters in production
DELETE the day, then INSERT, in one transaction	Reruns and backfills give the same table. A crash halfway rolls back, so there is never a half-loaded day.
extract and transform are generators	Memory stays flat whether the day has 15 rows or 15 million.
logging, not print, and a pure transform	Log viewers can filter levels and names. The pure transform is unit tested on the next page.
argparse with --date	The scheduler passes the date. The same file runs today, for a backfill, or by hand.
bad input exits with code 2	A non-zero exit code is how the scheduler knows the task failed.

Practice before you move on

Run the pipeline twice and check that the second run leaves the output exactly as the first did.
Move the output folder into the config and log its value when the run starts.