--- title: 'Workflow Schedulers' source: 'https://academia.sh/en/courses/data-pipelines/workflow-schedulers' course: 'Data Pipelines and Orchestration' language: en updated: '2026-10-21T05:02:59+00:00' license: 'CC BY-SA 4.0' objectives: - 'Splits the stages of the pipeline into per-period tasks and builds their dependencies as a directed acyclic graph.' - 'Counts the blast radius of a failed task and shows that the shape of the chain determines this number.' - 'Measures that a missing class of edges makes the blast radius look smaller than it is.' --- # Workflow Schedulers Writing the dependencies between the steps of a pipeline as a graph and measuring how many of the remaining tasks cannot run at all when one task fails: twelve periods with four task types give 48 nodes and 47 edges; because the consumption task also depends on the screening of the previous period, 5 tasks cannot run when the ingest task of the first period fails, 4 when the screening of the seventh period fails and 1 when its consumption fails; when the eleven edges between periods are not written, the total blast radius falls from 116 to 72 and 44 tasks are left behind without being rerun, whereas in a manual run all 48 tasks are rerun for a single failed task. Over eleven lessons the pipeline was built: it was ingested from the source, transformed, split into layers and reapplied to the past. All of these measurements carried a single silent assumption — someone runs the pipeline, the stages run in order, and all of them finish. The assumption holds for a pipeline run by hand, because the person running it keeps the order in their head and decides what to do on the spot when a step fails. This lesson takes that order out of the head and puts it into writing. Two questions are asked: how large a structure results when the dependencies between the steps of the pipeline are written completely, and how many of the rest cannot run at all when a single node of that structure fails. The answer to the second question is the **blast radius**, and it is the first number of this course. - **OO1.** The source, the pipeline and the task graph are **fictional**; no real orchestrator, workflow product or scheduler product is called. The seed is **20260218**, and the measurement does not depend on the run. - **OO2.** The rule of the course: the number of a pipeline is not that it ran correctly once, but that it **gives the same result when it is run again**; a pipeline that cannot be rerun counts as unmeasured. - **OO3.** The baseline is the **run without orchestration**: a single manual trigger, no retry, no atomic write, no backfill. - **OO4.** A **task** is the run of a stage **for a specific period**. There are four task types: `ingest`, `screen`, `consumption`, `summary`. - **OO5.** The directed acyclic graph and topological sorting were measured in the Data Structures course; **the algorithm is not repeated here**. The graph is only **built and used**. - **OO6.** The **blast radius** is the number of tasks that cannot run at all when one task fails. It is an integer; no ratio is printed on top of it. - **OO7.** Counted units: task, blocked task, rerun task, consumption row. **No raw time is written**; delay is written in **periods** or **tasks**. - **OO8.** The graph itself is written by hand. The edge list is an assumption and **can be incomplete**. ## Task, Not Stage The five stages of the pipeline are not the unit of a workflow scheduler. For a scheduler the unit is the **task**: the run of a stage for a specific period. `screen` is a stage, `screen@2026-07` is a task. The difference between them is measurable: there are five stages, and over twelve periods there are forty-eight tasks. The four task types overlap with the five stages of the pipeline as follows. `ingest` puts the source's load for that period into the pipeline; it has no counterpart in the stage list of the Introduction to Data Engineering course, because the data was taken as ready there. `screen` combines the first two stages, that is screening and deduplication, in a single task. `consumption` and `summary` are the third and the fourth stage. The fifth stage, `serve`, does not enter the graph: serving looks not at a period but at all twelve periods, and it does not run once per period. ```python # FICTIONAL. The meter readings, the five-stage pipeline and the task graph are # defined inside the lesson; no real store, orchestrator or scheduler is called. SEED, M32 = 20260218, 0xFFFFFFFF def rng(t): x = ((t ^ (t >> 16)) * 2246822507) & M32 x = ((x ^ (x >> 13)) * 3266489909) & M32 s = [(x ^ (x >> 16)) & M32] def next_value(): s[0] = (s[0] * 1664525 + 1013904223) & M32 return s[0] / 4294967296 return next_value REGION = ["east", "west", "north", "south", "central"] TYPE = ["residential", "commercial"] PERIODS = [f"2026-{a:02d}" for a in range(1, 13)] SUBSCRIBERS = {} for i in range(200): r = rng(SEED + 37 * i) SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)], "start": 10000 + int(r() * 60000)} def generate_source(): records = [] for i, (sub, a) in enumerate(SUBSCRIBERS.items()): e = a["start"] for p in PERIODS: r = rng(SEED + 101 * i + 7 * PERIODS.index(p)) e += 4 + int(r() * 46) records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"], "index": e, "unit": "m3"}) return records def s1_screen(records): REQUIRED = ("subscriber", "period", "region", "index", "unit") keep = [k for k in records if all(a in k for a in REQUIRED)] return keep, len(records) - len(keep) def s2_dedupe(records): seen, keep = set(), [] for k in records: key = (k["subscriber"], k["period"]) if key in seen: continue seen.add(key) keep.append(k) return keep, len(records) - len(keep) def s3_consumption(records): """Consumption from the period difference. Without a previous period the row cannot be computed.""" table = {(k["subscriber"], k["period"]): k for k in records} out, dropped = [], 0 for k in records: i = PERIODS.index(k["period"]) if i == 0: continue prev = table.get((k["subscriber"], PERIODS[i - 1])) if prev is None: dropped += 1 continue try: f = int(k["index"]) - int(prev["index"]) except (TypeError, ValueError): dropped += 1 continue out.append(dict(k, m3=f)) return out, dropped # ---- FICTIONAL task graph: four tasks per period TASKS = ("ingest", "screen", "consumption", "summary") DEPENDENCIES = {"ingest": (), "screen": ("ingest",), "consumption": ("screen",), "summary": ("consumption",)} def graph(): """(task , period) nodes. `consumption` also depends on the previous period's `screen`.""" nodes, edges = [], [] for p in PERIODS: for t in TASKS: nodes.append((t, p)) for d in DEPENDENCIES[t]: edges.append(((d, p), (t, p))) i = PERIODS.index(p) if i > 0: edges.append((("screen", PERIODS[i - 1]), ("consumption", p))) return nodes, edges def order_tasks(nodes, edges): """Execution order in a directed acyclic graph.""" incoming = {u: 0 for u in nodes} successors = {u: [] for u in nodes} for a, b in edges: incoming[b] += 1 successors[a].append(b) ready = [u for u in nodes if incoming[u] == 0] order = [] while ready: u = ready.pop(0) order.append(u) for v in successors[u]: incoming[v] -= 1 if incoming[v] == 0: ready.append(v) return order, successors def affected(nodes, edges, failed): """Which tasks can never run once a task fails.""" _, successors = order_tasks(nodes, edges) stack, closed = [failed], set() while stack: u = stack.pop() for v in successors[u]: if v not in closed: closed.add(v) stack.append(v) return closed ``` ## The Edge Between Periods The surprising part of the graph is that the `consumption` task depends on two ancestors at once: the `screen` of its own period and the `screen` of the **previous period**. This is not an invented link; it follows from the definition of the consumption stage. The stage finds consumption by subtracting the index of the previous period from the index of a subscriber, which means that if the previous period is not at hand, it cannot produce the row at all. The first block shows this with numbers. ```python SOURCE = generate_source() print(f"{'periods run':<22}{'input':>7}{'consumption':>13}{'dropped':>9}") for selection in (["2026-07"], ["2026-06", "2026-07"], PERIODS): v = [k for k in SOURCE if k["period"] in selection] v1, _ = s1_screen(v) v2, _ = s2_dedupe(v1) c, dropped = s3_consumption(v2) name = "+".join(selection) if len(selection) < 3 else "twelve periods" print(f"{name:<22}{len(v):>7}{len(c):>13}{dropped:>9}") NODES, EDGES = graph() ORDER, _ = order_tasks(NODES, EDGES) print() print("nodes:", len(NODES), " edges:", len(EDGES), " ordered tasks:", len(ORDER)) print("first three:", ORDER[:3]) print("last three:", ORDER[-3:]) ``` ``` periods run input consumption dropped 2026-07 200 0 200 2026-06+2026-07 400 200 200 twelve periods 2400 2200 0 nodes: 48 edges: 47 ordered tasks: 48 first three: [('ingest', '2026-01'), ('ingest', '2026-02'), ('ingest', '2026-03')] last three: [('summary', '2026-10'), ('summary', '2026-11'), ('summary', '2026-12')] ``` The first row proves the link: a run that puts only `2026-07` into the pipeline produces **zero** consumption rows from 200 input rows and drops all 200 of them. In the second row, once the previous period is added, the same 200 rows become computable. So the edge between periods is not a preference but the computation of the pipeline itself; when it is not written, the graph represents not the pipeline but an approximation of it. The third row is the trace of the same rule in the healthy pipeline. When all twelve periods are at hand, the dropped rows are **zero**, yet consumption still produces **2200** rows, not 2400: the 200 subscriptions of the first period have nothing before them, and these rows are left out of the computation by definition, not as an error. This gap at the start of the chain is visible in the graph as well — the `consumption@2026-01` node exists and runs, it has no rows it could produce. Once the graph is built, **48 nodes** and **47 edges** result. Within a period there are three edges (`ingest` → `screen` → `consumption` → `summary`), which makes thirty-six over twelve periods; eleven edges between periods are added to this. The ordering places all forty-eight tasks, so the graph is acyclic. The procedure that produces the ordering was itself measured in the Data Structures course and is not repeated here; the question here is not how the ordering is found but **what information the graph carries**. ## Blast Radius When a task fails, the first decision the scheduler makes is not to run its successors. How many tasks cannot run is read from the graph, and this number changes with the place of the failed task in the graph. The second block counts three failure points by comparing the same graph with a version from which the edges between periods have been removed. ```python FLAT = [(a, b) for a, b in EDGES if a[1] == b[1]] print("real edges:", len(EDGES), " within-period edges only:", len(FLAT)) print() print(f"{'failed task':<20}{'blocked':>11}{'flat chain':>14}{'rerun':>15}") for t, p in (("ingest", "2026-01"), ("screen", "2026-07"), ("consumption", "2026-07"), ("summary", "2026-07"), ("ingest", "2026-12")): e = len(affected(NODES, EDGES, (t, p))) z = len(affected(NODES, FLAT, (t, p))) print(f"{t + '@' + p:<20}{e:>11}{z:>14}{e + 1:>15}") print() t = sum(len(affected(NODES, EDGES, u)) for u in NODES) z = sum(len(affected(NODES, FLAT, u)) for u in NODES) print("total blast radius:", t, " flat chain:", z, " difference:", t - z) ``` ``` real edges: 47 within-period edges only: 36 failed task blocked flat chain rerun ingest@2026-01 5 3 6 screen@2026-07 4 2 5 consumption@2026-07 1 1 2 summary@2026-07 0 0 1 ingest@2026-12 3 3 4 total blast radius: 116 flat chain: 72 difference: 44 ``` The table turns a single sentence into a number: **the shape of the chain determines the size of the blast.** On the same pipeline, among the same forty-eight tasks, the number of blocked tasks varies between **0** and **5** depending on the failed task. When `summary@2026-07` fails, no task is blocked, because it has no successor. When `consumption@2026-07` fails, only the summary of its own period stops. When `ingest@2026-01` fails, five tasks cannot run, and two of them are tasks of **another period**: the edge between periods ties the screening of `2026-01` to the consumption of `2026-02`. The last column gives the real gain of orchestration. The failed task plus the blocked tasks is the set that has to be rerun: **6, 5, 2, 1** and **4** tasks respectively. This number cannot be known without the graph. The third column, on the other hand, is the flaw of orchestration itself. A graph without the eleven edges between periods is also built, ordered and run; nothing raises an error. But it shows the blast radius as **3** instead of 5 for `ingest@2026-01` and **2** instead of 4 for `screen@2026-07`. Summed over all nodes, the real graph gives **116** and the incomplete graph **72**: the **44 tasks** in between are not rerun on the day they fail and stay in place with their old results. That a graph works does not mean that it is correct. ## Checkpoint The graph says **what is blocked** when a task fails. What it does not say is **what has finished**. The scheduler records each task when it completes; this record is the **checkpoint**, and it determines where a restart continues from. The two pieces of information cannot stand in for each other: one looks downstream, the other backward. ```python print(f"{'failed task':<20}{'completed':>11}{'with checkpoint':>17}{'without checkpoint':>20}{'blocked':>9}") for t, p in (("ingest", "2026-01"), ("screen", "2026-07"), ("consumption", "2026-07")): i = ORDER.index((t, p)) print(f"{t + '@' + p:<20}{i:>11}{len(ORDER) - i:>17}{len(ORDER):>20}" f"{len(affected(NODES, EDGES, (t, p))):>9}") ``` ``` failed task completed with checkpoint without checkpoint blocked ingest@2026-01 0 48 48 5 screen@2026-07 18 30 48 4 consumption@2026-07 30 18 48 1 ``` When `screen@2026-07` fails, the **18 tasks** before it in the ordering have finished. If the checkpoint is kept, the restart runs **30 tasks**; if it is not kept, nothing is known to have finished, so **48 tasks** run. For `consumption@2026-07` the difference is larger: 48 tasks instead of 18. What the three rows say together is that the two numbers answer different questions. The number of blocked tasks gives **which results are no longer valid**; the checkpoint gives **which work will not be repeated**. The `ingest@2026-01` row shows this from the opposite side: when the first task fails, the completed tasks are zero and the checkpoint gains nothing, yet the blast radius is still five. This is why a scheduler has two separate records. ## The Cost of a Manual Run The baseline without orchestration is set against this table. Because the person running by hand has not put the dependency into writing, they do not know which tasks were broken when `screen@2026-07` fails. Two options are left: either rerun only the failed task and leave the consumption of `2026-08` with its old result, or do the safe thing and rerun **all forty-eight tasks**. The second option is correct and its cost is 9.6 times: 5 tasks are enough with the graph, whereas 48 tasks run by hand. The cost is not only the number of tasks, as the next two lessons will show. Rerunning forty-eight tasks means running the `ingest` task twelve times again, and if the write of the pipeline is not idempotent, this run duplicates rows. The gain of the graph doubles here: fewer tasks run, and the data of the periods that were not touched is never rewritten. On the other hand, the graph itself is not free. All forty-seven edges are written by hand, updated by hand as the code of the pipeline changes, and nothing checks their correctness. How a missing edge passes silently was measured above: the ordering is still complete, the run is still green, the blast radius still gives a number — only a wrong one. ## Summary - The unit of orchestration is not the stage but the **task**: the run of a stage for a specific period. Twelve periods and four task types give **48 nodes** and **47 edges**, and the ordering places all forty-eight. - The consumption task depends on the screening of the previous period, and this link is measured: when only `2026-07` is run, **0** consumption rows come out of 200 input rows; when `2026-06` is added, **200** come out. - The blast radius varies between **0** and **5** depending on the place of the failed task in the graph: 5 for `ingest@2026-01`, 4 for `screen@2026-07`, 1 for `consumption@2026-07`, 0 for `summary@2026-07`. - When the eleven edges between periods are not written, the graph still works but shows the total blast radius as **72** instead of **116**; the **44 tasks** in between are left without being rerun. - The checkpoint gives a separate number: when `screen@2026-07` fails, 18 tasks have finished, and the restart runs **30** tasks in the setup with a checkpoint and **48** in the setup without one. - In the baseline without orchestration, **all 48 tasks** are rerun for a single failed task; with the graph, **5 tasks** are enough for the same failure. The cost is that the forty-seven edges are written and updated by hand. ## Next Step The graph gives the order and says whom a failed task blocks, but there is one thing it never says: **who starts the chain**. Thirty-six of the forty-eight tasks are triggered by the completion of another task; twelve `ingest` tasks remain, and they have no ancestor. The question of what runs these tasks is more expensive than it looks, because if the answer is a fixed calendar, the task runs before the data arrives and finishes empty. The next lesson compares three trigger classes on the same pipeline and measures a single number: how many times and **when** the same job runs.