Orchestration with Prefect

9.1 Why orchestration? Flows and tasks (40 min)

Concept. An orchestrator runs the pipeline's steps in order, on a schedule, retries what fails for temporary reasons, stops what fails for real reasons, and keeps a record of every run. Prefect does this with two decorators:

• @task: one step (call the API, run dbt…). Prefect tracks its state: running, completed, failed, retrying.
• @flow: a function that calls tasks. Running a flow creates a flow run with its own log.

Your code stays ordinary Python; the decorators add the bookkeeping.

Example 1. Make sure the dbt panel exists (dbt build on Day 8), then create day09/hello_flow.py:

# day09/hello_flow.py
import duckdb
from prefect import flow, task


@task
def count_orders(week_start: str) -> int:
    con = duckdb.connect("data/northmart.duckdb", read_only=True)
    n = con.sql(f"""
        SELECT COALESCE(SUM(n_orders), 0) FROM marts.panel_customer_week
        WHERE week_start = DATE '{week_start}'
    """).fetchall()[0][0]
    con.close()
    return int(n)


@task
def describe(week_start: str, n: int) -> str:
    return f"Week of {week_start}: {n:,} orders"


@flow(log_prints=True)
def weekly_summary(week_start: str = "2026-03-02"):
    n = count_orders(week_start)
    message = describe(week_start, n)
    print(message)
    return message


if __name__ == "__main__":
    weekly_summary()
    weekly_summary("2026-02-23")

python day09/hello_flow.py

INFO | prefect - Starting temporary server on http://127.0.0.1:8013
INFO | Flow run 'uptight-ocelot' - Beginning flow run 'uptight-ocelot' for flow 'weekly-summary'
INFO | Task run 'count_orders-09f' - Finished in state Completed()
INFO | Task run 'describe-91a' - Finished in state Completed()
INFO | Flow run 'uptight-ocelot' - Week of 2026-03-02: 1,130 orders
INFO | Flow run 'uptight-ocelot' - Finished in state Completed()
...
INFO | Flow run 'remarkable-butterfly' - Week of 2026-02-23: 1,070 orders

Notes:

• Every run gets a random name (uptight-ocelot) and a log. Flow arguments (week_start) are parameters, recorded with the run.
• log_prints=True sends print() to the Prefect log.
• With no Prefect server configured, Prefect starts a temporary one for the run and stops it afterwards. That's fine for today's first sections; in 9.4 we start a permanent one.

Optional: prefect config set PREFECT_SERVER_ANALYTICS_ENABLED=false turns off Prefect's anonymous usage statistics.

Try it. Write day09/weeks_flow.py: a flow that takes a list of weeks, and a task that returns both orders and revenue for one week. Print the four weeks around the email (16 Feb to 9 Mar).

# day09/weeks_flow.py
import duckdb
from prefect import flow, task


@task
def week_totals(week_start: str) -> dict:
    con = duckdb.connect("data/northmart.duckdb", read_only=True)
    n, rev = con.sql(f"""
        SELECT SUM(n_orders), SUM(revenue) FROM marts.panel_customer_week
        WHERE week_start = DATE '{week_start}'
    """).fetchall()[0]
    con.close()
    return {"week": week_start, "orders": int(n), "revenue": round(rev)}


@flow(log_prints=True)
def around_the_email(weeks: list[str]):
    for w in weeks:
        r = week_totals(w)
        print(f"{r['week']}: {r['orders']:,} orders, ${r['revenue']:,}")


if __name__ == "__main__":
    around_the_email(["2026-02-16", "2026-02-23", "2026-03-02", "2026-03-09"])

2026-02-16: 1,178 orders, $179,326
2026-02-23: 1,070 orders, $167,578
2026-03-02: 1,130 orders, $180,172
2026-03-09: 1,086 orders, $173,571

9.2 Retries, logging and failure (40 min)

Concept. Some failures are temporary: a busy API, a dropped connection. Retrying after a pause usually fixes them. Others are real: a broken contract, a wrong password. Retrying those wastes time and hides the problem. Prefect lets you choose per task: @task(retries=3, retry_delay_seconds=2).

Create day09/retries_demo.py. The task fails twice on purpose, like an API under load:

# day09/retries_demo.py
from prefect import flow, task
from prefect.runtime import task_run


@task(retries=3, retry_delay_seconds=2)
def call_flaky_service() -> str:
    attempt = task_run.run_count
    print(f"attempt {attempt}")
    if attempt < 3:                       # fails twice, then works: like a busy API
        raise ConnectionError("service temporarily unavailable")
    return "data received"


@flow(log_prints=True)
def resilient():
    print(call_flaky_service())


if __name__ == "__main__":
    resilient()

Task run 'call_flaky_service-4e6' - attempt 1
Task run 'call_flaky_service-4e6' - Task run failed with exception: ConnectionError('service temporarily unavailable') - Retry 1/3 will start 2 second(s) from now
Task run 'call_flaky_service-4e6' - attempt 2
... Retry 2/3 will start 2 second(s) from now
Task run 'call_flaky_service-4e6' - attempt 3
Task run 'call_flaky_service-4e6' - Finished in state Completed()
Flow run 'mauve-hornet' - data received

Try it. Change the condition to if attempt < 9: so the service never recovers within the retries. What is the final state of the task and the flow?

Task run 'call_flaky_service-ed2' - ... Retry 3/3 will start 2 second(s) from now
Task run 'call_flaky_service-ed2' - attempt 4
Task run 'call_flaky_service-ed2' - Finished in state Failed('Task run encountered an exception ConnectionError: service temporarily unavailable')
Flow run 'speedy-crane' - Finished in state Failed(...)

9.3 The NorthMart flow: load, build, gate, publish (85 min)

Concept. The whole Module 1 pipeline becomes one flow with four tasks:

check_api ──► load_raw (dlt) ──► build_models (dbt build = models + contracts) ──► publish
  retries 3      retries 2              no retries: a failed contract is real               Parquet + manifest

The key design choice is the gate. dbt writes tables into DuckDB as it goes, so even a failed build can leave a damaged table there (Day 8, 8.5). Analysts therefore never read the warehouse directly. They read data/published/, and the publish task only runs if build_models succeeded. A failed run leaves the last good release untouched.

The tasks call our existing scripts with subprocess.run, the same commands you typed by hand. An orchestrator doesn't replace your code; it runs it reliably.

Create the folder pipeline/ and pipeline/northmart_flow.py:

# pipeline/northmart_flow.py
"""
The whole Module 1 pipeline as one Prefect flow:

    API health -> dlt load (raw) -> dbt build (models + contracts) -> publish marts

If any dbt contract fails, the flow stops BEFORE publishing, so analysts keep
the last good version of the data.

Run from the repository root (with the NorthMart API running):
    python pipeline/northmart_flow.py
    python pipeline/northmart_flow.py --as-of 2026-03-08     # replay a past week
"""
import argparse
import json
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path

import duckdb
import requests
from prefect import flow, get_run_logger, task

ROOT = Path(__file__).resolve().parent.parent
DB = ROOT / "data" / "northmart.duckdb"
PUBLISHED = ROOT / "data" / "published"
API = "http://127.0.0.1:8000"
MARTS = ["panel_customer_week", "customer_features"]


def run(cmd: list[str], cwd: Path = ROOT) -> str:
    """Run a command; raise with its output if it fails."""
    result = subprocess.run(cmd, cwd=cwd, capture_output=True, text=True)
    if result.returncode != 0:
        raise RuntimeError(f"{' '.join(cmd)} failed:\n{result.stdout[-3000:]}\n{result.stderr[-3000:]}")
    return result.stdout


@task(retries=3, retry_delay_seconds=5)
def check_api() -> str:
    r = requests.get(f"{API}/health", timeout=10)
    r.raise_for_status()
    today = r.json()["today"]
    get_run_logger().info(f"API is up; its 'today' is {today}")
    return today


@task
def set_clock(as_of: str) -> None:
    run([sys.executable, "api/clock.py", as_of])


@task(retries=2, retry_delay_seconds=10)
def load_raw() -> str:
    out = run([sys.executable, "day05/northmart_pipeline.py"])
    lines = [l for l in out.splitlines() if l.startswith("raw.")]
    for line in lines:
        get_run_logger().info(line)
    return "\n".join(lines)


@task
def build_models() -> str:
    """dbt build = models + tests + contracts. Any error raises and stops the flow."""
    logger = get_run_logger()
    result = subprocess.run(["dbt", "--no-use-colors", "build", "--profiles-dir", "."],
                            cwd=ROOT / "transform", capture_output=True, text=True)
    lines = [l.split("  ", 1)[-1].strip() for l in result.stdout.splitlines()]   # drop timestamps
    summary = next((l for l in lines if l.startswith("Done. PASS=")), "no summary")
    for line in lines:
        if " FAIL " in line or " ERROR " in line or " WARN " in line:
            logger.warning(line)
    logger.info(summary)
    if result.returncode != 0:
        raise RuntimeError(f"dbt build failed, data NOT published. {summary}")
    return summary


@task
def publish(as_of: str, dbt_summary: str) -> dict:
    """Export the marts to Parquet with a small manifest describing this release."""
    PUBLISHED.mkdir(parents=True, exist_ok=True)
    con = duckdb.connect(str(DB), read_only=True)
    manifest = {
        "published_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "data_as_of": as_of,
        "dbt": dbt_summary,
        "tables": {},
    }
    for name in MARTS:
        path = PUBLISHED / f"{name}.parquet"
        con.sql(f"COPY marts.{name} TO '{path}' (FORMAT parquet)")
        n = con.sql(f"SELECT COUNT(*) FROM marts.{name}").fetchall()[0][0]
        manifest["tables"][name] = {"rows": n, "file": str(path.relative_to(ROOT))}
    con.close()
    (PUBLISHED / "manifest.json").write_text(json.dumps(manifest, indent=2))
    get_run_logger().info(f"Published {len(MARTS)} tables to {PUBLISHED.relative_to(ROOT)}")
    return manifest


@flow(name="northmart-causal-data-mart", log_prints=True)
def northmart_pipeline(as_of: str | None = None) -> dict:
    if as_of:
        set_clock(as_of)
    today = check_api()
    load_raw()
    summary = build_models()
    manifest = publish(today, summary)
    print(f"Data as of {today}: " + ", ".join(f"{k} {v['rows']:,} rows" for k, v in manifest["tables"].items()))
    return manifest


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--as-of", help="move the API clock to this date first (YYYY-MM-DD)")
    args = parser.parse_args()
    northmart_pipeline(args.as_of)

With the API running in the other terminal:

python pipeline/northmart_flow.py

Task run 'check_api-3c4' - API is up; its 'today' is 2026-04-26
Task run 'load_raw-1ea' - raw.customers              4,000 rows
Task run 'load_raw-1ea' - raw.products                  40 rows
Task run 'load_raw-1ea' - raw.campaign_recipients    1,279 rows
Task run 'load_raw-1ea' - raw.orders                18,359 rows
Task run 'build_models-e4d' - 24 of 32 WARN 1 warn_overlap_by_region ...
Task run 'build_models-e4d' - Done. PASS=31 WARN=1 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=32
Task run 'publish-638' - Published 2 tables to data/published
Flow run 'cuddly-hoatzin' - Data as of 2026-04-26: panel_customer_week 64,000 rows, customer_features 4,000 rows
Flow run 'cuddly-hoatzin' - Finished in state Completed()

cat data/published/manifest.json

{
  "published_at": "2026-10-31T23:06:03+00:00",
  "data_as_of": "2026-04-26",
  "dbt": "Done. PASS=31 WARN=1 ERROR=0 SKIP=0 NO-OP=0 REUSED=0 TOTAL=32",
  "tables": {
    "panel_customer_week": {
      "rows": 64000,
      "file": "data/published/panel_customer_week.parquet"
    },
    "customer_features": {
      "rows": 4000,
      "file": "data/published/customer_features.parquet"
    }
  }
}

The manifest travels with the data: anyone who opens the Parquet files can see what date the data runs to and that every contract passed. Add data/published/ to .gitignore (it's already there if you copied the Day 1 version).

dbt --no-use-colors removes the colour codes dbt prints for terminals, which would otherwise clutter the Prefect log and the manifest.

9.4 Schedules and the Prefect dashboard (45 min)

Concept. A pipeline that only runs when someone remembers is not a pipeline. Prefect's serve() registers the flow as a deployment with a schedule and waits for scheduled times. Schedules need a permanent Prefect server (the temporary one can't schedule), so today you use three terminals.

Terminal 2 (already running): the NorthMart API.

Terminal 3 — the Prefect server:

source .venv/bin/activate
prefect server start

Open port 4200 when Codespaces offers it: that's the Prefect dashboard.

Codespaces tip. If the dashboard loads but says it can't reach the API, stop the server and start it like this instead: PREFECT_UI_API_URL="https://${CODESPACE_NAME}-4200.${GITHUB_CODESPACES_PORT_FORWARDING_DOMAIN}/api" prefect server start

Terminal 1 — point Prefect at the server, then serve the flow:

prefect config set PREFECT_API_URL=http://127.0.0.1:4200/api

Create pipeline/serve.py:

# pipeline/serve.py
"""Run the pipeline on a schedule. Leave this running; stop with Ctrl+C."""
from northmart_flow import northmart_pipeline

if __name__ == "__main__":
    northmart_pipeline.serve(
        name="daily-6am",
        cron="0 6 * * *",                 # every day at 06:00
        tags=["northmart", "module-1"],
    )

python pipeline/serve.py

Your flow 'northmart-causal-data-mart' is being served and polling for scheduled runs!

To trigger a run for this flow, use the following command:

        $ prefect deployment run 'northmart-causal-data-mart/daily-6am'

cron="0 6 * * *" means minute 0, hour 6, every day. You don't have to wait until 6 a.m.: open a fourth terminal (or use the dashboard's Run button) and trigger a run now:

prefect deployment run 'northmart-causal-data-mart/daily-6am'

Created flow run 'funny-tarsier'.
└── URL: http://127.0.0.1:4200/runs/flow-run/8e4096fd-...

In the dashboard, open Runs: you'll see the run, each task's state and duration, and the full log. Click Deployments → daily-6am to see the next scheduled runs.

Cleaning up. Ctrl+C stops serve.py and the server. To go back to temporary servers for single runs: prefect config unset PREFECT_API_URL. In production the server runs permanently on a shared machine or in Prefect Cloud; the flow code doesn't change.

9.5 Day 9 mini-project: a failed run that protects analysts (45 min)

Goal. Prove the gate works: make the data wrong in a way the loader can't fix, run the flow, and confirm nothing is published.

Step 1. Save the current manifest to compare later:

cp data/published/manifest.json /tmp/manifest_before.json

Step 2 — inject a missed load. Yesterday's backdate fault would be repaired by the next load (recipients use replace). A missing week is different: the incremental loader only asks for new orders, so the gap stays.

python day08/break_things.py drop_week
python pipeline/northmart_flow.py
echo "exit code: $?"

Task run 'load_raw-8e7' - raw.orders                17,121 rows
Task run 'build_models-a47' - 27 of 32 FAIL 1 assert_no_empty_weeks ....
Task run 'build_models-a47' - Done. PASS=30 WARN=1 ERROR=1 SKIP=0 NO-OP=0 REUSED=0 TOTAL=32
Task run 'build_models-a47' - Task run failed with exception: RuntimeError('dbt build failed, data NOT published. ...')
Flow run 'mega-rat' - Finished in state Failed(...)
exit code: 1

Step 3 — confirm the gate held.

cmp /tmp/manifest_before.json data/published/manifest.json && echo "nothing published"

nothing published

Analysts still have the last good release. The failed run is visible in the dashboard and returns exit code 1, which any alerting system can pick up.

Step 4 — recover. Running the flow again won't help: the loader thinks those orders are already loaded. A missed incremental load needs a full refresh:

python day08/break_things.py restore
python pipeline/northmart_flow.py

The run completes and publishes again. Write down this recovery procedure in docs/CONTRACTS.md under a heading Runbook: in a real team, whoever is on call at 6 a.m. needs it.

9.6 Save your work with Git (15 min)

git add .
git commit -m "Day 9: Prefect flow with contract gate, schedule and runbook"
git push

Day 9 checklist

• I can write Prefect flows and tasks with parameters and logging
• I can choose which tasks retry and explain why contract failures must not
• I can run the whole pipeline as one flow that publishes only after every contract passes
• I can serve a flow on a schedule, trigger it, and read runs in the dashboard
• I can recover from a missed incremental load with a full refresh