Lakehouse pipeline from raw events to a dashboard

Ten business questions, a design document that fixes grains and metric definitions, and the local platform (object storage, Airflow, DuckDB) running.

1. 1.1
   Create the repository and write the ten questions before anything else. Use an online shop: orders, order lines, customers (who move country), products. At least three questions need history and one needs the last hour.

   bashCopy

   gh repo create lakehouse --public --clone && cd lakehouse
   mkdir -p generate pipelines dbt dags governance dashboard tests
   cat > DESIGN.md <<'EOF'
   # Lakehouse design
   
   ## Questions
   Q1  Daily revenue by customer country, as it was at order time
   Q2  Orders placed in the last hour
   Q3  Repeat-purchase rate by signup month cohort
   Q4  Top 10 products by revenue, last 30 days
   Q5  Average order value by month, paid orders only
   Q6  Refund rate by product category
   Q7  Customers whose country changed, and revenue before vs after
   Q8  Median hours from order to shipment, by month
   Q9  Daily active customers (placed >= 1 order)
   Q10 Revenue lost to cancellations, by week
   EOF

2. 1.2
   Finish the design document: grains, SCD decisions, the single definition of revenue, layers, lateness and PII. This is the file you check the implementation against in every later phase.

   textCopy

   ## Facts and dimensions
   fct_order_lines   grain: one row per order line (order_id, line_no)   measures: quantity, unit_price, line_amount
   fct_orders        grain: one row per order (order_id)                  measures: amount, status timestamps
   dim_customers     SCD2 on country and segment; surrogate customer_key; natural customer_id
   dim_products      SCD1; category, brand
   dim_date          generated 2024-2027
   
   ## Metric definitions (one each)
   revenue        = SUM(line_amount) FROM fct_order_lines JOIN fct_orders USING (order_id) WHERE status IN ('paid','shipped')
   refund_rate    = COUNT(orders with status='refunded') / COUNT(orders with status IN ('paid','shipped','refunded'))
   repeat_rate    = customers with >= 2 paid orders / customers with >= 1 paid order, per signup month
   
   ## Layers
   raw/     s3://lake/raw/orders/dt=YYYY-MM-DD/batch-*.jsonl.gz     immutable; 730 days
   bronze/  s3://lake/bronze/orders/event_date=YYYY-MM-DD/*.parquet typed; quarantine table for rejects
   silver/  iceberg: orders, order_lines, customers                   merged by key, deletes applied, last_version guard
   gold/    dbt marts in DuckDB (warehouse.duckdb)                    tested, documented
   
   ## Lateness and change
   events arrive up to 3 days late -> loader reprocesses a 3-day trailing window; older -> manual backfill
   status changes arrive as new versions of the same order_id -> merged; version = updated_at
   deletes arrive as op='d' events -> applied in silver; kept in bronze history
   
   ## PII
   customers.email (direct, hash in marts), customers.dob (indirect, year only), addresses (direct, not loaded at all)

3. 1.3
   Write the Compose stack: MinIO with a bucket created on start, Airflow standalone with the DAGs folder mounted, and the dashboard.

   yamlCopy

   services:
     minio:
       image: minio/minio:RELEASE.2025-01-20T14-49-07Z
       command: server /data --console-address ":9001"
       environment:
         MINIO_ROOT_USER: lake
         MINIO_ROOT_PASSWORD: lake-secret-1
       ports: ["9000:9000", "9001:9001"]
       volumes: ["minio-data:/data"]
   
     minio-init:
       image: minio/mc:RELEASE.2025-01-17T23-25-50Z
       depends_on: [minio]
       entrypoint: >
         /bin/sh -c "until mc alias set local http://minio:9000 lake lake-secret-1; do sleep 1; done;
         mc mb --ignore-existing local/lake; mc anonymous set none local/lake; exit 0"
   
     airflow:
       image: apache/airflow:2.10.4-python3.12
       command: standalone
       environment:
         AIRFLOW__CORE__LOAD_EXAMPLES: "false"
         AIRFLOW__CORE__EXECUTOR: LocalExecutor
         AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: sqlite:////opt/airflow/airflow.db
         AWS_ACCESS_KEY_ID: lake
         AWS_SECRET_ACCESS_KEY: lake-secret-1
         AWS_ENDPOINT_URL: http://minio:9000
         _PIP_ADDITIONAL_REQUIREMENTS: "polars==1.19.0 pydantic==2.10.4 pyiceberg[s3fs,sql-sqlite]==0.8.1 duckdb==1.1.3 dbt-duckdb==1.9.1 boto3==1.35.90"
       ports: ["8080:8080"]
       volumes:
         - ./dags:/opt/airflow/dags
         - ./pipelines:/opt/airflow/pipelines
         - ./dbt:/opt/airflow/dbt
         - ./lake:/opt/airflow/lake         # iceberg catalog sqlite + warehouse.duckdb live here
       depends_on: [minio-init]
   
     dashboard:
       build: ./dashboard
       ports: ["8501:8501"]
       volumes: ["./lake:/lake:ro"]
       depends_on: [airflow]
   
   volumes:
     minio-data:

   _PIP_ADDITIONAL_REQUIREMENTS is fine for a lab; for anything longer-lived build a custom Airflow image with the packages baked in. AWS_ENDPOINT_URL makes boto3, s3fs and DuckDB's httpfs talk to MinIO.
4. 1.4
   Start the stack and confirm the pieces: MinIO console, an empty lake bucket, and Airflow with no DAG import errors yet.

   bashCopy

   mkdir -p lake && docker compose up -d
   sleep 60 && docker compose logs airflow 2>&1 | grep -i 'password' | head -2   # standalone prints the admin password

   Check: http://localhost:9001 (lake / lake-secret-1) shows the lake bucket; http://localhost:8080 logs in with the printed admin password.

A generator that produces a year of events including late arrivals, corrections and deletes, landing as immutable compressed files in the raw zone by arrival date.

1. 2.1
   Write the generator. It emits newline-delimited JSON events for customers and orders, with an op (c/u/d), an updated_at version, and deliberately messy behaviour: 5 % of orders arrive one to three days late, 10 % get a status update later, 1 % are deleted, and two customers change country mid-year.

   pythonCopy

   # generate/make_events.py
   import gzip
   import json
   import random
   from datetime import date, datetime, timedelta, timezone
   from pathlib import Path
   
   import boto3
   
   random.seed(7)
   START, DAYS = date(2026, 1, 1), 270
   COUNTRIES = ["DE", "FR", "PT", "NL", "ES"]
   PRODUCTS = [(f"P{i:03d}", random.choice(["books", "audio", "home", "toys"]), round(random.uniform(5, 120), 2)) for i in range(1, 61)]
   
   
   def ts(d: date, h: int = 12) -> str:
       return datetime(d.year, d.month, d.day, h, random.randint(0, 59), tzinfo=timezone.utc).isoformat()
   
   
   def customers(n=400):
       for i in range(1, n + 1):
           signup = START + timedelta(days=random.randint(0, 200))
           yield {"op": "c", "customer_id": i, "email": f"user{i}@example.com", "country": random.choice(COUNTRIES),
                  "segment": random.choice(["new", "regular", "vip"]), "dob": f"19{random.randint(60, 99)}-06-15",
                  "signup_date": signup.isoformat(), "updated_at": ts(signup, 9), "arrival_date": signup.isoformat()}
       for cid in (7, 42):   # country changes mid-year: SCD2 material
           moved = START + timedelta(days=150)
           yield {"op": "u", "customer_id": cid, "country": "PT", "updated_at": ts(moved, 10), "arrival_date": moved.isoformat()}
   
   
   def orders():
       oid = 1000
       for day in (START + timedelta(days=i) for i in range(DAYS)):
           for _ in range(random.randint(20, 60)):
               oid += 1
               cid = random.randint(1, 400)
               lines = [{"line_no": n + 1, "product_id": (p := random.choice(PRODUCTS))[0], "quantity": random.randint(1, 3), "unit_price": p[2]}
                        for n in range(random.randint(1, 4))]
               late = timedelta(days=random.choice([1, 2, 3])) if random.random() < 0.05 else timedelta(0)
               base = {"op": "c", "order_id": oid, "customer_id": cid, "ordered_at": ts(day), "status": "placed",
                       "lines": lines, "updated_at": ts(day), "arrival_date": (day + late).isoformat()}
               yield base
               r = random.random()
               if r < 0.85:
                   yield {**base, "op": "u", "status": "paid", "updated_at": ts(day, 13), "arrival_date": (day + late).isoformat()}
                   if random.random() < 0.9:
                       ship = day + timedelta(days=random.randint(1, 5))
                       yield {**base, "op": "u", "status": "shipped", "shipped_at": ts(ship, 8), "updated_at": ts(ship, 8), "arrival_date": ship.isoformat()}
                   if random.random() < 0.04:
                       ref = day + timedelta(days=random.randint(6, 20))
                       yield {**base, "op": "u", "status": "refunded", "updated_at": ts(ref, 15), "arrival_date": ref.isoformat()}
               elif r < 0.95:
                   yield {**base, "op": "u", "status": "cancelled", "updated_at": ts(day, 14), "arrival_date": day.isoformat()}
               if random.random() < 0.01:
                   yield {"op": "d", "order_id": oid, "updated_at": ts(day + timedelta(days=1), 9), "arrival_date": (day + timedelta(days=1)).isoformat()}
   
   
   def write(events, entity: str):
       by_day: dict[str, list] = {}
       for e in events:
           by_day.setdefault(e["arrival_date"], []).append(e)
       s3 = boto3.client("s3")
       for arrival, evs in by_day.items():
           key = f"raw/{entity}/dt={arrival}/batch-0001.jsonl.gz"
           body = gzip.compress("\n".join(json.dumps(e) for e in evs).encode())
           s3.put_object(Bucket="lake", Key=key, Body=body)
       print(f"{entity}: {sum(len(v) for v in by_day.values())} events over {len(by_day)} arrival days")
   
   
   if __name__ == "__main__":
       write(list(customers()), "customers")
       write(list(orders()), "orders")

   Run it with the MinIO credentials in the environment: AWS_ACCESS_KEY_ID=lake AWS_SECRET_ACCESS_KEY=lake-secret-1 AWS_ENDPOINT_URL=http://localhost:9000 python generate/make_events.py after pip install boto3. Raw files are keyed by arrival date, which is the point: event date is inside the record.
2. 2.2
   Confirm the raw zone and note its shape: one prefix per arrival day, compressed JSON lines, never to be modified.

   bashCopy

   docker compose exec -T minio sh -c 'mc alias set local http://localhost:9000 lake lake-secret-1 >/dev/null && mc ls --recursive local/lake/raw/ | head -5 && mc du local/lake/raw/'

   Check: About 270 raw/orders/dt=... prefixes plus raw/customers/..., a few tens of MB in total. Add a MinIO lifecycle rule later (mc ilm add --expire-days 730 local/lake/raw) to encode the retention policy.
3. 2.3
   Encode the raw layer's retention as a bucket lifecycle rule so the 730-day policy in DESIGN.md is enforced by the platform rather than remembered.

   bashCopy

   docker compose exec -T minio sh -c 'mc alias set local http://localhost:9000 lake lake-secret-1 >/dev/null && mc ilm rule add --expire-days 730 --prefix raw/ local/lake && mc ilm rule ls local/lake'

   Check: mc ilm rule ls shows one rule on the raw/ prefix expiring objects after 730 days.

A --day parameterised loader that reads the arrival-day files for a trailing window, validates each event, quarantines the bad ones, and overwrites bronze partitions by event date.

1. 3.1
   Write the loader with the shape from the Python and batch modules: logical date in, trailing window, pydantic validation, quarantine with reasons, partition overwrite, telemetry.

   pythonCopy

   # pipelines/load_bronze.py
   import argparse
   import gzip
   import io
   import json
   import logging
   import sys
   import time
   from datetime import date, datetime, timedelta
   from decimal import Decimal
   from typing import Literal
   
   import boto3
   import polars as pl
   from pydantic import BaseModel, ValidationError
   
   from pipelines.telemetry import record_run
   
   log = logging.getLogger("load_bronze")
   BUCKET, LOOKBACK = "lake", 3
   
   
   class Line(BaseModel):
       line_no: int
       product_id: str
       quantity: int
       unit_price: Decimal
   
   
   class OrderEvent(BaseModel):
       op: Literal["c", "u", "d"]
       order_id: int
       customer_id: int | None = None
       ordered_at: datetime | None = None
       status: Literal["placed", "paid", "shipped", "refunded", "cancelled"] | None = None
       shipped_at: datetime | None = None
       lines: list[Line] | None = None
       updated_at: datetime
       arrival_date: date
   
   
   def read_raw(s3, entity: str, arrival: date) -> list[dict]:
       prefix = f"raw/{entity}/dt={arrival.isoformat()}/"
       rows = []
       for obj in s3.list_objects_v2(Bucket=BUCKET, Prefix=prefix).get("Contents", []):
           body = s3.get_object(Bucket=BUCKET, Key=obj["Key"])["Body"].read()
           rows += [json.loads(l) for l in gzip.decompress(body).decode().splitlines() if l]
       return rows
   
   
   def validate(rows: list[dict]) -> tuple[list[dict], list[dict]]:
       good, bad = [], []
       for r in rows:
           try:
               ev = OrderEvent.model_validate(r)
               flat = ev.model_dump(mode="json")
               flat["event_date"] = (ev.ordered_at or ev.updated_at).date().isoformat()
               flat["lines"] = json.dumps(flat["lines"]) if flat.get("lines") is not None else None
               good.append(flat)
           except ValidationError as e:
               bad.append({"raw": json.dumps(r), "errors": json.dumps(e.errors()), "quarantined_at": datetime.utcnow().isoformat()})
       return good, bad
   
   
   def write_partition(s3, df: pl.DataFrame, entity: str, event_date: str) -> None:
       key = f"bronze/{entity}/event_date={event_date}/part-0000.parquet"
       buf = io.BytesIO()
       df.write_parquet(buf, compression="zstd")
       s3.put_object(Bucket=BUCKET, Key=key, Body=buf.getvalue())     # overwrite: idempotent
   
   
   def run(day: date) -> int:
       started = time.time()
       s3 = boto3.client("s3")
       arrivals = [day - timedelta(days=i) for i in range(LOOKBACK)]
       rows = [r for a in arrivals for r in read_raw(s3, "orders", a)]
       good, bad = validate(rows)
       if rows and len(bad) / len(rows) > 0.01:
           raise RuntimeError(f"reject rate {len(bad)/len(rows):.2%} > 1%; refusing to load")
       if bad:
           write_partition(s3, pl.DataFrame(bad), "orders_quarantine", day.isoformat())
       df = pl.DataFrame(good)
       n = 0
       if df.height:
           # a partition may receive events from several arrival days; rebuild each event_date partition from
           # ALL raw arrival days in the window that touch it, so late events land where they belong
           for (event_date,), part in df.group_by("event_date", maintain_order=True):
               write_partition(s3, part, "orders", event_date)
               n += part.height
       record_run("load_bronze", "bronze.orders", day, started, rows_in=len(rows), rows_out=n, rows_rejected=len(bad), ok=True)
       log.info("day=%s arrivals=%d in=%d out=%d rejected=%d", day, len(arrivals), len(rows), n, len(bad))
       return n
   
   
   if __name__ == "__main__":
       logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
       p = argparse.ArgumentParser()
       p.add_argument("--day", type=date.fromisoformat, required=True)
       a = p.parse_args()
       try:
           run(a.day)
       except Exception:
           log.exception("failed for %s", a.day)
           sys.exit(1)

   One subtlety worth understanding: bronze is partitioned by event date, but files arrive by arrival date. Rebuilding an event-date partition only from the three arrival days in the window would drop events for that date that arrived earlier. The merge step in the next phase makes this harmless (silver keeps every version by key), and the capstone module discusses the alternative of reading all raw days touching a partition. Keep the note in DESIGN.md.
2. 3.2
   Write the telemetry helper: one row per run into a DuckDB table the dashboard and the freshness check read.

   pythonCopy

   # pipelines/telemetry.py
   import os
   import time
   from datetime import date
   
   import duckdb
   
   DB = os.environ.get("WAREHOUSE_DB", "lake/warehouse.duckdb")
   
   
   def record_run(pipeline: str, table: str, logical_date: date, started: float, *, rows_in: int, rows_out: int, rows_rejected: int, ok: bool) -> None:
       con = duckdb.connect(DB)
       con.execute("CREATE SCHEMA IF NOT EXISTS ops")
       con.execute("""CREATE TABLE IF NOT EXISTS ops.pipeline_runs (
           pipeline VARCHAR, table_name VARCHAR, logical_date DATE, started_at TIMESTAMP, finished_at TIMESTAMP,
           status VARCHAR, rows_in BIGINT, rows_out BIGINT, rows_rejected BIGINT)""")
       con.execute("INSERT INTO ops.pipeline_runs VALUES (?, ?, ?, to_timestamp(?), to_timestamp(?), ?, ?, ?, ?)",
                   [pipeline, table, logical_date, started, time.time(), "success" if ok else "failed", rows_in, rows_out, rows_rejected])
       con.close()

3. 3.3
   Run the loader for one day, twice, and confirm the second run changes nothing. Then write that as the test.

   bashCopy

   export AWS_ACCESS_KEY_ID=lake AWS_SECRET_ACCESS_KEY=lake-secret-1 AWS_ENDPOINT_URL=http://localhost:9000
   pip install polars==1.19.0 pydantic==2.10.4 duckdb==1.1.3 boto3==1.35.90 pytest==8.3.4
   python -m pipelines.load_bronze --day 2026-03-04
   python -m pipelines.load_bronze --day 2026-03-04
   duckdb lake/warehouse.duckdb "SELECT logical_date, rows_in, rows_out, rows_rejected FROM ops.pipeline_runs ORDER BY started_at"

   Check: Two runs with identical rows_out; in MinIO the bronze/orders/event_date=2026-03-0{1,2,3,4} prefixes each have exactly one Parquet file (late events from those days were included by the lookback).
4. 3.4
   Add the idempotency and late-event tests against a temporary bucket prefix (or a second bucket) so they can run in CI against a MinIO service container.

   pythonCopy

   # tests/test_load_idempotent.py
   import os
   from datetime import date
   
   import duckdb
   
   from pipelines import load_bronze
   
   
   def bronze_stats(event_date: str) -> tuple:
       con = duckdb.connect()
       con.execute("INSTALL httpfs; LOAD httpfs; SET s3_endpoint='localhost:9000'; SET s3_use_ssl=false; SET s3_url_style='path';")
       con.execute(f"SET s3_access_key_id='{os.environ['AWS_ACCESS_KEY_ID']}'; SET s3_secret_access_key='{os.environ['AWS_SECRET_ACCESS_KEY']}';")
       return con.execute(f"SELECT COUNT(*), COUNT(DISTINCT order_id) FROM read_parquet('s3://lake/bronze/orders/event_date={event_date}/*.parquet')").fetchone()
   
   
   def test_second_run_changes_nothing():
       load_bronze.run(date(2026, 3, 4))
       first = bronze_stats("2026-03-04")
       load_bronze.run(date(2026, 3, 4))
       assert bronze_stats("2026-03-04") == first
   
   
   def test_late_event_lands_in_its_event_date_partition():
       # the generator makes ~5% of orders arrive 1-3 days late; those ordered on 03-01 but arriving 03-03
       # must be in the 03-01 partition after a run on 03-03 with a 3-day lookback
       load_bronze.run(date(2026, 3, 3))
       con = duckdb.connect()
       con.execute("INSTALL httpfs; LOAD httpfs; SET s3_endpoint='localhost:9000'; SET s3_use_ssl=false; SET s3_url_style='path';")
       con.execute(f"SET s3_access_key_id='{os.environ['AWS_ACCESS_KEY_ID']}'; SET s3_secret_access_key='{os.environ['AWS_SECRET_ACCESS_KEY']}';")
       late = con.execute("SELECT COUNT(*) FROM read_parquet('s3://lake/bronze/orders/event_date=2026-03-01/*.parquet') WHERE arrival_date > '2026-03-01'").fetchone()[0]
       assert late > 0

   Check: pytest -q tests/test_load_idempotent.py passes with the stack running.

Silver Iceberg tables that hold the current state of orders and customers, deduplicated by key with a version guard, deletes applied, history available through snapshots.

1. 4.1
   Create the Iceberg catalog (SQLite, file-based) and the silver tables with PyIceberg. Partition orders by day of ordered_at.

   pythonCopy

   # pipelines/catalog.py
   import os
   
   from pyiceberg.catalog.sql import SqlCatalog
   
   WAREHOUSE = os.environ.get("ICEBERG_WAREHOUSE", "s3://lake/silver")
   
   
   def catalog() -> SqlCatalog:
       return SqlCatalog(
           "lake",
           uri="sqlite:///lake/iceberg_catalog.db",
           warehouse=WAREHOUSE,
           **{
               "s3.endpoint": os.environ.get("AWS_ENDPOINT_URL", "http://localhost:9000"),
               "s3.access-key-id": os.environ["AWS_ACCESS_KEY_ID"],
               "s3.secret-access-key": os.environ["AWS_SECRET_ACCESS_KEY"],
               "s3.path-style-access": "true",
           },
       )

   A SQLite catalog is enough for one machine; the storage module lists REST, Glue and Nessie catalogs for teams.
2. 4.2
   Write the merge. PyIceberg 0.8 supports upsert on tables with an identifier field; implement the version guard by first reducing the batch to the latest event per key, then applying deletes and upserts. Keep last_version on the table so a replayed batch is a no-op.

   pythonCopy

   # pipelines/merge_silver.py
   import argparse
   import logging
   import os
   import sys
   import time
   from datetime import date, timedelta
   
   import duckdb
   import pyarrow as pa
   from pyiceberg.exceptions import NoSuchTableError
   from pyiceberg.expressions import In
   
   from pipelines.catalog import catalog
   from pipelines.telemetry import record_run
   
   log = logging.getLogger("merge_silver")
   
   SCHEMA_SQL = """
     SELECT order_id::BIGINT AS order_id, customer_id::BIGINT AS customer_id, ordered_at::TIMESTAMP AS ordered_at,
            status::VARCHAR AS status, shipped_at::TIMESTAMP AS shipped_at, lines::VARCHAR AS lines,
            updated_at::TIMESTAMP AS last_version, op::VARCHAR AS op
   """
   
   
   def latest_events(day: date, lookback: int) -> pa.Table:
       con = duckdb.connect()
       con.execute("INSTALL httpfs; LOAD httpfs; SET s3_url_style='path'; SET s3_use_ssl=false;")
       con.execute(f"SET s3_endpoint='{os.environ['AWS_ENDPOINT_URL'].replace('http://','')}';")
       con.execute(f"SET s3_access_key_id='{os.environ['AWS_ACCESS_KEY_ID']}'; SET s3_secret_access_key='{os.environ['AWS_SECRET_ACCESS_KEY']}';")
       days = ",".join(f"'s3://lake/bronze/orders/event_date={(day - timedelta(days=i)).isoformat()}/*.parquet'" for i in range(lookback))
       return con.execute(f"""
           WITH ev AS ({SCHEMA_SQL} FROM read_parquet([{days}], union_by_name=true)),
           ranked AS (SELECT *, ROW_NUMBER() OVER (PARTITION BY order_id ORDER BY last_version DESC) AS rn FROM ev)
           SELECT * EXCLUDE rn FROM ranked WHERE rn = 1
       """).arrow()
   
   
   def run(day: date, lookback: int = 3) -> int:
       started = time.time()
       cat = catalog()
       batch = latest_events(day, lookback)
       try:
           table = cat.load_table("silver.orders")
       except NoSuchTableError:
           cat.create_namespace_if_not_exists("silver")
           table = cat.create_table("silver.orders", schema=batch.drop_columns(["op"]).schema)
           table = cat.load_table("silver.orders")
   
       deletes = batch.filter(pa.compute.equal(batch["op"], "d"))
       upserts = batch.filter(pa.compute.not_equal(batch["op"], "d")).drop_columns(["op"])
   
       # version guard: only rows newer than what silver holds
       existing = table.scan(row_filter=In("order_id", upserts["order_id"].to_pylist())).to_arrow() if len(upserts) else None
       if existing is not None and len(existing):
           con = duckdb.connect()
           upserts = con.execute("""
               SELECT u.* FROM upserts u LEFT JOIN existing e USING (order_id)
               WHERE e.order_id IS NULL OR u.last_version > e.last_version
           """).arrow()
   
       if len(upserts):
           table.upsert(upserts, join_cols=["order_id"])
       if len(deletes):
           table.delete(In("order_id", deletes["order_id"].to_pylist()))
   
       record_run("merge_silver", "silver.orders", day, started, rows_in=len(batch), rows_out=len(upserts), rows_rejected=0, ok=True)
       log.info("day=%s batch=%d upserted=%d deleted=%d", day, len(batch), len(upserts), len(deletes))
       return len(upserts)
   
   
   if __name__ == "__main__":
       logging.basicConfig(level=logging.INFO)
       p = argparse.ArgumentParser(); p.add_argument("--day", type=date.fromisoformat, required=True)
       try:
           run(p.parse_args().day)
       except Exception:
           log.exception("merge failed"); sys.exit(1)

   If your PyIceberg version lacks upsert, implement it as delete-by-key followed by append, which is how the merge is expressed in older versions; the version guard is what keeps either form idempotent. Do the same for silver.customers with customer_id as the key.
3. 4.3
   Run the merge for the same day twice and for a later day, then prove the properties: no duplicates, deletes applied, a replay is a no-op, and time travel shows the deleted rows in an older snapshot.

   bashCopy

   python -m pipelines.merge_silver --day 2026-03-04
   python -m pipelines.merge_silver --day 2026-03-04
   python - <<'EOF'
   from pipelines.catalog import catalog
   t = catalog().load_table("silver.orders")
   df = t.scan().to_arrow().to_pandas()
   print("rows", len(df), "distinct keys", df.order_id.nunique())
   snaps = list(t.snapshots())
   print("snapshots", len(snaps))
   old = t.scan(snapshot_id=snaps[0].snapshot_id).to_arrow().num_rows
   print("rows in first snapshot", old)
   EOF

   Check: rows == distinct keys; the second run adds a snapshot with no data change (or none at all); the first snapshot's row count differs from the current one where deletes were applied.
4. 4.4
   Add the maintenance job: expire snapshots older than seven days and note that compaction (rewriting small files) is done through the engine of your choice; for this lab, a weekly rewrite via DuckDB export is acceptable. Schedule it later in the orchestrator.

   pythonCopy

   # pipelines/maintenance.py
   from datetime import datetime, timedelta, timezone
   
   from pipelines.catalog import catalog
   
   
   def expire(table_name: str, days: int = 7) -> None:
       t = catalog().load_table(table_name)
       cutoff = int((datetime.now(timezone.utc) - timedelta(days=days)).timestamp() * 1000)
       old = [s for s in t.snapshots() if s.timestamp_ms < cutoff and s.snapshot_id != t.current_snapshot().snapshot_id]
       print(f"{table_name}: {len(old)} snapshots older than {days} days")
       # PyIceberg exposes snapshot expiry through table maintenance in newer releases; if unavailable,
       # run the expiry with Spark/Trino/DuckDB's iceberg extension against the same catalog.
   
   
   if __name__ == "__main__":
       expire("silver.orders")
       expire("silver.customers")

   Record the retention decision in DESIGN.md: snapshots expire after 7 days, which also bounds how long a deleted customer's rows survive in time travel — the governance module's erasure requirement.

A dbt project on DuckDB reading the Iceberg tables: staging, an SCD2 snapshot of customers, dimensions, facts, the metric definitions, tests, documentation and an exposure for the dashboard.

1. 5.1
   Initialise the dbt project with the DuckDB adapter, pointing at lake/warehouse.duckdb, and configure DuckDB's Iceberg and S3 access in the profile so models can read silver directly.

   yamlCopy

   # dbt/profiles.yml
   lakehouse:
     target: dev
     outputs:
       dev:
         type: duckdb
         path: ../lake/warehouse.duckdb
         extensions: [httpfs, iceberg]
         settings:
           s3_endpoint: localhost:9000
           s3_use_ssl: false
           s3_url_style: path
           s3_access_key_id: lake
           s3_secret_access_key: lake-secret-1
         threads: 4

   Inside the Airflow container the endpoint is minio:9000; use an environment variable ({{ env_var('S3_ENDPOINT', 'localhost:9000') }}) in the profile so the same file works in both places.
2. 5.2
   Write the sources and staging models. DuckDB's iceberg_scan reads the table's current metadata; point it at the metadata file path from the catalog (PyIceberg prints it via table.metadata_location) or, simpler for this lab, export silver to Parquet after each merge and read that. Choose one and record it.

   sqlCopy

   -- dbt/models/staging/stg_orders.sql
   {{ config(materialized='view') }}
   with src as (
       select * from iceberg_scan('{{ var("orders_metadata") }}')
   )
   select
       order_id,
       customer_id,
       ordered_at,
       cast(ordered_at as date)              as order_date,
       status,
       shipped_at,
       lines,
       last_version                          as updated_at
   from src

   Pass the metadata location with dbt build --vars '{orders_metadata: "s3://lake/silver/silver.db/orders/metadata/v12.metadata.json"}'; the Airflow task reads it from the catalog and passes it in. The lines JSON column is exploded in an intermediate model into one row per line.
3. 5.3
   Build the intermediate line explosion, the SCD2 snapshot, the dimensions and the facts. The snapshot gives Q1 and Q7 their history.

   sqlCopy

   -- dbt/models/intermediate/int_order_lines.sql
   select
       o.order_id,
       o.customer_id,
       o.order_date,
       o.status,
       (l ->> 'line_no')::int             as line_no,
       (l ->> 'product_id')               as product_id,
       (l ->> 'quantity')::int            as quantity,
       (l ->> 'unit_price')::decimal(12,2) as unit_price
   from {{ ref('stg_orders') }} o,
        unnest(from_json(o.lines, '["json"]')) as t(l)
   where o.lines is not null
   
   -- dbt/snapshots/customers_snapshot.sql
   {% snapshot customers_snapshot %}
   {{ config(target_schema='snapshots', unique_key='customer_id', strategy='timestamp', updated_at='updated_at') }}
   select * from {{ ref('stg_customers') }}
   {% endsnapshot %}
   
   -- dbt/models/marts/dim_customers.sql
   select
       {{ dbt_utils.generate_surrogate_key(['customer_id', 'dbt_valid_from']) }} as customer_key,
       customer_id,
       md5(email)                                     as email_hash,      -- PII: hashed in marts
       country, segment, signup_date,
       date_trunc('month', signup_date)               as signup_month,
       dbt_valid_from                                 as valid_from,
       coalesce(dbt_valid_to, timestamp '9999-12-31') as valid_to,
       dbt_valid_to is null                           as is_current
   from {{ ref('customers_snapshot') }}
   
   -- dbt/models/marts/fct_order_lines.sql
   select
       l.order_id, l.line_no, l.order_date, l.product_id, l.quantity, l.unit_price,
       l.quantity * l.unit_price as line_amount,
       c.customer_key                                   -- version of the customer AS OF the order
   from {{ ref('int_order_lines') }} l
   left join {{ ref('dim_customers') }} c
     on c.customer_id = l.customer_id
    and l.order_date >= cast(c.valid_from as date) and l.order_date < cast(c.valid_to as date)

   fct_orders is the order-grain fact with status and timestamps (Q8 uses shipped_at - ordered_at). Define revenue, refund_rate and repeat_rate as their own models under marts/metrics/ exactly as DESIGN.md states them.
4. 5.4
   Add the tests and the exposure: column tests, the SCD2 invariants, a seasonal volume test, and the dashboard declared as an exposure so lineage reaches it.

   yamlCopy

   # dbt/models/marts/_marts.yml
   version: 2
   models:
     - name: fct_order_lines
       description: One row per order line. revenue = SUM(line_amount) for orders with status in ('paid','shipped').
       meta: {owner: you@example.com, product_tier: 1, sla: {freshness: "by 06:00 UTC"}}
       tests:
         - dbt_utils.unique_combination_of_columns: {combination_of_columns: [order_id, line_no]}
       columns:
         - name: customer_key
           tests: [not_null, {relationships: {to: ref('dim_customers'), field: customer_key}}]
         - name: line_amount
           tests: [{dbt_utils.accepted_range: {min_value: 0, inclusive: true}}]
     - name: dim_customers
       description: SCD2 customer dimension; one current row per customer_id; email hashed.
       columns:
         - name: customer_key
           tests: [unique, not_null]
       tests:
         - dbt_utils.expression_is_true:
             expression: "valid_from < valid_to"
   
   exposures:
     - name: orders_dashboard
       type: dashboard
       url: http://localhost:8501
       owner: {name: You, email: you@example.com}
       depends_on: [ref('fct_order_lines'), ref('fct_orders'), ref('dim_customers'), ref('dim_products')]

   Add a singular test tests/assert_one_current_row_per_customer.sql that returns customers with more than one is_current row, and the seasonal volume test from the quality module against fct_orders.
5. 5.5
   Build and test, then generate the docs and confirm the lineage graph runs from the silver sources through the snapshot to the dashboard exposure.

   bashCopy

   cd dbt && dbt deps && dbt snapshot && dbt build --vars "{orders_metadata: '$(python -c "from pipelines.catalog import catalog; print(catalog().load_table('silver.orders').metadata_location)")'}"
   dbt docs generate && dbt docs serve --port 8081

   Check: dbt build ends with all models built and all tests passing; the docs site shows the exposure at the right edge of the lineage graph. Store target/manifest.json as prod-manifest.json for slim CI later.

Two DAGs connected by a dataset with retries and a freshness SLA, governance tags with masked views and an erasure job, the dashboard with a trust panel, and the one-command demo.

1. 6.1
   Write the load DAG: a sensor for the raw prefix, then load, merge and telemetry, producing the silver dataset. Everything derives its date from the data interval.

   pythonCopy

   # dags/orders_load.py
   from datetime import datetime, timedelta
   
   from airflow.datasets import Dataset
   from airflow.decorators import dag, task
   from airflow.providers.amazon.aws.sensors.s3 import S3KeySensor
   
   SILVER_ORDERS = Dataset("s3://lake/silver/orders")
   
   
   @dag(dag_id="orders_load", schedule="0 2 * * *", start_date=datetime(2026, 1, 1), catchup=False, max_active_runs=1,
        default_args={"retries": 2, "retry_delay": timedelta(minutes=2), "execution_timeout": timedelta(minutes=20)})
   def orders_load():
       wait = S3KeySensor(task_id="wait_for_raw", bucket_name="lake", aws_conn_id="aws_default",
                          bucket_key="raw/orders/dt={{ data_interval_start | ds }}/batch-0001.jsonl.gz",
                          poke_interval=60, timeout=6 * 3600, mode="reschedule")
   
       @task
       def load(data_interval_start=None):
           from pipelines.load_bronze import run
           return run(data_interval_start.date())
   
       @task(outlets=[SILVER_ORDERS])
       def merge(data_interval_start=None):
           from pipelines.merge_silver import run
           return run(data_interval_start.date())
   
       wait >> load() >> merge()
   
   
   orders_load()

   Create the aws_default connection in Airflow pointing at MinIO (Admin → Connections: type Amazon Web Services, login lake, password lake-secret-1, extra {"endpoint_url": "http://minio:9000"}), or set it via AIRFLOW_CONN_AWS_DEFAULT in Compose.
2. 6.2
   Write the marts DAG scheduled on the dataset: it resolves the current Iceberg metadata location, runs dbt build, and checks freshness of the telemetry table. Then trigger the load DAG for a few dates and watch the marts DAG follow automatically; finally backfill a week.

   pythonCopy

   # dags/orders_marts.py
   from datetime import datetime, timedelta
   
   from airflow.datasets import Dataset
   from airflow.decorators import dag, task
   from airflow.operators.bash import BashOperator
   
   SILVER_ORDERS = Dataset("s3://lake/silver/orders")
   
   
   @dag(dag_id="orders_marts", schedule=[SILVER_ORDERS], start_date=datetime(2026, 1, 1), catchup=False,
        default_args={"retries": 1, "execution_timeout": timedelta(minutes=30)})
   def orders_marts():
       @task
       def metadata_location() -> str:
           from pipelines.catalog import catalog
           return catalog().load_table("silver.orders").metadata_location
   
       build = BashOperator(
           task_id="dbt_build",
           cwd="/opt/airflow/dbt",
           bash_command="dbt snapshot --profiles-dir . && dbt build --profiles-dir . --vars \"{orders_metadata: '{{ ti.xcom_pull(task_ids='metadata_location') }}'}\"",
           env={"S3_ENDPOINT": "minio:9000"},
       )
   
       @task
       def freshness_check():
           import duckdb
           con = duckdb.connect("/opt/airflow/lake/warehouse.duckdb", read_only=True)
           age_h = con.execute("SELECT date_diff('hour', max(finished_at), now()) FROM ops.pipeline_runs WHERE pipeline='merge_silver' AND status='success'").fetchone()[0]
           if age_h is None or age_h > 30:
               raise RuntimeError(f"silver.orders is stale: last success {age_h} h ago")
   
       metadata_location() >> build >> freshness_check()
   
   
   orders_marts()

   Check: After airflow dags backfill orders_load --start-date 2026-03-01 --end-date 2026-03-07 --max-active-runs 1 (inside the container), each load run's dataset update triggers an orders_marts run, and the marts reflect the week.
3. 6.3
   Governance: tag PII in the dbt YAML (meta: {pii: direct, mask: hash} on email; pii: indirect on dob), generate a masked view for an analyst role, and implement the erasure job for one customer across bronze history, silver and snapshots; prove with a query and record the run.

   sqlCopy

   -- governance/masked_views.sql (run against warehouse.duckdb)
   CREATE SCHEMA IF NOT EXISTS analyst;
   CREATE OR REPLACE VIEW analyst.dim_customers AS
   SELECT customer_key, customer_id, email_hash, country, segment,
          date_trunc('year', signup_date) AS signup_year,     -- coarsened, not raw
          valid_from, valid_to, is_current
   FROM main_marts.dim_customers;
   
   -- governance/erasure.sql: customer 123 asked to be forgotten
   CREATE TABLE IF NOT EXISTS ops.erasure_log (customer_id BIGINT, requested_at TIMESTAMP, completed_at TIMESTAMP, layers VARCHAR);
   -- silver + snapshots: handled by pipelines/erase.py using PyIceberg delete + snapshot expiry
   -- marts: rebuilt from silver on the next dbt run; force it now:
   DELETE FROM snapshots.customers_snapshot WHERE customer_id = 123;
   DELETE FROM main_marts.dim_customers WHERE customer_id = 123;
   UPDATE main_marts.fct_orders SET customer_id = NULL WHERE customer_id = 123;
   INSERT INTO ops.erasure_log VALUES (123, now() - interval 1 hour, now(), 'silver,snapshots,marts');

   Write pipelines/erase.py to delete the customer from silver.customers (PyIceberg delete) and to expire snapshots immediately for that table, then run dbt build so marts are rebuilt. Add record_of_processing.md listing datasets, fields, purpose, retention (raw 730 d, snapshots 7 d) and processors (none).
4. 6.4
   Build the dashboard: ten questions answered from the marts, plus a trust panel reading ops.pipeline_runs and the dbt run_results.json for last update, freshness status and test results.

   pythonCopy

   # dashboard/app.py
   import json
   from pathlib import Path
   
   import duckdb
   import streamlit as st
   
   con = duckdb.connect("/lake/warehouse.duckdb", read_only=True)
   st.set_page_config(page_title="Orders lakehouse", layout="wide")
   
   # --- trust panel ---
   last = con.execute("SELECT max(finished_at), date_diff('hour', max(finished_at), now()) FROM ops.pipeline_runs WHERE pipeline='merge_silver' AND status='success'").fetchone()
   results = json.loads(Path("/lake/dbt_run_results.json").read_text()) if Path("/lake/dbt_run_results.json").exists() else {"results": []}
   tests = [r for r in results["results"] if r["unique_id"].startswith("test.")]
   failed = [t for t in tests if t["status"] != "pass"]
   c1, c2, c3 = st.columns(3)
   c1.metric("Data as of", str(last[0])[:16] if last[0] else "never")
   c2.metric("Freshness", "OK" if last[1] is not None and last[1] <= 30 else "STALE", f"{last[1]} h ago" if last[1] is not None else "")
   c3.metric("dbt tests", f"{len(tests) - len(failed)} / {len(tests)} passed", delta=None if not failed else f"-{len(failed)} failing", delta_color="inverse")
   if failed or last[1] is None or last[1] > 30:
       st.warning("Numbers below may be stale or unverified — see the trust panel.")
   
   # --- the questions ---
   st.subheader("Q1 Daily revenue by customer country (country as of order time)")
   st.dataframe(con.execute("""
       SELECT l.order_date, c.country, ROUND(SUM(l.line_amount), 2) AS revenue
       FROM main_marts.fct_order_lines l JOIN main_marts.dim_customers c USING (customer_key)
       JOIN main_marts.fct_orders o USING (order_id)
       WHERE o.status IN ('paid','shipped') GROUP BY 1, 2 ORDER BY 1 DESC, 3 DESC LIMIT 200
   """).df())
   
   st.subheader("Q2 Orders in the last hour (silver, freshness 15 min target)")
   st.dataframe(con.execute("SELECT * FROM main_staging.stg_orders WHERE ordered_at >= now() - interval 1 hour ORDER BY ordered_at DESC").df())
   # ... Q3-Q10 follow the metric definitions in DESIGN.md

   Copy dbt/target/run_results.json to lake/dbt_run_results.json as the last step of the marts DAG so the dashboard can read it. For Q2 note in DESIGN.md why a 15-minute schedule of the load DAG (not streaming) satisfies "last hour" here, and what would change the decision.
5. 6.5
   Write the Makefile that runs the whole demo, then run it end to end on a clean checkout and time it. Commit everything, add a CI workflow (lint, the two tests against a MinIO service container, dbt slim build), and finish DESIGN.md with the evidence section: each guarantee pointing at its test, run record or file, plus residual risks.

   textCopy

   # Makefile
   .PHONY: demo up generate load merge marts dashboard clean
   export AWS_ACCESS_KEY_ID=lake AWS_SECRET_ACCESS_KEY=lake-secret-1 AWS_ENDPOINT_URL=http://localhost:9000
   
   up:        ; docker compose up -d && sleep 30
   generate:  ; python generate/make_events.py
   load:      ; for d in 2026-03-01 2026-03-02 2026-03-03 2026-03-04 2026-03-05 2026-03-06 2026-03-07; do python -m pipelines.load_bronze --day $$d; done
   merge:     ; for d in 2026-03-01 2026-03-02 2026-03-03 2026-03-04 2026-03-05 2026-03-06 2026-03-07; do python -m pipelines.merge_silver --day $$d; done
   marts:     ; cd dbt && dbt deps && dbt snapshot --profiles-dir . && dbt build --profiles-dir . --vars "{orders_metadata: '$$(python -c "from pipelines.catalog import catalog; print(catalog().load_table('silver.orders').metadata_location)")'}" && cp target/run_results.json ../lake/
   demo: up generate load merge marts
   	@echo "open http://localhost:8501 (dashboard), :8080 (airflow), :9001 (minio)"
   clean:     ; docker compose down -v && rm -rf lake/*

   Check: make clean && make demo completes in under thirty minutes on a laptop; the dashboard shows a green trust panel and answers the questions; pytest -q passes; the CI workflow is green on a pull request. Residual risks recorded: single-node catalog, no compaction job yet, raw layer not yet under a lifecycle rule in MinIO.

DuckDB cannot read from MinIO (HTTP 403 or connection refused)

Set s3_url_style='path', s3_use_ssl=false and the endpoint without http:// (localhost:9000 on the host, minio:9000 inside containers). The credentials are lake / lake-secret-1.

PyIceberg raises about a missing s3.endpoint or signature errors

The catalog properties must include s3.endpoint, the keys and s3.path-style-access=true (see pipelines/catalog.py); also export AWS_ENDPOINT_URL for s3fs. Inside Airflow the endpoint is http://minio:9000.

The Airflow container has import errors for the DAGs

_PIP_ADDITIONAL_REQUIREMENTS installs on every start and can take minutes; check docker compose logs airflow until pip finishes, then refresh. For a persistent setup build an image with the packages.

dbt build cannot find iceberg_scan or the extension

The profile must list extensions: [httpfs, iceberg]; DuckDB downloads them on first use, which needs network access. Confirm with duckdb -c "INSTALL iceberg; LOAD iceberg;".

The metadata location passed to dbt is stale, so marts miss the latest merge

The marts DAG's metadata_location task must run after the merge (it does, via the dataset trigger). For manual runs, re-read it from the catalog each time; do not hard-code a v12.metadata.json path.

The SCD2 snapshot has two current rows for a customer

The timestamp strategy needs a strictly increasing updated_at per key; the generator emits one, but if you added events by hand with equal timestamps the snapshot cannot order them. Fix the data, run the singular test, and see the quality module on SCD2 invariants.

The load test fails in CI but passes locally

CI needs the same MinIO service (a services: block in the workflow with the same credentials) and the generator run first; the tests assume raw data for early March exists.

• Add the Real-time pipeline with Kafka project: produce the same order events to a topic and land them with a streaming job, then compare the two paths for Q2.
• Move the catalog to a REST catalog and the warehouse to a cloud object store with the Data platform with IaC and CI project.
• Replace the DuckDB marts with a cloud warehouse reading the same Iceberg tables and measure query cost with the warehouse module's techniques.
• Put the SRE track's SLOs on the freshness metric and page when tier-1 marts are late.
• Run the governance module's mock data subject access request against your own pipeline and time it.