Capstone: build a lakehouse pipeline

The brief and the deliverables

Pick a domain with events and entities: an online shop (orders, customers, products, page views), a ride service (trips, drivers, riders), a SaaS product (accounts, subscriptions, usage events). Write ten business questions first, in plain words, at least three of which need history ("revenue by customer country as it was at order time") and at least one of which needs recent data ("orders in the last hour"). The questions drive the model, the model drives the pipeline, and the pipeline drives the platform — never the other way around. Generate realistic synthetic data with a script you commit: a few million events over a year, late arrivals, corrections, deletes, and a slowly changing customer table.

Phase 1: model before you move anything

From the questions, declare the facts and their grains (one row per order line; one row per page view), the dimensions and which attributes are Type 2 (customer country, plan), the conformed date dimension, and the one definition of each headline metric. Draw the layer diagram: raw (immutable files by arrival date), bronze (typed Parquet by event date), silver (table-format entities and events, deduplicated, with deletes applied), gold (dbt marts), and where each of the ten questions is answered. Write all of it into DESIGN.md before writing a line of pipeline code; the document is what you will check the implementation against.

QUESTIONS (10)                       ANSWERED BY
  Q1 daily revenue by country        fct_orders x dim_customer (SCD2, country as of order)
  Q2 orders in the last hour         silver.orders (freshness SLA 15 min) via a direct query
  Q3 repeat-purchase rate by cohort  fct_orders + dim_customer.signup_month
  ...

FACTS                                GRAIN                      MEASURES
  fct_orders                         one row per order line     quantity, unit_price, line_amount
  fct_page_views                     one row per page view      duration_ms

DIMENSIONS                           SCD                        NOTES
  dim_customer                       type 2: country, segment   surrogate customer_key
  dim_product                        type 1                     category, brand
  dim_date                           n/a                        generated 2020-2030

METRICS (one definition each)
  revenue = SUM(line_amount) FROM fct_orders WHERE status IN ('paid','shipped')

LAYERS
  raw/     jsonl.gz by arrival date, immutable, 730-day retention
  bronze/  parquet by event_date, typed, quarantine table for rejects
  silver/  iceberg: orders, order_items, customers (deletes applied, last_lsn guard)
  gold/    dbt marts, tested, documented, tier-1 SLA 06:00 UTC

LATENESS: events up to 3 days late -> 3-day trailing reprocess; later -> manual backfill
PII: customer.email (direct, hash), customer.dob (indirect, year only), addresses (direct, restricted)

Phase 2: raw to silver, idempotently

Build the loader as the Python and batch modules described: a --day parameter, an incremental extract with a persisted watermark and overlap, pydantic validation with a quarantine table and a reject-rate threshold, Parquet output to bronze partitioned by event date, and a MERGE into silver Iceberg tables keyed on the natural key with a version guard, including deletes. Run it twice for the same day and prove the row counts do not change. Then simulate three kinds of trouble — a crash mid-load, a late batch for a day already processed, and a corrected record — and prove each is handled by the trailing-window reprocess and the merge.

from datetime import date

import duckdb

from pipelines.orders import run


def test_rerun_is_idempotent(tmp_path, fixture_events):
    lake = tmp_path / "lake"
    run(day=date(2026, 3, 1), lake=lake, source=fixture_events)
    first = duckdb.sql(f"SELECT COUNT(*), SUM(amount) FROM read_parquet('{lake}/bronze/orders/**/*.parquet')").fetchone()

    run(day=date(2026, 3, 1), lake=lake, source=fixture_events)   # same day again
    second = duckdb.sql(f"SELECT COUNT(*), SUM(amount) FROM read_parquet('{lake}/bronze/orders/**/*.parquet')").fetchone()

    assert first == second, f"rerun changed the data: {first} -> {second}"


def test_late_event_lands_in_its_own_partition(tmp_path, fixture_events_with_late):
    lake = tmp_path / "lake"
    run(day=date(2026, 3, 4), lake=lake, source=fixture_events_with_late, lookback_days=3)
    rows = duckdb.sql(
        f"SELECT event_date, COUNT(*) FROM read_parquet('{lake}/bronze/orders/**/*.parquet') GROUP BY 1 ORDER BY 1"
    ).fetchall()
    assert (date(2026, 3, 2), 1) in rows          # the late March 2nd event is in the March 2nd partition

Add compaction and snapshot expiry as a weekly maintenance task, and a reconciliation query that compares silver row counts and a checksum against the synthetic source for a window. Record every run in the telemetry table with rows in, out and rejected.

Phase 3: marts, tests and schedule

Build the dbt project over silver: staging models per silver table, an SCD2 snapshot of customers (or use the CDC-driven history), the dimensions and facts from the design, and the metric models. Add the standard column tests, the SCD2 invariants (one current row per key, no overlapping validity), a seasonal volume test and a relationships test between facts and dimensions. Document grain and metric definitions on the marts and declare the dashboard as an exposure. Then orchestrate: the loader DAG produces a dataset; the dbt DAG is scheduled on it; both have retries, timeouts, an on-failure callback and a freshness SLA. Backfill one week through the orchestrator and verify the marts.

models:
  - name: fct_orders
    description: One row per order line, non-cancelled. revenue = SUM(line_amount) WHERE status IN ('paid','shipped').
    meta:
      owner: data-platform@acme.example
      product_tier: 1
      classification: internal
      sla: {freshness: "by 06:00 UTC daily", completeness: "±30% same-weekday baseline"}
    columns:
      - name: order_id
        tests: [not_null]
      - name: line_no
        tests: [not_null]
      - name: customer_key
        tests:
          - not_null
          - relationships: {to: ref('dim_customer'), field: customer_key}
      - name: line_amount
        tests:
          - dbt_utils.accepted_range: {min_value: 0, inclusive: true}
    tests:
      - dbt_utils.unique_combination_of_columns: {combination_of_columns: [order_id, line_no]}
      - dbt_utils.recency: {datepart: hour, field: loaded_at, interval: 8}

exposures:
  - name: revenue_dashboard
    type: dashboard
    owner: {name: Finance analytics, email: finance-analytics@acme.example}
    depends_on: [ref('fct_orders'), ref('dim_customer'), ref('dim_date')]

Phase 4: governance and the recent-data question

Tag every PII column, generate the masking policies (or masked views in DuckDB), implement row-level access for one attribute, and write the record of processing. Implement the retention job for raw files and the erasure job for one synthetic customer across silver, gold and snapshots, and prove with a query that the person is gone. Add the governance CI check: no model without an owner, no PII without a mask.

Then answer the question that needs recent data. Choose deliberately between a fifteen-minute micro-batch of the loader (simplest, usually enough) and a small streaming path — events to a topic, a windowed aggregation, a freshness SLA in minutes. Justify the choice in the design document in terms of the consumer's need and the operational cost. If you build the stream, make its output idempotent and its watermark explicit, and show a comparison against the batch result for the same window.

Phase 5: platform, evidence and the write-up

Put the infrastructure in Terraform with a policy check in the plan, the repository under the CI workflow from the platform module, and production behind an approval. Build the dashboard that answers the ten questions from the marts, with a trust panel showing last update, SLA status and test results. Finally write the evidence section of DESIGN.md: for each claim (idempotent, handles late data, tested, governed, observable, deployed by CI) point at the test, the run record, the policy or the workflow that proves it, and include the reconciliation results and the backfill log. Add a residual-risks section: what is not handled, what you would do next with more time.

• A reader with no context can run make demo (or the documented commands) and reproduce the pipeline on the synthetic data in under thirty minutes.
• Every business question is answered by a named mart with a documented grain and metric definition.
• Rerun, crash, late data, correction and delete scenarios each have a test or a recorded drill.
• Freshness alerts fire when the schedule is skipped; the dashboard trust panel reflects it.
• The erasure drill removes a customer from every layer including snapshots, with a log entry.
• CI blocks: lint failures, failing tests, DAG parse errors, a public bucket, an untagged PII column.

When it is done you will have built and documented the full loop that the track kept returning to: model from questions, land raw immutably, transform idempotently, test what must be true, orchestrate on data readiness, watch freshness and lineage, govern the personal data, and keep all of it as reviewed, deployable code. That loop is the job, and the repository is the proof.