Lab 01 — Mini Pipeline Orchestrator

Phase: 08 — Training Orchestration & MLOps | Difficulty: ⭐⭐⭐⭐☆ | Time: 4–6 hours

Airflow, Dagster, Kubeflow, DVC, and dbt are the same five mechanisms in different YAML. Build them once — DAG execution, content-addressed caching, retries, ledger, lineage — and you'll configure any of them by mechanism instead of superstition.

What you build

  • Pipeline.topo_order — Kahn's algorithm with cycle detection that names the cycle
  • cache_key — H(version ‖ params ‖ upstream artifact hashes): the formula from which incremental recomputation falls out with zero invalidation logic
  • run — topological execution with cache hits, including early cutoff (re-run task produces identical bytes → downstream stays cached)
  • _run_with_retries — bounded retries with a transient/permanent classifier and the original exception surfaced on give-up
  • Ledger + lineage — immutable execution records; lineage() (provenance: what made this model?) and downstream_of() (taint: this input was corrupted — what's affected?)

Key concepts

ConceptWhat to understand
Content vs timestampsHashes are ground truth; timestamps lie under re-runs and clock skew
Invalidation propagationParam change → key change → re-run → output-hash change → downstream keys change. No invalidation code exists anywhere
Early cutoffIdentical output bytes → unchanged artifact hash → downstream cache keys unchanged → skip (test_version_bump demonstrates it)
The version termCache keys without code identity serve stale artifacts after every refactor — the #1 homemade-cache bug
Transient vs permanentRetry timeouts; fail fast on ValueError — retrying bugs hides them in noise

Run

pytest test_lab.py -v                       # your lab.py
LAB_MODULE=solution pytest test_lab.py -v   # reference (14 tests)
python solution.py                          # caching demo: run, re-run, param change

Suggested TODO order

  1. add_task + topo_order — get the DAG tests green
  2. run without caching — make the execution-counter fixture show all tasks running
  3. cache_key + the cache path in run — hit/miss, propagation, early-cutoff tests
  4. _run_with_retries — the Flaky fixture drives all three retry tests
  5. latest_record / lineage / downstream_of

Success criteria

  • All 14 tests pass
  • You can explain early cutoff and why it's correct (and name the build systems that do it)
  • You can state what breaks if tasks are impure (undeclared inputs) or nondeterministic (unseeded RNG) — and therefore why those disciplines are caching requirements
  • You can map each mechanism to Airflow/MLflow/DVC equivalents

Extensions

  • Parallel execution of independent tasks (ThreadPoolExecutor over each topo "level")
  • Persistent cache + ledger (SQLite) surviving process restarts
  • Stage-then-commit artifact writes (atomic rename) and a kill-mid-task crash test
  • --force task and --downstream-of task re-run flags (the operational verbs)

Interview Q&A

Q: How does your cache know to re-run downstream tasks when an upstream changes? It doesn't "know" — the knowledge is structural. Downstream keys include upstream output hashes; an upstream change that alters its output alters those hashes, which alters downstream keys, which miss the cache. And if the upstream re-runs but produces identical bytes, downstream keys are unchanged and correctly skip (early cutoff). All invalidation behavior derives from the key formula.

Q: A task succeeded but wrote a corrupt artifact, and three downstream models shipped before anyone noticed. What do you need? The taint query: downstream_of(task) over the lineage ledger gives the affected closure; the ledger's artifact hashes identify exactly which model versions consumed the corrupt hash; the registry (Phase 08 WARMUP Ch. 7) maps those to deployments. Then prevention: a data-contract gate (Phase 01) between that task and its consumers, so corruption fails the pipeline instead of flowing.

References