--- title: 'Dependency and Triggering' source: 'https://academia.sh/en/courses/data-pipelines/dependency-and-triggering' course: 'Data Pipelines and Orchestration' language: en updated: '2026-10-21T05:02:58+00:00' license: 'CC BY-SA 4.0' objectives: - 'Runs time, event and readiness triggers on the same arrival plan and counts the empty and partial runs.' - 'Measures that the choice of trigger is a trade-off between lag and missing periods.' - 'Shows that when the readiness criterion is tied to a fixed number, the task is never triggered.' --- # Dependency and Triggering Measuring what starts the task at the head of the chain and how many times and when the same job runs: while the load of the source starts one step late in three periods and arrives in two parts in four periods, the fixed calendar finishes 3 of 12 runs empty and 3 partial, and by leaving six periods missing it drops consumption from 2200 to 900 rows; when a safety window of two periods is added to the calendar, the runs rise to 24 and all twelve periods are processed a second time, yet one period is still missing; the readiness trigger gives the full result with 12 runs and without empty or partial runs, and its cost is a total lag of seven periods; when the readiness criterion is tied to a fixed row count, however, four periods are never triggered. The previous lesson built the pipeline as a graph of forty-eight tasks and counted how many of them cannot run at all when a task fails. There was only one thing the graph did not say: who starts the chain. Thirty-six tasks are triggered by the completion of another task — the dependency trigger is the graph itself. The remaining twelve `ingest` tasks have no ancestor, and it is not the graph but a decision from outside that runs them. This lesson divides that decision into three classes and measures them on the same pipeline. The measured number is the answer to a simple question: **how many times and when did the same job run.** A task running extra is not free, while a task not running at all is silent; the two flaws sit side by side in the same table. - **OO9.** The source, the pipeline and the arrival plan are **fictional**; no real scheduler, workflow product or messaging system is called. The seed is **20260218**, and the measurement does not depend on the run. - **OO10.** A **step** is a period. **No raw time is written**; delay is counted in **periods**. - **OO11.** The load of the source starts either in its own period or one step late, and arrives in one or two **parts**. Both pieces of information are derived from the seed, not chosen by hand. - **OO12.** An **empty run** is a task that finishes without seeing any rows. A **partial run** is a task that sees only a part of the rows of the period. - **OO13.** In this lesson the write is **keyed**: the second run of the same period does not duplicate rows. Duplication is the subject of the next lesson and is kept out so that it does not contaminate the numbers here. - **OO14.** The baseline is the **run without orchestration**: a fixed calendar set by hand, no retry, no readiness criterion. - **OO15.** The queue, topic and delivery guarantee procedure was measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum and is **not repeated**. The only question here is **when** the trigger starts the task. - **OO16.** Resolution: periods **12**, rows per period **200**, healthy consumption **2200**. **No ratio is printed on serving**; the value itself is written. ## Three Trigger Classes The **time trigger** is the cheapest: the task runs at a moment written in the calendar and never looks at the source. This is the written-down form of running the pipeline by hand, and it is the baseline of this lesson. Its contract is one-sided — the scheduler makes a promise, the source does not. The **event trigger** hands the decision to the source: when the source drops a part of the load into the pipeline, it sends a notification, and the task runs on that notification. Because the source is the party that knows when it is ready, this looks more correct. The problem is **what** the notification announces: if a load arrives in two parts, two notifications are sent and the task runs twice. The **readiness trigger** leaves the decision to the pipeline itself: the task waits until it verifies that the load of the period is complete. It produces no empty or partial run, but because it waits, it is delayed, and **the criterion of readiness** has to be written. When that criterion is written wrongly, the task never runs. ```python # FICTIONAL. The meter readings, the five-stage pipeline and the arrival plan are # defined inside the lesson; no real scheduler, orchestrator or messaging system 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): 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 def s4_summary(records): totals = {} for k in records: key = (k["region"], k["period"]) totals[key] = totals.get(key, 0) + k["m3"] return totals, 0 def s5_serve(summary): region = {} for (b, d), v in summary.items(): region.setdefault(b, []).append(v) return {b: round(sum(v) / len(v), 2) for b, v in region.items()}, 0 def read_result(records): v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) summ, _ = s4_summary(v) serving, _ = s5_serve(summ) return serving, len(v) ``` ## The Fictional Arrival Plan What separates the triggers is the behavior of the source. If the source kept to the calendar, all three classes would give the same result; the distinction comes from the source arriving **sometimes late and sometimes split**. The second block derives this behavior from the seed: the load of each period starts at its own step or one step later, and arrives in one or two parts. ```python SOURCE = generate_source() BASELINE_SERVING, BASELINE_N = read_result(SOURCE) HORIZON = 13 def arrival_plan(late_share=0.25, split_share=0.35): """FICTIONAL: the load of each period arrives in one or two parts , some periods start one step late.""" r = rng(SEED + 11) plan = {} for p in PERIODS: start = PERIODS.index(p) + (1 if r() < late_share else 0) parts = 2 if r() < split_share else 1 plan[p] = {"start": start, "parts": parts, "done": start + parts - 1} return plan PLAN = arrival_plan() def visible(period, step): """Rows that have reached the pipeline by this step.""" p, rows = PLAN[period], [k for k in SOURCE if k["period"] == period] if step < p["start"]: return [] arrived = min(p["parts"], step - p["start"] + 1) return rows[:len(rows) * arrived // p["parts"]] def schedule(mode): """(step , period) pairs: when and for which period the task runs.""" if mode == "time": # fixed calendar return [(t, PERIODS[t]) for t in range(12)] if mode == "time_window": # calendar , last two periods return sorted((t, PERIODS[g]) for t in range(HORIZON) for g in (t, t - 1) if 0 <= g < 12) if mode == "event": # each part sends a notification return sorted((PLAN[p]["start"] + j, p) for p in PERIODS for j in range(PLAN[p]["parts"])) return [(PLAN[p]["done"], p) for p in PERIODS] # readiness def measure(mode): """The write is keyed: the second run of the same period does not duplicate rows.""" store, empty, partial, counts, complete = {}, 0, 0, {}, {} for step, p in schedule(mode): v = visible(p, step) if not v: empty += 1 elif len(v) < 200: partial += 1 counts[p] = counts.get(p, 0) + 1 for k in v: store[(k["subscriber"], k["period"])] = k if p not in complete and sum(1 for x in store if x[1] == p) == 200: complete[p] = step - PERIODS.index(p) serving, n = read_result(list(store.values())) return {"runs": len(schedule(mode)), "empty": empty, "partial": partial, "twice": sum(1 for p in counts if counts[p] > 1), "missing": len(PERIODS) - len(complete), "lag": sum(complete.values()), "consumption": n, "diff": {b: round(serving[b] - BASELINE_SERVING[b], 2) for b in BASELINE_SERVING}} print("late-starting periods:", [p for p in PERIODS if PLAN[p]["start"] != PERIODS.index(p)]) print("periods arriving in two parts:", [p for p in PERIODS if PLAN[p]["parts"] == 2]) print("load completion lag (periods):", sum(PLAN[p]["done"] - PERIODS.index(p) for p in PERIODS)) ``` ``` late-starting periods: ['2026-02', '2026-10', '2026-12'] periods arriving in two parts: ['2026-01', '2026-02', '2026-06', '2026-09'] load completion lag (periods): 7 ``` Three of the twelve periods start one step late, four are split into two parts, and a total lag of seven periods builds up. This is not a faulty source; collecting a part of the meter readings late and sending large loads in pieces are ordinary matters. What is to be measured is this: against the same source, how many tasks do the four trigger setups run, and what do they leave behind. ## How Many Times and When the Same Job Ran Four setups are compared. `time` looks at the calendar once. `time + two periods` adds a safety window, set by hand, to the same calendar: at each step it reprocesses both that period and the one before it. `event` runs on the notification of each load part. `readiness` waits until the two hundred rows of the period are complete. ```python print(f"{'trigger':<20}{'runs':>6}{'empty':>7}{'partial':>9}{'twice':>7}" f"{'missing':>9}{'lag':>9}{'consumption':>13}") for mode, name in (("time", "time"), ("time_window", "time + two periods"), ("event", "event"), ("readiness", "readiness")): o = measure(mode) print(f"{name:<20}{o['runs']:>6}{o['empty']:>7}{o['partial']:>9}{o['twice']:>7}" f"{o['missing']:>9}{o['lag']:>9}{o['consumption']:>13}") print() print("time deviation :", measure("time")["diff"]) print("time + two periods :", measure("time_window")["diff"]) print("event and readiness :", measure("readiness")["diff"]) ``` ``` trigger runs empty partial twice missing lag consumption time 12 3 3 0 6 0 900 time + two periods 24 3 4 12 1 5 2000 event 16 0 4 4 0 7 2200 readiness 12 0 0 0 0 7 2200 time deviation : {'central': -317.28, 'north': -382.88, 'east': -175.69, 'west': -240.59, 'south': -251.72} time + two periods : {'central': -100.63, 'north': -145.19, 'east': -65.54, 'west': -86.82, 'south': -80.46} event and readiness : {'central': 0.0, 'north': 0.0, 'east': 0.0, 'west': 0.0, 'south': 0.0} ``` The first row is the baseline and the most misleading row in the table. The fixed calendar writes **zero in the lag column**; no period was processed late, because no period was waited for. In the same row **six periods remain missing**: three of the twelve runs finish empty, three partial. Consumption comes out as **900** rows instead of 2200, and serving collapses in all five regions (between −175.69 and −382.88). **Zero lag is zero not because it measured waiting, but because there was no waiting at all.** The second row is the answer given in the field to this flaw: adding a safety window without leaving the calendar. The number of runs goes from 12 to **24**, and **all twelve periods are processed a second time** — that is, half of the pipeline is unnecessary work. In return, the missing periods fall from six to one and consumption rises to 2000. But the one remaining period keeps making the pipeline pay: `2026-02` both starts one step late and is split into two parts, completes at the third step, and the two-period window no longer looks at it at the third step. Serving is still wrong in every region (between −65.54 and −145.19). **Widening the window is choosing a boundary, and every boundary that is chosen leaves some arrival plan outside.** The third row shows both the strength and the flaw of the event trigger. The missing periods are zero, the serving deviation is empty, the result is complete. On the other hand, **16 runs** were made: the four periods that arrive in two parts were processed twice, and the first run of those four is a **partial** run. Because this lesson keeps the write keyed, the partial run does no harm; the next lesson will measure what the same partial run does when the write is unkeyed. The fourth row is the setup that is looked for, and it states plainly what is bought. The readiness trigger gives the complete result **with 12 runs**: no empty run, no partial run, no period processed a second time. The cost it pays is **the seven in the lag column** — because it waits for the load to complete, it has finished a total of seven periods late. This number is exactly the same as the lag of the source, that is, the trigger adds no lag of its own; it only **makes the lag of the source visible**. ## The Flaw of the Readiness Criterion The cost of the readiness trigger is not only lag. From now on a **criterion** determines when the task runs, and that criterion is written by hand. In the setup above the criterion was of the form "two hundred rows arrived"; the number two hundred is the current subscriber count of the source and is an assumption. ```python # Starting with the ninth period, four subscribers ended their subscriptions: the source sends 196 rows. LEAVING = {"A197", "A198", "A199", "A200"} SOURCE2 = [k for k in SOURCE if not (k["period"] >= "2026-09" and k["subscriber"] in LEAVING)] COUNT = {p: sum(1 for k in SOURCE2 if k["period"] == p) for p in PERIODS} print("rows per period:", sorted(set(COUNT.values())), " total:", len(SOURCE2)) print(f"{'readiness criterion':<28}{'periods run':>12}{'rows':>7}{'consumption':>13}") results = {} for name, threshold in (("source-reported parts", 0), ("fixed 200 rows", 200)): running = [p for p in PERIODS if COUNT[p] >= threshold] accepted = [k for k in SOURCE2 if k["period"] in running] serving, n = read_result(accepted) results[name] = serving print(f"{name:<28}{len(running):>12}{len(accepted):>7}{n:>13}") a, b = results["source-reported parts"], results["fixed 200 rows"] print("deviation of fixed threshold:", {x: round(b[x] - a[x], 2) for x in a}) ``` ``` rows per period: [196, 200] total: 2384 readiness criterion periods run rows consumption source-reported parts 12 2384 2184 fixed 200 rows 8 1600 1400 deviation of fixed threshold: {'central': -37.59, 'north': -10.75, 'east': -18.5, 'west': 8.11, 'south': -13.98} ``` Because four subscribers ended their subscriptions, the source sends 196 rows after the ninth period. This is not a corruption; the source sends the correct data completely. The readiness criterion tied to a fixed threshold, however, waits for the second hundredth row, so **it never triggers in four periods**: the periods that run are **8** instead of 12, consumption is **1400** instead of 2184. No task raises an error, no log line is written, and the state of the pipeline stays waiting. The deviation row says how this flaw looks in serving: while central falls by −37.59, **west rises by +8.11**. Because the lost periods are not distributed equally among the regions, even the sign of the deviation is not consistent; that is, it is impossible to conclude "four periods are missing" by looking at serving. The right criterion is not a fixed number but **the completeness of parts reported by the source itself**; when it is used, all twelve periods run and consumption stays at 2184 rows. ## Summary - The thirty-six tasks in the graph are triggered by dependency; a decision from outside starts the twelve `ingest` tasks, and this decision is a time, event or readiness trigger. - The fixed calendar, the baseline, writes **zero in the lag column** but leaves **six periods missing**: 3 of 12 runs finish empty, 3 partial, and consumption is **900** rows instead of 2200. - Adding a safety window of two periods to the calendar raises the runs to **24** and processes all twelve periods a second time; the missing periods fall to one, but serving is still wrong in every region. - The event trigger gives the complete result; its cost is **16 runs** and four periods processed a second time; the first run of the four periods that arrive in two parts is a **partial run**. - The readiness trigger gives the complete result **with 12 runs** and without empty or partial runs; its cost is the **total lag of seven periods**, and this lag is exactly the same as that of the source. - When the readiness criterion is tied to a fixed row count, four periods **never trigger**: consumption falls from 2184 to 1400, the sign of the deviation changes with the region, and no task raises an error. ## Next Step All the runs in this lesson had one thing in common: none of them failed. The task either ran or was never triggered; when it ran, it wrote all of the rows it saw. In a real pipeline there is a third outcome — the task runs, starts writing and **fails in the middle**. The partial run was harmless in this lesson, because the write was keyed and the missing part was completed in the next run. The next lesson removes that guarantee and measures two questions: what a partially written partition does to serving, and whether retrying the failed task repairs the pipeline or duplicates rows.