2. Ingestion Pipeline¶

Points: 10 — Provide a Python ingestion script that loads data from the source into storage. Batch-based, modular, readable, and documented.

Overview¶

Two layers move data into PostgreSQL:

Backfill (pipelines/backfill/) — one-shot loaders that pull 2 years of history. Run once at setup time, then on demand if you want to extend the window.
Daily refresh (pipelines/ingestion/weather.py) — pulls the next-few-hours precipitation forecast every morning.

Plus a daily call back to the backfill scripts to keep the current year and last week of weather in sync (Airflow runs them with date arguments).

flowchart TD
    SZ["Stadt Zürich MIV CSV<br/>(one file per year)"] --> BT["backfill_traffic.py"]
    OMA["Open-Meteo archive"] --> BW["backfill_weather.py"]
    OMF["Open-Meteo forecast"] --> WF["weather.py"]
    BT -->|dlt merge| PG1["traffic_raw.hourly_counts"]
    BW -->|dlt merge| PG2["weather_raw.hourly_history"]
    WF -->|dlt replace| PG3["weather_raw.hourly_forecast"]

All three scripts use dlt — same destination, same config (config.yaml), same connection string.

Traffic backfill¶

File: pipelines/backfill/backfill_traffic.py

Downloads the Stadt Zürich MIV hourly CSV for each year in the window (default: last 2 years) and merges rows into traffic_raw.hourly_counts.

A few things worth flagging:

The CSV comes gzipped on the wire and has a UTF-8 BOM. Streaming with iter_lines(decode_unicode=True) mis-decodes the gzipped bytes, so the script downloads to a NamedTemporaryFile and re-opens it with encoding="utf-8-sig".
Only rows with AnzFahrzeugeStatus="Gemessen" are kept — anything else is imputed by the city and we don't want it in the average.
Coordinates are LV95 (Swiss). Conversion to WGS84 happens in the recommender, not here, so the raw table keeps the original numbers.
The locally-orchestrated runs use write_disposition="merge" on (msid, ts). The cloud run is append (BigQuery rejects the staging schema when dlt tries to add _dlt_load_id as REQUIRED on a merged table), with the dataset wiped before each backfill to keep things idempotent.

What lands in traffic_raw.hourly_counts:

Column	Notes
`msid`	measuring section id (one counter, one direction)
`zsid`	counter station id
`zsname`, `achse`, `richtung`	counter name, street, direction
`ekoord`, `nkoord`	LV95 east/north
`ts`	hour the count belongs to (Europe/Zurich)
`vehicle_count`	vehicles in that hour
`ingested_at`	UTC batch timestamp

Run:

docker compose run --rm dev uv run python pipelines/backfill/backfill_traffic.py
docker compose run --rm dev uv run python pipelines/backfill/backfill_traffic.py --years 2024 2025

Row counts after the 2-year backfill:

2024 -> 1,665,527
2025 -> 1,734,772
2026 ->   729,963 (year to date)
counters -> 215-216

Weather backfill¶

File: pipelines/backfill/backfill_weather.py

Pulls 2 years of hourly weather from the Open-Meteo archive endpoint at four bounding-box corners around the city (defined in config.yaml). Each row is keyed by (latitude, longitude, ts). Same merge-vs-append story as traffic: merge locally, append in the cloud.

visibility is in the forecast but not in the archive, so the column list drops it. Everything else lines up.

Row count: 4 points × 731 days × 24h = 70,176.

Run:

docker compose run --rm dev uv run python pipelines/backfill/backfill_weather.py
docker compose run --rm dev uv run python pipelines/backfill/backfill_weather.py \
    --start 2024-01-01 --end 2024-12-31

Weather forecast (daily)¶

File: pipelines/ingestion/weather.py

Pulls the next 7 days of hourly forecast and writes it to weather_raw.hourly_forecast with write_disposition="replace". The recommender hits the Open-Meteo forecast endpoint directly at request time, so this table is mainly there for inspection and future "what did we think was going to happen?" queries.

Batch design¶

Property	How
Idempotent	merge on natural keys (`(msid, ts)`, `(lat, lon, ts)`); replace for forecast
Independent	each script writes to its own schema, no cross-references
Resumable	a failed year can be re-run on its own with `--years` / `--start` / `--end`

How to verify¶

$ docker compose exec -T pgdatabase psql -U root -d rush -c \
    "select count(*) from traffic_raw.hourly_counts;"
  count
---------
 4130262

$ docker compose exec -T pgdatabase psql -U root -d rush -c \
    "select count(distinct msid) from traffic_raw.hourly_counts;"
 count
-------
   216

$ docker compose exec -T pgdatabase psql -U root -d rush -c \
    "select count(*) from weather_raw.hourly_history;"
 count
-------
 70176

Cloud ingestion (GCS + BigQuery)¶

The cloud target runs the same scripts inside the rush-ingest Cloud Run Job. The image (Dockerfile.ingest) sets RUSH_TARGET=bigquery, which switches config.make_destination() to the BigQuery dlt destination. dlt picks up the attached service account through Application Default Credentials, so no key file is shipped in the image. See Orchestration: Cloud target.