5. Workflow Orchestration

Points: 10 — Implement a workflow orchestration tool that schedules the batch pipeline, supports backfills, and can trigger future runs.

---

Overview

Apache Airflow 2.10.5 with LocalExecutor in Docker Compose. Two local DAGs and (still around for now) two cloud DAGs.

Local DAGs — dags/rush_pipeline.py

flowchart TD
    subgraph BD["rush_backfill (manual)"]
        BT["01_backfill_traffic"]
        BW["02_backfill_weather"]
        TR["03_trigger_daily"]
        BT --> BW --> TR
    end

    subgraph DD["rush_daily (02:00 UTC)"]
        F["10_ingest_weather_forecast"]
        WH["20_ingest_weather_history<br/>(last 7 days)"]
        TH["30_ingest_traffic_history<br/>(current year)"]
        BM["40_build_marts (dbt build)"]
        F --> BM
        WH --> BM
        TH --> BM
    end

    TR -.->|TriggerDagRunOperator| DD

---

DAG 1: rush_backfill

One-shot manual trigger. Run it once after setup to load 2 years of history, or any time you want to extend the window.

Property	Value
DAG ID	rush_backfill
Schedule	None
Tags	rush, backfill

Tasks:

task_id	runs
01_backfill_traffic	pipelines/backfill/backfill_traffic.py
02_backfill_weather	pipelines/backfill/backfill_weather.py
03_trigger_daily	triggers rush_daily once on success

Sequential on purpose — both scripts are heavy and parallel runs would just fight for the dlt staging slot.

---

DAG 2: rush_daily

Daily refresh. Pulls the next-few-hours forecast, fills in the last 7 days of weather (in case Open-Meteo updates the archive), refreshes the current year of traffic, then runs dbt build.

Property	Value
DAG ID	rush_daily
Schedule	0 2 * * * (02:00 UTC)
Tags	rush, daily

Tasks:

task_id	runs
10_ingest_weather_forecast	pipelines/ingestion/weather.py
20_ingest_weather_history	backfill_weather.py --start ...-7d --end today
30_ingest_traffic_history	backfill_traffic.py --years $(date +%Y)
40_build_marts	dbt build

The three ingests run in parallel. 40_build_marts waits for all three.

Why 02:00 UTC? That's about 03:00 / 04:00 Zurich time — late enough for the Stadt Zürich daily CSV to be updated, early enough that the marts are fresh before anyone uses the recommender in the morning.

---

Backfills

Built into the design: trigger rush_backfill manually for any window. The two scripts accept --years and --start/--end so you can re-do a single month or stretch back another year.

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

Both are idempotent (dlt merge on natural keys).

---

Defaults

default_args = {
    "owner": "rush",
    "retries": 1,
    "retry_delay": timedelta(minutes=5),
}

LocalExecutor, dedicated PostgreSQL 16 metadata DB. DAGs folder mounted from ./dags.

---

How to verify

$ docker compose exec airflow-scheduler airflow dags list | grep rush
rush_backfill | /opt/airflow/dags/rush_pipeline.py | rush
rush_daily    | /opt/airflow/dags/rush_pipeline.py | rush

Trigger from the UI (http://localhost:8080) or:

$ docker compose exec airflow-scheduler airflow dags trigger rush_backfill

After it finishes, rush_daily is triggered automatically.

---

Cloud target

The cloud side runs without Airflow. Two Cloud Run Jobs and two Cloud Scheduler triggers do the same work as the local DAGs:

Job	Image	Schedule	Does
rush-ingest	Dockerfile.ingest	0 2 * * * UTC	weather forecast + 7-day weather history + current-year traffic, all into BigQuery via dlt
rush-dbt	Dockerfile.dbt	0 3 * * * UTC	dbt build --target prod against the rush dataset

Both jobs run as the rush-runner service account. Scheduler invokes them through roles/run.invoker. Images live in an Artifact Registry repo created by Terraform (terraform/cloud_run.tf).

The scripts that the jobs run come from the same Python package as the local DAGs (pipelines/backfill/*, pipelines/ingestion/weather.py). The only difference is the env var RUSH_TARGET=bigquery baked into the ingest image, which switches the dlt destination through config.make_destination().

Push images and apply the stack:

$ gcloud auth configure-docker europe-west6-docker.pkg.dev
$ docker build -f Dockerfile.ingest -t europe-west6-docker.pkg.dev/$PROJECT/rush/ingest:latest .
$ docker build -f Dockerfile.dbt    -t europe-west6-docker.pkg.dev/$PROJECT/rush/dbt:latest .
$ docker build -f Dockerfile.ui     -t europe-west6-docker.pkg.dev/$PROJECT/rush/ui:latest .
$ docker push europe-west6-docker.pkg.dev/$PROJECT/rush/ingest:latest
$ docker push europe-west6-docker.pkg.dev/$PROJECT/rush/dbt:latest
$ docker push europe-west6-docker.pkg.dev/$PROJECT/rush/ui:latest
$ cd terraform && terraform apply

UI

The Streamlit app (pipelines/ui/app.py) is a Cloud Run Service (rush-ui), built from Dockerfile.ui. It shares the runner service account, so the recommender's BigQuery lookups go through the metadata server (no key file in the image).

Run it locally:

$ docker compose run --rm -p 8501:8501 dev uv run streamlit run pipelines/ui/app.py

Set ui_public = false in terraform.tfvars if you'd rather keep it behind IAM.