Workflow Schedulers

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.

# 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.

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.

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.

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.