Project: build an end-to-end data pipeline

I am building a small batch data pipeline in Python and DuckDB for a coffee shop. Input: one JSON-lines file per day, each line an order with id, customer_name, items (a list of item_id and qty), status and created_at. Output: a clean order-lines table, a rejects table with reasons, and a daily sales table.

Do not write code yet. List the stages, the tables with their columns, the data quality problems I should expect in the raw files, and how to make each stage safe to re-run.

import json
import random
from datetime import datetime, timedelta
from pathlib import Path

MENU = {
    "espresso": 120, "cappuccino": 180, "cold-brew": 220, "filter-coffee": 80,
    "masala-chai": 90, "croissant": 150, "banana-bread": 130, "veg-sandwich": 160,
}
OUT = Path("data/raw")


def make_day(day: str, seed: int, n: int = 40) -> None:
    rng = random.Random(seed)
    prefix = day.replace("-", "")
    start = datetime.fromisoformat(day + "T08:00:00")
    rows = []
    for i in range(1, n + 1):
        items = [
            {"item_id": item, "qty": rng.randint(1, 3)}
            for item in rng.sample(list(MENU), rng.randint(1, 3))
        ]
        rows.append({
            "id": f"{prefix}-{i:03d}",
            "customer_name": rng.choice(["Asha", "Ravi", "Meera", "Imran", "Kavya"]),
            "items": items,
            "status": rng.choice(["picked_up"] * 8 + ["cancelled"]),
            "created_at": (start + timedelta(minutes=rng.randint(0, 14 * 60))).isoformat(),
        })
    # Three deliberately bad rows, so the pipeline has something to catch.
    rows.append(dict(rows[0]))  # duplicate order id
    rows.append({**rows[1], "id": f"{prefix}-900",
                 "items": [{"item_id": "pizza", "qty": 1}]})  # item not on the menu
    rows.append({**rows[2], "id": f"{prefix}-901",
                 "items": [{"item_id": "espresso", "qty": -2}]})  # negative quantity
    OUT.mkdir(parents=True, exist_ok=True)
    with open(OUT / f"orders_{day}.jsonl", "w", encoding="utf-8") as f:
        for r in rows:
            f.write(json.dumps(r) + "\n")


if __name__ == "__main__":
    for k, day in enumerate(["2026-10-01", "2026-10-02", "2026-10-03"]):
        make_day(day, seed=k)
    with open("data/menu.json", "w", encoding="utf-8") as f:
        json.dump(MENU, f)
    print("Wrote 3 days of raw orders to data/raw/")

python generate_data.py

import json
import sys
from pathlib import Path

import duckdb

DB = "warehouse.duckdb"
RAW_DIR = Path("data/raw")
MAX_REJECT_RATE = 0.10


def ingest(con, day: str) -> int:
    """Stage 1: load one day of raw JSON lines, exactly as received."""
    con.execute("""
        CREATE TABLE IF NOT EXISTS raw_orders (
            load_day VARCHAR, line_no INTEGER, payload VARCHAR
        )""")
    con.execute("DELETE FROM raw_orders WHERE load_day = ?", [day])  # makes re-runs safe
    path = RAW_DIR / f"orders_{day}.jsonl"
    with open(path, encoding="utf-8") as f:
        rows = [(day, n, line.strip()) for n, line in enumerate(f, 1) if line.strip()]
    con.executemany("INSERT INTO raw_orders VALUES (?, ?, ?)", rows)
    return len(rows)


def load_menu(con) -> None:
    with open("data/menu.json", encoding="utf-8") as f:
        menu = json.load(f)
    con.execute("CREATE OR REPLACE TABLE dim_menu (item_id VARCHAR, price INTEGER)")
    con.executemany("INSERT INTO dim_menu VALUES (?, ?)", list(menu.items()))


def transform(con, day: str) -> None:
    """Stage 2: parse the raw JSON into clean tables. Bad orders go to rejects."""
    con.execute("""
        CREATE TABLE IF NOT EXISTS fct_order_lines (
            load_day VARCHAR, order_id VARCHAR, customer_name VARCHAR, status VARCHAR,
            created_at TIMESTAMP, item_id VARCHAR, qty INTEGER, price INTEGER, line_total INTEGER
        )""")
    con.execute("""
        CREATE TABLE IF NOT EXISTS rejects (
            load_day VARCHAR, order_id VARCHAR, reason VARCHAR
        )""")
    con.execute("DELETE FROM fct_order_lines WHERE load_day = ?", [day])
    con.execute("DELETE FROM rejects WHERE load_day = ?", [day])

    # One row per raw line, with the JSON fields pulled out.
    con.execute("""
        CREATE OR REPLACE TEMP TABLE parsed AS
        SELECT
            load_day,
            payload::JSON ->> '$.id'            AS order_id,
            payload::JSON ->> '$.customer_name' AS customer_name,
            payload::JSON ->> '$.status'        AS status,
            CAST(payload::JSON ->> '$.created_at' AS TIMESTAMP) AS created_at,
            from_json(payload::JSON -> '$.items',
                      '[{"item_id": "VARCHAR", "qty": "INTEGER"}]') AS items,
            row_number() OVER (
                PARTITION BY payload::JSON ->> '$.id' ORDER BY line_no
            ) AS seen_nth
        FROM raw_orders
        WHERE load_day = ?""", [day])

    # One row per item in each first-seen order.
    con.execute("""
        CREATE OR REPLACE TEMP TABLE lines AS
        SELECT load_day, order_id, customer_name, status, created_at,
               item.item_id AS item_id, item.qty AS qty
        FROM (SELECT *, unnest(items) AS item FROM parsed WHERE seen_nth = 1)""")

    con.execute("""
        INSERT INTO rejects
        SELECT load_day, order_id, 'duplicate order id' FROM parsed WHERE seen_nth > 1""")
    con.execute("""
        INSERT INTO rejects
        SELECT DISTINCT load_day, order_id, 'unknown item_id' FROM lines
        WHERE item_id NOT IN (SELECT item_id FROM dim_menu)""")
    con.execute("""
        INSERT INTO rejects
        SELECT DISTINCT load_day, order_id, 'qty must be positive' FROM lines
        WHERE qty <= 0""")

    # A single bad line rejects the whole order.
    con.execute("""
        INSERT INTO fct_order_lines
        SELECT l.load_day, l.order_id, l.customer_name, l.status, l.created_at,
               l.item_id, l.qty, m.price, l.qty * m.price
        FROM lines l
        JOIN dim_menu m USING (item_id)
        WHERE l.order_id NOT IN (SELECT order_id FROM rejects WHERE load_day = ?)""", [day])


def check(con, day: str) -> list[str]:
    """Stage 3: return a list of problems. An empty list means the data is healthy."""
    problems = []
    n = con.execute(
        "SELECT count(*) FROM fct_order_lines WHERE load_day = ?", [day]).fetchone()[0]
    if n == 0:
        problems.append("no rows loaded for the day")
    dup = con.execute("""
        SELECT count(*) FROM (
            SELECT order_id, item_id FROM fct_order_lines WHERE load_day = ?
            GROUP BY order_id, item_id HAVING count(*) > 1)""", [day]).fetchone()[0]
    if dup:
        problems.append(f"{dup} order lines repeat the same item within an order")
    bad = con.execute(
        "SELECT count(*) FROM fct_order_lines WHERE load_day = ? AND line_total <= 0",
        [day]).fetchone()[0]
    if bad:
        problems.append(f"{bad} lines with a non-positive total")
    total = con.execute(
        "SELECT count(*) FROM raw_orders WHERE load_day = ?", [day]).fetchone()[0]
    rejected = con.execute(
        "SELECT count(*) FROM rejects WHERE load_day = ?", [day]).fetchone()[0]
    if total and rejected / total > MAX_REJECT_RATE:
        problems.append(f"reject rate too high: {rejected} of {total} orders")
    return problems


def publish(con, day: str) -> None:
    """Stage 4: build the reporting table. Only runs after checks pass."""
    con.execute("""
        CREATE TABLE IF NOT EXISTS daily_sales (
            load_day VARCHAR, item_id VARCHAR, units INTEGER, revenue INTEGER
        )""")
    con.execute("DELETE FROM daily_sales WHERE load_day = ?", [day])
    con.execute("""
        INSERT INTO daily_sales
        SELECT load_day, item_id, sum(qty), sum(line_total)
        FROM fct_order_lines
        WHERE load_day = ? AND status = 'picked_up'
        GROUP BY load_day, item_id""", [day])


def run(day: str) -> None:
    con = duckdb.connect(DB)
    loaded = ingest(con, day)
    load_menu(con)
    transform(con, day)
    problems = check(con, day)
    if problems:
        print(f"[{day}] FAILED checks, not publishing: {problems}")
        sys.exit(1)
    publish(con, day)
    rejected = con.execute(
        "SELECT count(*) FROM rejects WHERE load_day = ?", [day]).fetchone()[0]
    print(f"[{day}] ok: {loaded} raw lines, {rejected} rejected, report updated")


if __name__ == "__main__":
    run(sys.argv[1])

python pipeline.py 2026-10-01
python pipeline.py 2026-10-02
python pipeline.py 2026-10-03

import duckdb

con = duckdb.connect("warehouse.duckdb", read_only=True)

print("Revenue by day (picked-up orders only)")
for day, revenue in con.execute(
    "SELECT load_day, sum(revenue) FROM daily_sales GROUP BY load_day ORDER BY load_day"
).fetchall():
    print(f"  {day}  Rs {revenue}")

print("\nBest sellers, all days")
for item, units in con.execute(
    "SELECT item_id, sum(units) AS u FROM daily_sales GROUP BY item_id ORDER BY u DESC LIMIT 3"
).fetchall():
    print(f"  {item:<14} {units} units")

print("\nRejected orders")
for day, order_id, reason in con.execute(
    "SELECT load_day, order_id, reason FROM rejects ORDER BY load_day, order_id"
).fetchall():
    print(f"  {day}  {order_id}  {reason}")

python report.py

import duckdb
import pytest

import pipeline


@pytest.fixture
def con():
    c = duckdb.connect(":memory:")
    pipeline.ingest(c, "2026-10-01")
    pipeline.load_menu(c)
    pipeline.transform(c, "2026-10-01")
    return c


def test_bad_orders_are_rejected_with_reasons(con):
    reasons = {r[0] for r in con.execute("SELECT reason FROM rejects").fetchall()}
    assert reasons == {"duplicate order id", "unknown item_id", "qty must be positive"}


def test_clean_table_has_no_bad_rows(con):
    bad = con.execute(
        "SELECT count(*) FROM fct_order_lines WHERE qty <= 0 OR price IS NULL").fetchone()[0]
    assert bad == 0


def test_checks_pass_on_good_day(con):
    assert pipeline.check(con, "2026-10-01") == []


def test_rerun_does_not_duplicate(con):
    before = con.execute("SELECT count(*) FROM fct_order_lines").fetchone()[0]
    pipeline.ingest(con, "2026-10-01")
    pipeline.transform(con, "2026-10-01")
    after = con.execute("SELECT count(*) FROM fct_order_lines").fetchone()[0]
    assert before == after

pytest -q

You are a senior data engineer doing a code review. Review the pipeline below for correctness, idempotency, failure handling and data quality gaps. List the 5 most important problems, ordered by risk, and for each give a concrete fix.

<code>
[paste pipeline.py]
</code>