Project 9: ETL Pipeline & Warehouse

The scenario

Spokes runs 9 docking stations and a few hundred bikes. Every night each station uploads the day's trips. You are the company's first data engineer, and the operations lead has two questions: "How many trips do we do per day, smoothed over a week?" and "Which routes are busiest, so we can rebalance bikes?" Before you can answer, you have to survive the data:

• Files alternate between CSV with naive local timestamps (2026-03-05 08:15:00, meaning America/New_York) and JSON Lines with UTC offsets (2026-03-06T08:15:00-05:00).
• Every file has an exact duplicate row, an unparseable timestamp, a trip to an unknown station (S99), and a trip whose end is before its start.
• From day two, each file carries two late-arriving trips from the previous day; from day three, it also re-delivers a trip already sent.
• On 2026-03-08 a new station snapshot arrives: Harbor Point is renamed Harbor Point Plaza, University Ave gets more docks, and Ferry Terminal opens. The same night, clocks spring forward.

The deterministic generator (etl/generator.py, complete in the starter) writes seven days of this: 255 raw trip rows and two station snapshots. A correct pipeline loads exactly 222 trips, rejects 21 rows with reasons, removes 12 duplicates, and keeps 11 station versions, whether it runs once or seven times.

Learning goals

Feature specification

Architecture

flowchart LR
  subgraph raw[raw/ files]
    T[trips_*.csv / *.jsonl]
    S[stations_*.csv]
  end
  subgraph dag[DAG runner: graphlib + retries]
    direction LR
    st[stations] --> ex[extract] --> va[validate] --> tr[transform] --> lo[load] --> su[summary]
  end
  S --> st
  T --> ex
  va -- rejected rows --> R[(state/rejected.jsonl)]
  lo -- after commit --> M[(state/manifest.json)]
  st --> W
  lo --> W[(warehouse.db)]
  su --> RS[RunSummary]
    

erDiagram
  DIM_DATE ||--o{ FACT_TRIP : date_key
  DIM_STATION ||--o{ FACT_TRIP : start_station_key
  DIM_STATION ||--o{ FACT_TRIP : end_station_key
  FACT_TRIP {
    text trip_id PK
    int date_key FK
    int start_station_key FK
    int end_station_key FK
    text start_utc
    text end_utc
    int duration_s
    text bike_id
    text member_type
  }
  DIM_STATION {
    int station_key PK
    text station_id
    text name
    int capacity
    text valid_from
    text valid_to
    int is_current
  }
  DIM_DATE {
    int date_key PK
    text date
    int year
    int month
    int day
    text weekday
    int is_weekend
  }
  ETL_STATE {
    text key PK
    text value
  }
    

The grain of fact_trip is one row per trip, keyed by the natural key trip_id. Facts reference the station version in effect on the trip's local date, so a trip on 2026-03-07 shows "Harbor Point" forever, while one on 2026-03-08 shows "Harbor Point Plaza".

Getting started

pixi run projects                    # list projects
pixi run start p09                   # copy the starter to work/p09_etl_pipeline/
pixi run grade p09                   # run all 63 tests against your work/ copy
pixi run practicum grade p09 -k extract    # one milestone (extra arguments go to pytest)
cd work/p09_etl_pipeline
pixi run python -m etl.pipeline raw warehouse.db state --generate   # once everything passes

The starter's generator.py and models.py are complete; everything else has its real signature and docstring, a body of raise NotImplementedError, and a # TODO describing the approach. Some imports were trimmed, so add what you need as you go. Explore the data first:

pixi run python -c "from pathlib import Path; from etl.generator import generate; generate(Path('raw'))"
head -5 raw/trips_2026-03-05.csv raw/trips_2026-03-06.jsonl raw/stations_2026-03-08.csv

Milestones

INSERT INTO fact_trip (trip_id, date_key, ...) VALUES (?, ?, ...)
ON CONFLICT (trip_id) DO UPDATE SET
    date_key = excluded.date_key,
    duration_s = excluded.duration_s, ...

Rubric

Stretch goals

• Quarantine table: also load rejected rows into rejected_trip(source, reasons_json, row_json) so analysts can query them.
• Allowed lateness: reject trips older than the watermark minus two days with a too_late reason, and test the boundary.
• Data contract: write the raw trip contract as a JSON schema, version it, and fail the run on an incompatible change.
• DuckDB export: pixi add duckdb, then attach the SQLite warehouse (INSTALL sqlite; ATTACH 'warehouse.db' (TYPE sqlite)) and write fact_trip to Parquet partitioned by year=/month=; re-run the analytics in DuckDB.
• Lineage: emit an OpenLineage-shaped JSON event per DAG task with inputs, outputs, and row counts.
• Orchestrator: wrap run_pipeline in a Prefect flow or Dagster assets with a daily schedule and a backfill command.

Reflection quiz

When you're done, carry these skills into the Capstone, where the pipeline feeds a real application.