Pipeline Orchestration with Prefect
Workflow scheduling, dependency management, and error handling.
What you'll learn
- 1Days 51–53Flow and task decorators, local orchestration, and state management.
By the end: Students rebuild
pipeline.shas a **Prefect flow**. Each step is an observable **task**, related steps are grouped in **subflows**, and independent API calls run **concurrently**. Every run is visible in the Prefect UI. The fragile exit-code juggling of the shell script becomes an explicit, tested **state policy**, and the pipeline reports through **state hooks** (alerts) and **artifacts** (a run report economists can read). - 2Days 54–57Retries, caching, scheduling, and logging.
By the end:
daily-refreshruns **by itself** on a schedule aligned with Bank of Canada and Statistics Canada publication times. Transient failures are **retried** (only those that time can fix), unchanged work is **skipped** through caching and data fingerprints, and no two runs ever fight over the DuckDB lock. Every run leaves structured **logs** and a queryable **run history**, and a **watchdog** flow alerts when the refresh silently stops happening. - 3Days 58–60Orchestrating the daily macro-dashboard refresh: dlt → DuckDB → dbt → GX.
By the end: The macro dashboard pipeline from the course notebook is fully automated and production-ready: **dlt** extraction runs as first-class Prefect tasks, with per-source retries. **Statistics Canada releases** trigger Write–Audit–Publish within minutes of publication. **dbt** rebuilds only what the new data affects, with model-level results in the UI. **Great Expectations** gates publication. The notebook's dashboards are regenerated as HTML, with "as of" dates and data-quality banners.
Tools
The ten days
Before you start: module setup and project structure
Audience: MSc Economics students who completed Modules 1–5 (Days 1–50)
Environment: GitHub Codespaces · .venv (dlt, dbt, now Prefect) · .venv-gx (Great Expectations) · macro.duckdb
Time per day: ~2.5 hours (including breaks and exercises)
Goal by end of Day 53: Students rebuild pipeline.sh as a Prefect flow. Each step is an observable task, related steps are grouped in subflows, and independent API calls run concurrently. Every run is visible in the Prefect UI. The fragile exit-code juggling of the shell script becomes an explicit, tested state policy, and the pipeline reports through state hooks (alerts) and artifacts (a run report economists can read).
Why orchestration, and why now?
At the end of Module 5 you listed what pipeline.sh can't do:
pipeline.sh today | What an orchestrator gives you |
|---|---|
| runs only when someone types it | schedules and event triggers (Days 54–57) |
| one failure → rerun everything | per-step retries and caching (Days 54–57) |
| "did last night's run work?" → read logs | a UI with every run, step, duration and log line |
| ` | |
| no record of why a run ended as it did | state messages, hooks, artifacts |
| accidental overlapping runs fight over the DuckDB lock | concurrency controls (Days 54–57) |
Prefect is a Python orchestration framework. You add two decorators (@flow and @task) to ordinary Python functions, and Prefect tracks, logs, retries and schedules them. There's no special DSL and no rewrite: your Module 1–5 code stays as it is.
Orchestration ≠ transformation. Prefect doesn't replace dlt, dbt or GX. It coordinates them: in what order, under which conditions, with which retries, and what to do when something goes wrong. Think of it as the conductor, not another instrument.
The plan for Module 6
| Days | Focus |
|---|---|
| 51–53 | flows & tasks, local orchestration (subflows, concurrency, wrapping existing tools), states |
| 54–57 | retries, caching, scheduling, logging and deployments |
| 58–60 | orchestrating the daily macro-dashboard refresh: dlt → WAP → dbt → GX, triggered by releases |
Target structure (by end of Day 53)
macro-data-course/
├── orchestration/ ← NEW
│ ├── __init__.py
│ ├── settings.py ← paths, interpreters, environment
│ ├── tasks/
│ │ ├── __init__.py
│ │ ├── commands.py ← run_step(): scripts/CLIs as observable tasks
│ │ └── reporting.py ← artifacts: run report, KPI briefing
│ └── flows/
│ ├── __init__.py
│ ├── hello.py ← Day 51 learning flows
│ ├── cpi_briefing.py
│ ├── concurrency_demo.py ← Day 52
│ └── daily_refresh.py ← Days 52–53: pipeline.sh as a flow
├── pipeline.sh ← kept for comparison until Day 60