Day 52: Local Orchestration · Applied Decision Intelligence & Data Architecture

Written lesson

Time	Block
0:00 – 0:20	52.1 Dependencies: data vs `wait_for`
0:20 – 0:50	52.2 Concurrency: `submit`, `map` and futures
0:50 – 0:58	Break
0:58 – 1:18	52.3 Subflows: structuring a large workflow
1:18 – 1:45	52.4 Wrapping existing tools as tasks
1:45 – 1:53	Break
1:53 – 2:05	52.5 Inspecting runs from the command line
2:05 – 2:30	52.6 Mini-project: `daily_refresh` v1

52.1 Dependencies: Data vs `wait_for` (20 min)

Prefect infers dependencies from data: if a task receives another task's output, it runs after it. But some steps depend on each other without passing data:

dbt must run after the StatCan tables are published, but it doesn't take their output as a Python argument. It reads the warehouse.

For these side-effect dependencies, use wait_for:

from prefect import flow, task


@task
def publish_statcan_tables():
    print("publishing raw tables…")          # writes to DuckDB; returns nothing useful


@task
def dbt_build():
    print("running dbt…")                    # reads DuckDB


@flow(log_prints=True)
def ordering_demo():
    published = publish_statcan_tables.submit()
    dbt_build.submit(wait_for=[published]).wait()


if __name__ == "__main__":
    ordering_demo()

(.submit() and futures are explained in the next block. For now, read wait_for=[published] as "don't start before published has finished".)

Dependency type	Expressed by	Example
data	passing a task's output as input	`yoy = year_over_year(levels)`
side effect	`wait_for=[...]`	dbt after publish; GX after dbt

Most of our pipeline steps communicate through the warehouse, not through Python values. That's normal for data platforms, and it's why wait_for will appear often.

52.2 Concurrency: `submit`, `map` and Futures (30 min)

Calling a task directly (fetch(x)) runs it and waits for the result. That's sequential. Many of our steps are independent API calls that spend most of their time waiting on the network. Running them concurrently saves time.

Call style	Behaviour	Returns
`my_task(x)`	run now, wait	the result
`my_task.submit(x)`	start in the background, continue immediately	a future
`my_task.map([x1, x2, x3])`	submit once per item	a list of futures

A future is a placeholder for a result that isn't ready yet. Call .result() to wait for it and get the value, or .wait() to just wait.

Create orchestration/flows/concurrency_demo.py:

import time

import requests
from prefect import flow, get_run_logger, task
from prefect.task_runners import ThreadPoolTaskRunner

SERIES = ["V39079", "FXCADUSD", "BD.CDN.2YR.DQ.YLD", "BD.CDN.5YR.DQ.YLD", "BD.CDN.10YR.DQ.YLD"]


@task
def fetch_valet(series_id: str, recent: int = 250) -> int:
    r = requests.get(f"https://www.bankofcanada.ca/valet/observations/{series_id}/json",
                     params={"recent": recent}, timeout=30)
    r.raise_for_status()
    n = len(r.json().get("observations", []))
    get_run_logger().info(f"{series_id}: {n} observations")
    return n


@flow
def sequential(series: list[str] = SERIES) -> int:
    return sum(fetch_valet(s) for s in series)


@flow(task_runner=ThreadPoolTaskRunner(max_workers=5))
def concurrent(series: list[str] = SERIES) -> int:
    futures = fetch_valet.map(series)            # five task runs start together
    return sum(futures.result())                 # wait for all, collect results


if __name__ == "__main__":
    for f in (sequential, concurrent):
        t0 = time.perf_counter()
        total = f()
        print(f"{f.name:<12} {total} observations in {time.perf_counter() - t0:.1f}s")

python -m orchestration.flows.concurrency_demo

Compare the timings, and in the UI compare the two flows' timelines: bars one after another vs bars overlapping.

ThreadPoolTaskRunner(max_workers=5) decides how submitted tasks run: here, up to five at once in threads. Threads suit I/O-bound work (waiting on networks or disks).
futures.result() on a list of futures waits for all of them and returns their results in order.

When concurrency hurts

Good candidates	Bad candidates
independent API calls to different providers	many calls to one provider (rate limits, Day 29)
independent read-only queries	anything that writes to DuckDB (one writer at a time!)
steps that wait on the network	CPU-heavy pandas work (threads don't speed it up)

For our pipeline: BoC, FRED and StatCan fetches could overlap, but every load into macro.duckdb must be sequential. That's why Day 50's design keeps writing, auditing and publishing strictly in order. Days 54–57 add concurrency limits so that "only one DuckDB writer at a time" is enforced by Prefect rather than by discipline.

Always resolve your futures. If a flow submits tasks and never calls .result() or .wait(), it may finish before they do, and their failures may go unnoticed. Make waiting explicit.

Try it 52.2

Change max_workers to 2 and re-run concurrent. Predict the duration first. Then think about a polite max_workers for 30 series from one provider, and why.

Answer

With 2 workers, the five calls run in three waves (2 + 2 + 1), so the time is roughly three single-call durations. For 30 series from one provider, a small number (2–4) plus the throttle from Day 29 is polite. The bottleneck we must respect is the provider's capacity, not our CPU.

52.3 Subflows: Structuring a Large Workflow (20 min)

A subflow is simply a flow called from inside another flow. It appears in the UI as a nested run, with its own tasks, logs and state.

from prefect import flow, task


@task
def step(name: str):
    print(f"running {name}")


@flow(log_prints=True)
def ingest():
    step("ingest API series")
    step("WAP StatCan tables")


@flow(log_prints=True)
def transform():
    step("dbt build")


@flow(log_prints=True)
def quality():
    step("GX warehouse checkpoints")


@flow(name="daily-refresh-skeleton", log_prints=True)
def daily_refresh_skeleton():
    ingest()
    transform()
    quality()


if __name__ == "__main__":
    daily_refresh_skeleton()

In the UI, the parent run contains three child flow runs, each expandable.

Use subflows when:

a group of tasks forms a meaningful stage (ingest, transform, quality);
the stage should be runnable on its own (e.g. re-run only quality after fixing a threshold);
you want separate state and logs per stage.

Structure	Use for
task	one step: one API, one tool call
subflow	one stage: several related tasks, independently runnable
parent flow	the end-to-end process

This mirrors the architecture you've built: extract/load (Module 3), transform (Module 4), quality (Module 5).

52.4 Wrapping Existing Tools as Tasks (28 min)

Our pipeline steps already exist as scripts and CLIs, each with a meaningful exit code (Days 10, 30, 46, 50):

Step	Command	Environment	Exit codes
API ingestion	`python ingest.py --only boc,fred,statcan`	`.venv`	0 ok · 2 partial · 1 failed
StatCan tables (WAP)	`python ingest_statcan_tables.py`	`.venv` (+ GX subprocess)	0 ok · 2 warnings/quarantine · 1 all quarantined
Transform	`dbt build --target prod`	`.venv`	0 ok · non-zero failed
Warehouse quality	`quality/run_quality.sh`	`.venv-gx`	0 ok · 2 warnings · 1 critical

Should we call the Python code directly, or run the scripts?

Call Python functions in-process	Run as a subprocess
richer integration (return values, exceptions)	works for any tool and any environment (GX in `.venv-gx`, shell scripts)
shares one process, and one DuckDB lock	each step has its own process, so locks are released when it exits
needs all libraries in one environment	each step keeps its own environment

Given Day 50's lesson (separate processes release DuckDB cleanly) and the separate GX environment, we wrap each pipeline step as a subprocess task. It's reliable, and it reuses the exit-code conventions you already designed.

Central settings

Create orchestration/settings.py:

"""Paths and interpreters used by the orchestration layer."""
import os
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]
VENV_PY = sys.executable                                   # .venv (Prefect, dlt, dbt)
VENV_BIN = Path(sys.executable).parent
GX_PY = str(REPO / ".venv-gx" / "bin" / "python")
DBT = str(VENV_BIN / "dbt")
DBT_DIR = REPO / "macro_dbt"

ENV = {
    **os.environ,
    "MACRO_DUCKDB_PATH": os.environ.get("MACRO_DUCKDB_PATH", str(REPO / "macro.duckdb")),
    "MACRO_DUCKDB_URL": os.environ.get("MACRO_DUCKDB_URL", f"duckdb:///{REPO / 'macro.duckdb'}"),
}

sys.executable is the Python interpreter currently running, the .venv one. So ingest.py runs with exactly the libraries Prefect sees.

The generic step task

Create orchestration/tasks/commands.py:

"""Run pipeline steps (scripts, CLIs) as observable Prefect tasks."""
import subprocess
import time

from prefect import get_run_logger, task

from orchestration.settings import ENV, REPO


@task(tags=["pipeline-step"])
def run_step(name: str, cmd: list[str], cwd: str | None = None) -> dict:
    """Run one command; stream its output into Prefect logs; return exit code and duration."""
    logger = get_run_logger()
    logger.info(f"▶ {name}: {' '.join(cmd)}")
    t0 = time.perf_counter()

    proc = subprocess.Popen(cmd, cwd=cwd or REPO, env=ENV, text=True,
                            stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
    for line in proc.stdout:                           # stream output line by line
        logger.info(f"[{name}] {line.rstrip()}")
    code = proc.wait()

    duration = round(time.perf_counter() - t0, 1)
    logger.info(f"■ {name}: exit code {code} after {duration}s")
    return {"step": name, "exit_code": code, "duration_s": duration}

subprocess.Popen + iterating proc.stdout streams output as it happens, so a long dbt build shows progress in the UI instead of one burst at the end.
stderr=subprocess.STDOUT merges error output into the same stream.
The task doesn't decide what an exit code means. It reports it. Interpretation belongs to the flow's state policy (Day 53).

52.5 Inspecting Runs from the Command Line (13 min)

The UI is convenient, but the CLI is scriptable and works over SSH:

prefect flow-run ls --limit 5                 # recent runs with state and duration
prefect flow-run inspect <flow-run-id>        # full details of one run
prefect flow-run logs <flow-run-id>           # its log lines

(Copy an ID from ls or from the UI.) Each run records its parameters, start/end times, state, state message and task runs, the audit trail from 51.1.

52.6 Mini-Project: `daily_refresh` v1 (25 min)

Rebuild pipeline.sh as a flow with subflows per stage, using run_step. Today, keep pipeline.sh's behaviour: tolerate exit code 2 from ingestion; stop on anything else. Tomorrow we replace this with an explicit state policy.

Create orchestration/flows/daily_refresh.py:

"""Daily macro refresh — v1: pipeline.sh as a Prefect flow (Day 52)."""
from prefect import flow, get_run_logger

from orchestration.settings import DBT, DBT_DIR, GX_PY, VENV_PY
from orchestration.tasks.commands import run_step


def require(result: dict, allowed=(0,)):
    """Mimic pipeline.sh: raise unless the exit code is allowed."""
    if result["exit_code"] not in allowed:
        raise RuntimeError(f"{result['step']} failed with exit code {result['exit_code']}")
    return result


@flow(name="ingest")
def ingest_flow() -> list[dict]:
    api = require(run_step("ingest_api", [VENV_PY, "ingest.py", "--only", "boc,fred,statcan"]),
                  allowed=(0, 2))
    wap = require(run_step("wap_statcan_tables", [VENV_PY, "ingest_statcan_tables.py"]),
                  allowed=(0, 2))
    return [api, wap]


@flow(name="transform")
def transform_flow() -> list[dict]:
    deps = require(run_step("dbt_deps", [DBT, "deps", "--quiet"], cwd=str(DBT_DIR)))
    build = require(run_step("dbt_build", [DBT, "build", "--target", "prod"], cwd=str(DBT_DIR)))
    docs = require(run_step("dbt_docs", [DBT, "docs", "generate", "--target", "prod"],
                            cwd=str(DBT_DIR)))
    return [deps, build, docs]


@flow(name="quality")
def quality_flow() -> list[dict]:
    return [require(run_step("gx_warehouse", ["bash", "quality/run_quality.sh"]),
                    allowed=(0, 2))]


@flow(name="daily-refresh")
def daily_refresh() -> list[dict]:
    logger = get_run_logger()
    results = ingest_flow() + transform_flow() + quality_flow()
    for r in results:
        logger.info(f"{r['step']:<22} exit {r['exit_code']}  {r['duration_s']:>7.1f}s")
    return results


if __name__ == "__main__":
    daily_refresh()

GX_PY isn't used yet: run_quality.sh activates .venv-gx itself. We'll use it on Day 53 for finer-grained quality steps.

Run it (the full refresh takes a while: dlt, the WAP audit and dbt all run for real):

python -m orchestration.flows.daily_refresh

In the UI, open the daily-refresh run: three subflows, eight step tasks, a timeline showing where the time goes, and every line of every tool's output in the logs. Compare that with scrolling through pipeline.sh output in a terminal.

Try it 52.6

From the timeline, which step dominates the runtime? Is that what you expected?
The API ingestion (BoC, FRED, StatCan vectors) is one step. Could it be split into three concurrent steps? What would stop you? (Hint: where does each provider's data get written?)

Discussion

Usually the StatCan full-table download (when a table was republished) or dbt build dominates. On days with no new StatCan release, WAP is fast thanks to change detection (Day 30).
Fetching could overlap, but ingest.py also writes each source into macro.duckdb, and DuckDB allows one writer. Splitting safely would mean fetching concurrently but loading sequentially, a change to the ingestion code, not just the flow. That's a good example of the orchestrator exposing an architectural constraint.

Day 52 checklist

I distinguish data dependencies from side-effect dependencies (wait_for)
I can run tasks concurrently with submit/map, resolve futures, and know when concurrency is unsafe (DuckDB writes)
I structure large workflows with subflows per stage
Existing scripts and CLIs run as streamed, observable subprocess tasks with exit codes
daily_refresh v1 reproduces pipeline.sh, with full visibility in the UI

git add orchestration && git commit -m "Day 52: concurrency, subflows, daily_refresh v1" && git push

Local Orchestration

What you'll learn today

Today's plan