Every pipeline is a DAG. Here's what that buys you.
Airflow, Dagster, dbt, Spark's planner, Make, Bazel — they all run the same three algorithms from the graph chapter of your algorithms book. Topological sort decides what can run. The longest path tells you how long the pipeline will take. Cycle detection is why the scheduler refuses your graph before it runs. Then, because no graph post is complete without it, the minimum spanning tree.
Everything below runs on a real pipeline: the one that would feed the shift investigator from raw events. Durations are illustrative; the structure is the point.
The pipeline
Twelve tasks. Raw events land (bronze), get cleaned and typed (silver), get aggregated into the tables the agent queries (gold), and a few quality checks gate the publish step. Each edge is "must finish before." The scheduler's whole job is to respect those edges while running as much as possible in parallel.
1 · Topological sort: what can run now
Kahn's algorithm (CLRS §22.4, Kleinberg–Tardos §3.6): count each task's incoming edges, start with every task whose count is zero, run them, decrement their downstream counts, and repeat. The tasks that hit zero together are a layer — they can run in parallel. That's the scheduler's main loop, and it's about twelve lines.
The sort is O(V + E): each task and each dependency is touched once. The ordering within a layer is arbitrary — which is exactly why two runs of the same DAG can execute tasks in different orders, and why every task needs to be correct regardless of what else ran beside it. That's the first place idempotency shows up.
2 · Critical path: how long it will take
If you have unlimited workers, the pipeline's wall time isn't the sum of task durations — it's the longest path through the DAG, counting duration. Longest path is NP-hard on general graphs, but on a DAG it's a single pass in topological order (CLRS §24.2 flips the sign and runs the shortest-path DAG algorithm): for each task, its earliest finish is its duration plus the latest earliest-finish of its predecessors.
The highlighted chain is the critical path. Speeding up any task off it changes nothing. Speeding up a task on it shortens the pipeline — until a different path becomes critical. This is PERT from the 1950s, it's the same recurrence as rod cutting, and it's the first thing to compute when someone asks why the nightly run takes four hours.
3 · Cycle detection: why the scheduler says no
A dependency cycle means a task is waiting on something that is, transitively, waiting on it. Nothing ever runs. Every scheduler checks for this when it loads the graph, with the depth-first search coloring scheme from CLRS §22.3: white = unvisited, grey = on the current DFS path, black = finished. An edge into a grey node is a back edge, and a back edge is a cycle.
Click the bad edge on and off. A DAG is precisely a directed graph with no back edges under DFS — that's the definition and the test in one. It's also why "DAG" is in the name: Airflow raises on load if DFS finds grey.
What this means for the pipeline you're running
- Retries re-run a sub-DAG. When a task fails, the scheduler reruns it and everything reachable from it — a DFS from the failed node. Everything upstream stays done. This only works if rerunning a task is safe, which is why idempotent writes (overwrite-by-partition,
MERGEon a key) aren't optional: the graph structure assumes them. - Lineage is reverse reachability. "Which raw sources does this gold table depend on?" is DFS on the reversed graph. Data catalogs that show lineage are drawing that traversal.
- Backfills are the DAG times a date range. Rerunning March is the same DAG instantiated 31 times, with the date as a parameter. If tasks write to date-partitioned locations, the 31 instances don't collide. If they append, you get March twice.
- Spark is a DAG under your DataFrame.
df.explain()shows it: the lazy transformations become a DAG of stages split at shuffles, and the scheduler runs independent stages in parallel. Narrow transformations stay in one stage; wide ones (groupBy, join) cut the graph. Reading that plan is reading a DAG. - Fan-in is where the bugs live. A task with several parents runs when the last one finishes. If one parent is late, the task silently reads a stale version of the others unless each parent's output is versioned. Watch the nodes with in-degree ≥ 2 in the animation above; those are the ones to instrument.
A pipeline as code
Here's the same DAG as plain Python — no framework — with the three algorithms. This is what a scheduler does, minus the retries, logging, and UI. Run it and it prints the layers and the critical path from the chart.
from collections import defaultdict, deque
TASKS = { # name: (duration_min, [upstream])
"land_events": (8, []),
"land_orders": (5, []),
"land_downtime": (4, []),
"clean_events": (20, ["land_events"]),
"clean_orders": (6, ["land_orders"]),
"clean_downtime": (5, ["land_downtime"]),
"run_timeline": (25, ["clean_events", "clean_orders"]),
"loss_breakdown": (15, ["clean_events", "clean_downtime"]),
"shift_kpis": (12, ["run_timeline", "loss_breakdown"]),
"dq_checks": (7, ["run_timeline", "loss_breakdown"]),
"publish_gold": (3, ["shift_kpis", "dq_checks"]),
"notify": (1, ["publish_gold"]),
}
def layers(tasks): # Kahn's algorithm
indeg = {t: len(up) for t, (_, up) in tasks.items()}
down = defaultdict(list)
for t, (_, up) in tasks.items():
for u in up: down[u].append(t)
ready = deque(t for t, d in indeg.items() if d == 0)
out = []
while ready:
layer = sorted(ready); ready.clear(); out.append(layer)
for t in layer:
for v in down[t]:
indeg[v] -= 1
if indeg[v] == 0: ready.append(v)
if sum(map(len, out)) != len(tasks):
raise ValueError("cycle: some tasks never became ready")
return out
def critical_path(tasks): # longest path on a DAG, in topological order
finish, via = {}, {}
for layer in layers(tasks):
for t in layer:
dur, up = tasks[t]
prev = max(up, key=lambda u: finish[u], default=None)
finish[t] = dur + (finish[prev] if prev else 0)
via[t] = prev
end = max(finish, key=finish.get)
path = []
while end: path.append(end); end = via[end]
return finish[path[0]], path[::-1]
def has_cycle(tasks): # DFS with white/grey/black
WHITE, GREY, BLACK = 0, 1, 2
color = {t: WHITE for t in tasks}
def dfs(t):
color[t] = GREY
for u in tasks[t][1]:
if color[u] == GREY or (color[u] == WHITE and dfs(u)): return True
color[t] = BLACK
return False
return any(color[t] == WHITE and dfs(t) for t in tasks)4 · And the spanning tree
Different problem, same chapter. You have plants and a distribution center and you want to connect them all with the least total cable, or lane-miles, or whatever the edge cost is, with no redundant links. That's a minimum spanning tree, and it's the cleanest example of a greedy algorithm that is provably optimal — Kleinberg–Tardos §4.5 proves it with the cut property: the cheapest edge crossing any cut is in some MST. Abdul Bari's lecture 44 covers Prim and Kruskal side by side; here's Kruskal, because union-find is the data structure worth seeing.
Sort the edges by cost. Take each one unless it would close a cycle — and "would it close a cycle" is a union-find query: are both endpoints already in the same component? Accept, merge the components; reject otherwise. With path compression and union by rank, each query is effectively constant time, so Kruskal is O(E log E) and the log is the sort.
Where it shows up in data work: clustering (single-linkage is an MST with the longest edges cut), network design, approximating the traveling salesman (the MST is within a factor of two of the optimal tour — KT §11.1), and the "connect these N data sources with the fewest cross-region links" question that comes up in every platform design review.