--- title: 'Data Pipelines and Orchestration' source: 'https://academia.sh/en/courses/data-pipelines' language: en updated: '2026-10-21T05:02:58+00:00' license: 'CC BY-SA 4.0' lessons: 16 outcomes: - 'Chooses ingestion methods by the constraints of the source system and counts the change each one misses' - 'Splits transformation pipelines into layers, tests them with data tests and shows how a backfill makes the report inconsistent' - 'Builds a workflow with dependencies as a graph and measures the blast radius when one task fails' --- # Data Pipelines and Orchestration Counting who runs the pipeline, when and in what order, and what the rest does when a step fails: the rows full load, incremental load and change data capture carry and the changes they miss, the calls a connector completes under pagination and rate limits, the work the number of topics and partitions distributes to a consumer group, the cost of windowing and carrying state, at-most-once delivery losing 19 rows and shifting every region while at-least-once delivery's 19 duplicates leave the serving layer unchanged, the loop opened by processed data returning to the operational system, the effect on the cells read of whether transformation happens at the source or the target, the task dependency resolution orders, the rows the raw, refined and serving layers keep, the corruption data tests catch, a partial backfill pushing the report from 16.98 to 29.57 away from the full application with no check reporting it, the number of tasks blocked when one task fails in a 48-node graph, the difference between time, event and data readiness triggers, the rows a retry multiplies, how many periods later freshness and volume indicators show the corruption, and environment and version management of pipeline code. ## Data Ingestion Counting how data is ingested from the source into the pipeline and what each method misses: the rows transported by full load, the late-arriving change that incremental load misses, and the case where change data capture sees both; the calls a connector completes under pagination and a rate limit and the requests a retry adds; the work that topic and partition counts distribute to a consumer group and the order preserved within each partition; the result a windowing decision changes for a late-arriving record and the cost of carrying state; the 19 rows that at-most-once delivery loses, shifting every region, against the 19 duplicates of at-least-once delivery that never change the serving thanks to a keyed write; the loop that writing processed data back to an operational system opens and the number of stages after which that loop returns to its own input. ### Ingestion Types Source: https://academia.sh/en/courses/data-pipelines/ingestion-types Counting side by side what three ingestion methods transport and what each misses: a graph built from four tasks per period carries 48 nodes and 47 edges, full load transports 15,560 rows over twelve runs and misses no change, incremental load transports 2,360 rows and misses 148 changes, and change data capture transports 2,508 rows and misses none. The 40 late-arriving rows and 108 uncorrected rows that incremental load misses pull the serving down in every region, yet the row count check stays silent at a ratio of 0.0167 and only the distribution check catches the corruption, at 0.3469 in stage 4. The previous course measured, across twelve lessons, **where** data sits and in **what layout**: partitioning, file layout, compression, and cells read were counted side by side. **Who fills** that layout, **when**, and **in what order** was never asked. A pipeline does not fill itself; every period someone triggers it, the steps run in a sequence, and what the rest should do when one of them drops out of that sequence has to be written down in advance. This course measures the pipeline through the **run**, and its rule is this: a pipeline's number is not that it ran correctly once but that it **gives the same result when run again**; a pipeline that cannot be rerun counts as **unmeasured**. In every lesson three numbers sit side by side: the unorchestrated baseline, the orchestrated setup, and the cost. This lesson builds the trio and the pipeline's task graph, then asks the first question: by **which method** is data ingested from the source, and what does that method **miss**. - **DI1.** The code is **fictional**; no real store, engine, messaging system, or orchestrator is called. The source is the fictional meter readings of a measurement network: 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DI2.** The pipeline is deterministic; no randomness library is used and all numbers are independent of the run. - **DI3.** The baseline is the **unorchestrated run**: a single manual trigger, no retries, no atomic write, no backfill. - **DI4.** The task graph carries four **tasks** per period: `ingest`, `screen`, `consumption`, `summary`. `consumption` also depends on the `screen` task of the previous period. - **DI5.** A late-arriving row is the `late_arrival` corruption of the Introduction to Data Engineering course: 40 rows of the 2026-07 period are absent from that period's run and arrive before the next run. It is not an attack but a producer-side accident. - **DI6.** The master data correction fixes the region of 12 subscribers before the tenth run, and the correction is applied to **all periods** of those subscribers. - **DI7.** The target is keyed: the `(subscriber, period)` pair identifies a row, and when the same key is written again it is overwritten. The write is therefore **idempotent**. - **DI8.** Transported rows are the sum of the number of rows the `ingest` task passes into the pipeline over twelve runs. The step of the row ratio is 1/2400, that is 0.0004; no ratio is printed on the serving. #### The Task Graph of the Pipeline A pipeline is a **directed acyclic graph** of interdependent tasks. The graph concept and topological ordering were measured in the Graphs topic of the Data Structures course; the algorithm is not repeated here, the graph is **built and used**. There are four tasks per period and the one real subtlety is this: consumption is a **difference** computation, so the `consumption` task depends on the `screen` task of its own period as well as on the `screen` task of the **previous period**. The chain reaching across periods explains why, in later lessons, a whole queue waits when one task fails. ```python # FICTIONAL. No real store, engine, or orchestrator is called; the meter # readings of the measurement network are fictional, seed 20260218. 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(): # one reading per subscriber per period 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 # ---- PIPELINE: five stages. Each stage counts the rows it drops. 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): # per-region mean over the periods 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), summ # ---- FICTIONAL task graph: four tasks per period. TASKS = ("ingest", "screen", "consumption", "summary") DEPENDENCIES = {"ingest": (), "screen": ("ingest",), "consumption": ("screen",), "summary": ("consumption",)} def graph(): """`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): 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 nodes, edges = graph() order = order_tasks(nodes, edges) raw = generate_source() serving, n, _ = read_result(raw) print("graph", len(nodes), "nodes", len(edges), "edges , order", len(order), "tasks") print("first three tasks", order[:3]) print("source", len(raw), "rows , consumption", n, "rows") print("healthy serving", serving) ``` ``` graph 48 nodes 47 edges , order 48 tasks first three tasks [('ingest', '2026-01'), ('ingest', '2026-02'), ('ingest', '2026-03')] source 2400 rows , consumption 2200 rows healthy serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} ``` Forty-eight nodes, forty-seven edges, and a graph that can be fully ordered. The edge count being one less than the node count says the graph is a **forest in one piece**: apart from the `ingest` tasks of the twelve periods, every node has exactly one predecessor. What this lesson measures is not the shape of the graph but what the very first node does: **what the `ingest` task asks of the source**. #### Three Ingestion Methods **Full load** reads the whole of the source as it is at that moment on every run and rewrites the target from scratch. **Incremental load** asks only for the periods after the last ingested period; the boundary is a period number and advances on every run. **Change data capture** reads the source's own change stream: it transports insert and update events and never transports a row that was not touched. The block below models the source over time and runs the three methods over the same twelve runs. Two producer-side events fall on the source: forty rows of 2026-07 arrive late, and the region of the twelfth subscriber is corrected before the tenth run. ```python # ---- FICTIONAL source system over time. Two producer-side events: # 1) The `late_arrival` corruption of the Introduction to Data Engineering # course: 40 rows of 2026-07 are absent in that period's run and arrive # before the next run. # 2) Master data correction: the region of 12 subscribers was entered wrongly # and is corrected for all of their periods before run 10. LATE = [f"A{i:03d}" for i in range(1, 41)] WRONG = [f"A{i:03d}" for i in range(101, 113)] def source_snapshot(run): """The state of the source as visible at the moment of run number `run`.""" records = [] for k in generate_source(): i = PERIODS.index(k["period"]) + 1 if i > run or (i == 7 == run and k["subscriber"] in LATE): continue y = dict(k) if k["subscriber"] in WRONG and run < 10: y["region"] = REGION[(REGION.index(k["region"]) + 1) % 5] records.append(y) return records def full_load(): """On every run the whole source is reread , the target is wiped and written.""" target, transported = {}, 0 for run in range(1, 13): snapshot = source_snapshot(run) transported += len(snapshot) target = {(k["subscriber"], k["period"]): k for k in snapshot} return list(target.values()), transported def incremental_load(): """Only the periods after the last ingested period are read.""" target, transported, boundary = {}, 0, 0 for run in range(1, 13): fresh = [k for k in source_snapshot(run) if PERIODS.index(k["period"]) + 1 > boundary] transported += len(fresh) for k in fresh: target[(k["subscriber"], k["period"])] = k boundary = run return list(target.values()), transported def change_data_capture(): """The source's change stream is read , insert and update events are transported.""" target, transported, prev = {}, 0, {} for run in range(1, 13): snapshot = {(k["subscriber"], k["period"]): k for k in source_snapshot(run)} events = [k for key, k in snapshot.items() if prev.get(key) != k] transported += len(events) for k in events: target[(k["subscriber"], k["period"])] = k prev = snapshot return list(target.values()), transported actual = {(k["subscriber"], k["period"]): k for k in raw} correct_serving, _, correct_summary = read_result(raw) print("method transported target missing wrong consumption row ratio distribution width") for name, f in (("full", full_load), ("incremental", incremental_load), ("change capture", change_data_capture)): tgt, transported = f() tm = {(k["subscriber"], k["period"]): k for k in tgt} missing = len(set(actual) - set(tm)) wrong = sum(1 for a in tm if actual[a] != tm[a]) s, t, summ = read_result(tgt) width = max(abs(v - correct_summary[a]) / max(abs(correct_summary[a]), 1) for a, v in summ.items()) print(f"{name:<14} {transported:>11} {len(tm):>6} {missing:>7} {wrong:>5} {t:>11} " f"{abs(len(tm) - 2400) / 2400:>9.4f} {width:>18.4f}") print() for name, f in (("full", full_load), ("incremental", incremental_load), ("change capture", change_data_capture)): s, _, _ = read_result(f()[0]) print(f"{name:<14}", {b: round(s.get(b, 0) - correct_serving[b], 2) for b in correct_serving}) ``` ``` method transported target missing wrong consumption row ratio distribution width full 15560 2400 0 0 2200 0.0000 0.0000 incremental 2360 2360 40 108 2120 0.0167 0.3469 change capture 2508 2400 0 0 2200 0.0000 0.0000 full {'central': 0.0, 'north': 0.0, 'east': 0.0, 'west': 0.0, 'south': 0.0} incremental {'central': -11.18, 'north': -28.1, 'east': -39.09, 'west': -36.91, 'south': -70.0} change capture {'central': 0.0, 'north': 0.0, 'east': 0.0, 'west': 0.0, 'south': 0.0} ``` #### Transported Rows versus Missed Changes The first column of the table writes the cost of full load: over twelve runs **15,560 rows** were transported, whereas the rows that end up in the target are **2400**. The price of transporting the same data six and a half times is that it misses no change. The forty late rows settle into place on their own in the eighth run, and the region correction spreads to all periods in the tenth run, because every run rereads the source. Full load has no flaw; it is only expensive, and **its expense grows linearly as the source grows**. Incremental load transports the fewest rows: **2360**. Its cost sits on the same row. The seventh run sees only 160 rows of the period; the forty rows arrive later, but the boundary is now at seven and those rows are never requested again. The region correction likewise lands only on the periods after the tenth run: nine periods of twelve subscribers, that is **108 rows**, stay with their old region in the target. In total **148 changes are missed**, of which 40 are missing rows and 108 are wrong rows. Incremental load does not **know** what it missed: it has no mechanism that would notice that something behind the boundary changed. Change data capture sees both events, and the rows it transports for this are **2508**. That is only **148 rows** more than incremental load and **13,052 rows** fewer than full load. The difference is exactly the number of missed changes, and this is no coincidence: what change data capture transports is **the change itself**. Its cost is not in rows but at the source; the source has to publish its change stream, and for a source that does not publish it the method cannot be set up at all. #### Which Check Sees the Missed Change Incremental load's target pulls the serving down in every region: the smallest deviation is **−11.18** in central, the largest **−70.00** in south. Is there anyone who would see this deviation? The check set built in the Introduction to Data Engineering course gives two different answers to this question. The ratio of the row count check is **0.0167**; because the default tolerance is 0.05, the check **stays green**. The schema, type, and uniqueness checks are green as well, because the schema of the missing rows is not broken, they are merely absent. The region set is also complete: the region of twelve subscribers is wrong, but the wrong region is on the list too. The only check that catches it is the **distribution check**, and at the summary layer the largest cell deviation of **0.3469** exceeds the 0.25 threshold. This check runs at stage 4; that is, the corruption becomes visible only after passing through ingestion, screening, deduplication, and the consumption computation. The detection lag is **4 stages**, and by then the 2120-row intermediate table has already been written. Method choice turns here into a choice of measure: incremental load cuts transported rows by a factor of 6.6, misses 148 changes in return, and this leak is reported only by the most expensive check, at the latest stage. #### Summary - The pipeline is a directed acyclic graph built from four tasks per period: 48 nodes, 47 edges, and a run order that can be fully sorted. `consumption` depends on the `screen` task of the previous period, so the chain reaches across periods. - Full load misses no change and pays its cost in transported rows: for a target of 2400 rows, 15,560 rows are transported over twelve runs. - Incremental load transports the fewest rows (2360) but cannot see what happens behind the boundary: 40 late-arriving rows and 108 uncorrected rows, 148 changes in total, are missed. - Change data capture sees both events and transports only as many rows more than incremental load as the number of missed changes, 148; its condition is that the source publishes its change stream. - The row count check does not see the missed change, at a ratio of 0.0167; only the distribution check reports it, at 0.3469 and with a lag of 4 stages. #### Next Step In this lesson the `ingest` task read the source in a **single move**: `source_snapshot` was called and the rows were in hand. Source systems are not read like this. An endpoint serves data page by page, limits how many calls can be made per window, and some of the calls drop temporarily. The next lesson puts the same `ingest` task under these constraints and counts two things: **the calls completed and the rows ingested under the constraint**. The trade-off to be measured is this: raising the page size lowers the call count, but the rows missed by a single dropped call grow by the same ratio. ### Connectors and API Consumption Source: https://academia.sh/en/courses/data-pipelines/connectors-and-api-consumption Counting the calls a connector completes and the rows it receives under pagination and a rate limit: with a quota of eight calls per window, on a 25-row page 83 of 96 calls complete and 2075 rows are received, on a 100-row page 20 of 24 calls complete and 2000 rows are received, so the rows missed per dropped call grow from 25 to 100 with the page size. Retry eats the quota on the small page and drops the rows received from 2075 to 2025, delays 112 pages by one window, and the 2026-12 period is never received; in return, the serving deviation falls from −374 to −125, because the loss stops being scattered and collects at the tail of the queue. The connector's own call ledger reports the shortfall without delay: expected 96, completed 83. The previous lesson treated the `ingest` task as a single move: the current state of the source was called and the rows were in hand. Source systems are not read like this. A source that serves data through an endpoint delivers it **page by page**, **limits** how many calls can be made in a given window, and some of the calls drop temporarily. A single run of the `ingest` task is no longer one operation but a sequence of dozens of calls. This lesson's measure is two numbers: **the calls completed under the constraint** and **the rows received**. Pagination and rate limiting themselves were measured in the Web API Design, SQL Fundamentals, and Building Applications with Model APIs courses; the procedure is not repeated here. The only question asked here is this: under these constraints, how many of the rows the pipeline wants can a **connector** bring in, and what trace does the row it fails to bring leave behind. - **DI9.** The code is **fictional**; no real endpoint, connector product, or orchestrator is called. The source is again 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DI10.** The endpoint serves the rows of a period in pieces of **page size** and reports how many rows the period holds; the connector therefore knows in advance how many pages to request. - **DI11.** The rate limit is modeled as a **quota per window**: each period has one window, and at most `quota` calls can complete in that window. A page that does not fit is left to the next window; that is a delay of one **period**. Time is not written as raw duration. - **DI12.** The transient failure rate is 0.15 and is derived from the seed; the same attempt at the same page gives the same result on every run. The pipeline is deterministic. - **DI13.** The retry allowance is **eaten from the quota**: the repeat of a failed call is also a call. The backoff procedure was measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum and is not repeated here. - **DI14.** A page whose allowance runs out is **abandoned** and the connector moves on; the rows of that page are never received. - **DI15.** The baseline is the unorchestrated run: no retry, no queue, a single manual trigger. Retry is the only mechanism this lesson adds. - **DI16.** The step of the row ratio is 1/2400, that is 0.0004; no ratio is printed on the serving. #### A Connector Under Constraint The block below sets up the common definition and adds to it a paginated endpoint and a connector that consumes it. The connector has a single queue: in each period the pages of that period enter the queue, pages are pulled until the window quota runs out, and the rest roll over to the next window. ```python # FICTIONAL. No real endpoint, connector product, or orchestrator is called; # the meter readings of the measurement network are fictional, seed 20260218. 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(): # one reading per subscriber per period 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 # ---- PIPELINE: five stages. Each stage counts the rows it drops. 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): # per-region mean over the periods 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) # ---- FICTIONAL endpoint: paginated read , quota per window , transient failure. def page(records, period, number, page_size): rows = [k for k in records if k["period"] == period] return rows[number * page_size:(number + 1) * page_size] def call_fails(period, number, attempt, failure_rate): """The transient failure is derived from the seed , it is independent of the run.""" r = rng(SEED + 13 * PERIODS.index(period) + 3 * number + 101 * attempt) return r() < failure_rate def connector(records, page_size, quota, failure_rate=0.15, retries=0): """Every period has one window ; at most `quota` calls complete in a window. A page that does not fit is left to the next window , that is a delay of one period.""" count = (200 + page_size - 1) // page_size received, completed, failed, delayed, queue = [], 0, 0, 0, [] for p in PERIODS: queue += [(p, n) for n in range(count)] remaining = quota while queue and remaining > 0: qp, n = queue[0] passed, attempt = False, 0 while attempt <= retries and remaining > 0: remaining -= 1 if call_fails(qp, n, attempt, failure_rate): failed += 1 attempt += 1 continue completed += 1 passed = True break if passed or attempt > retries: # either received or the allowance ran out queue.pop(0) if passed: received.extend(page(records, qp, n, page_size)) delayed += len(queue) # pages waiting one more window return received, completed, failed, delayed, len(queue) raw = generate_source() correct_serving, correct_n = read_result(raw) print("quota 8 calls/window , transient failure rate 0.15") print("size pages retries expected completed failed delayed unfetched received consumption") for page_size in (25, 50, 100): for retries in (0, 2): rec, completed, failed, delayed, left = connector(raw, page_size, 8, 0.15, retries) s, t = read_result(rec) count = (200 + page_size - 1) // page_size print(f"{page_size:>4} {count:>6} {retries:>7} {12 * count:>9} {completed:>9} {failed:>6} " f"{delayed:>7} {left:>9} {len(rec):>8} {t:>11}") ``` ``` quota 8 calls/window , transient failure rate 0.15 size pages retries expected completed failed delayed unfetched received consumption 25 8 0 96 83 13 0 0 2075 1575 25 8 2 96 81 15 112 15 2025 1825 50 4 0 48 39 9 0 0 1950 1300 50 4 2 48 48 12 0 0 2400 2200 100 2 0 24 20 4 0 0 2000 1400 100 2 2 24 24 6 0 0 2400 2200 ``` #### Page Size and the Rows Missed per Call The three rows without retry give the first trade-off. With a page size of 25 a period is eight pages, and twelve periods ask for **96 calls**; raising the size to 100 brings the same job down to **24 calls**. The call count falls in inverse proportion to the page size, and against a source with a rate limit this is a direct gain. The price of the gain is written in the same rows. At size 25, 13 calls drop and 325 rows are missed; at size 50, 9 calls drop but 450 rows are missed; at size 100, only 4 calls drop and again 400 rows are missed. The product holds exactly in every row: the rows missed are **the number of dropped calls times the page size**. Quadrupling the page size cuts the call count to a quarter and **quadruples the cost of a single failure**. The connector setting is chosen at the point where these two curves cross, and which one weighs more depends on the failure rate of the source. The consumption column shows how the cost grows inside the pipeline. At size 50, 450 rows are missed but the consumption rows fall from 2200 to **1300**: each missed row not only lowers the consumption of its own period, it also makes the consumption of **the next period** uncomputable, because consumption is a difference computation. A shortfall of 450 rows turns into a loss of 900 rows. #### Retry Eating the Quota The most rows a window can carry is the product of the quota and the page size. With a quota of eight, size 25 gives a ceiling of **200 rows** per window, size 50 gives **400 rows**, and size 100 gives **800 rows**. The demand of a period, though, is exactly 200 rows. At size 25 the ceiling and the demand are **exactly equal** and the connector has no slack at all; at size 50 the ceiling is twice the demand, and at size 100 it is four times. What decides whether retry can find room for itself is this slack, not the failure rate. In the size 50 and size 100 rows retry is **nearly free**: because 48 and 24 calls stay under the quota, failed calls can be repeated and the rows received rise to 2400 and the consumption to 2200. The deviation is zeroed. In the size 25 row the same mechanism turns around. The expected 96 calls are already at the exact limit of the quota; because retries eat from the quota, a few pages cannot keep up in each window and the queue grows. The result is 2075 rows received **without** retry against **2025 rows** received **with** retry. Adding the mechanism has lowered the rows received. Two more numbers sit beside it: **112 pages** have been delayed by at least one window, and **15 pages** are still waiting in the queue when the twelve windows are over. This is precisely the second claim of this course: the orchestration mechanism also breaks things, and only the one who looks at its own counter sees what it broke. #### The Connector's Own Call Ledger The next block puts three setups side by side and asks **where** the loss falls. ```python FULL_SET = set((b, d) for b in REGION for d in PERIODS) print("size quota retries received row ratio partition_period received rows per period") for page_size, quota, retries in ((25, 8, 0), (25, 8, 2), (25, 12, 2)): rec, completed, failed, delayed, left = connector(raw, page_size, quota, 0.15, retries) counts = [sum(1 for k in rec if k["period"] == p) for p in PERIODS] seen = set((k["region"], k["period"]) for k in rec) print(f"{page_size:>4} {quota:>5} {retries:>7} {len(rec):>8} {(2400 - len(rec)) / 2400:>9.4f} " f"{str(seen != FULL_SET):>16} {counts}") print() for page_size, quota, retries in ((25, 8, 0), (25, 8, 2), (25, 12, 2)): s, t = read_result(connector(raw, page_size, quota, 0.15, retries)[0]) print(f"size {page_size} quota {quota} retries {retries}", {b: round(s.get(b, 0) - correct_serving[b], 2) for b in correct_serving}) ``` ``` size quota retries received row ratio partition_period received rows per period 25 8 0 2075 0.1354 False [175, 175, 175, 150, 175, 125, 200, 175, 200, 175, 150, 200] 25 8 2 2025 0.1562 True [200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 25, 0] 25 12 2 2400 0.0000 False [200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200, 200] size 25 quota 8 retries 0 {'central': -373.72, 'north': -302.19, 'east': -278.0, 'west': -190.91, 'south': -367.28} size 25 quota 8 retries 2 {'central': -124.85, 'north': -118.25, 'east': -82.06, 'west': -78.39, 'south': -105.05} size 25 quota 12 retries 2 {'central': 0.0, 'north': 0.0, 'east': 0.0, 'west': 0.0, 'south': 0.0} ``` The period sequence in the second row is the most counterintuitive result of the lesson. The setup with retry **receives fewer rows** but its serving deviation is **smaller**: −125 … −78 instead of −374 … −191. The reason lies not in the amount of the loss but in **its place**. In the setup without retry the loss is scattered over twelve periods and pulls the total of every period down; in the setup with retry the first ten periods have been received **completely**, 2026-11 has received only 25 rows because the queue grew, and 2026-12 has **not been received at all**. Because the region mean is computed over the periods that exist, the serving looks repaired. An improvement in the number served does not mean the data has improved. The answer to who reports the loss also splits across the two rows. For the scattered loss, the period-aware completeness check **stays green**, because every region is still visible in every period; the only thing that catches it is the row count check, at a ratio of 0.1354. For the queue loss the same check **fires**, because there is a period with no rows at all. The cheaper of the two is not inside the pipeline but in the connector's own ledger: against the expected 96 calls, 83 have completed, and this difference is known **without waiting for any stage**, the moment the calls finish. Raising the quota from eight to twelve produces the last row: all 96 calls complete, 2400 rows are received, the deviation empties, and no delayed page remains. #### Summary - A connector's work is counted under constraint: a page size of 25 needs 96 calls, a page size of 100 needs 24; the call count is inversely proportional to the page size, and against a source with a rate limit this is a direct gain. - The missed rows are the number of dropped calls times the page size: 13 calls 325 rows, 9 calls 450 rows, 4 calls 400 rows. Enlarging the page size raises the cost of a single failure by the same ratio. - The missed rows grow inside the pipeline: because consumption is a difference computation, a shortfall of 450 rows turns into a loss of 900 rows. - Under the quota retry is free and brings back all 2400 rows; at the limit of the quota it eats the quota, delays 112 pages, and lowers the rows received from 2075 to 2025. - When the loss collects at the tail of the queue the serving deviation shrinks (−125 … −78) but the last two periods collapse: 2026-11 receives only 25 rows and 2026-12 is not received at all. The row count does not see this, the period-aware completeness check does, and the cheapest report comes from the connector's call ledger without delay. #### Next Step The connector in this lesson **pulled** the source: it decided for itself when to read, spent its own quota, and reported what it missed from its own ledger. A mechanism in which the source **pushes** the data on its own turns this balance around. Records are written to a **topic**, the topic is split into **partitions**, and a **consumer group** reads the partitions. The next lesson measures this mechanism through two numbers: the work that the partition count distributes among consumers, and the order preserved per partition. The trade-off to be measured is this: as the partition count grows, parallelism grows, but when the records of the same subscriber fall into different partitions, the order on which the consumption computation relies breaks. ### Messaging Systems Source: https://academia.sh/en/courses/data-pipelines/messaging-systems Counting side by side the work that the partition count distributes to a consumer group and the order preserved per partition: in a group of four consumers a single partition leaves three consumers idle, four partitions put all of them to work. In a topic partitioned by the subscriber key no subscriber is split and consumption stays at 2200 rows at every partition count, but the load skews between 708 and 492. The most balanced key, the record key, squeezes the load into 583 to 621 and splits all 200 subscribers: consumption falls to 527 rows at four partitions and the serving shifts between 616 and 977 in every region. The period key, on the other hand, loses 1,200 of the 2200 rows at two partitions while moving the serving only between −15.29 and +16.15. The previous two lessons treated ingestion through **pulling**: the pipeline went to the source, decided for itself what to ask, and spent its own quota. A mechanism in which the source **pushes** the data on its own turns this balance around. Records are written to a **topic**, the topic is split into **partitions**, and a **consumer group** reads the partitions. Queues, the publish–subscribe pattern, offsets, and delivery semantics were measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum; the procedure is not repeated here. This lesson's measure is two numbers: **the work the partition count distributes to consumers** and **the order preserved per partition**. They are two faces of the same setting. As the partition count grows, parallelism grows; but when the records of the same subscriber fall into different partitions, the order on which the consumption computation relies breaks and the pipeline loses rows. - **DI17.** The code is **fictional**; no real messaging system, stream processor, or orchestrator is called. The source is again 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DI18.** Writing to the topic goes period by period; after all subscriber records of a period have been written, the next period begins. - **DI19.** A record falls into a single partition according to the **partition key**, and the same key always goes to the same partition. The **offset** within a partition preserves the write order; there is no such order across partitions. - **DI20.** A partition is read by **a single consumer** within the group. The partition count is therefore the ceiling of the number of active consumers in the same group. The group has four consumers. - **DI21.** Three partition keys are tested: `subscriber`, `record` (the combination of subscriber and period), and `period`. All three are fields of the source itself; none is invented. - **DI22.** Each consumer computes consumption **only from the records in its own partition**; if the previous period of a subscriber is in another partition, that row cannot be computed and is dropped. - **DI23.** The baseline is the unorchestrated run; on the healthy pipeline consumption is 2200 rows and the serving deviation is empty. - **DI24.** The partition count and the idle consumers are whole numbers, no ratio is printed. The step of the row ratio is 1/2400, that is 0.0004; no ratio is printed on the serving. #### Topic, Partition, and Consumer Group The block below sets up the common definition and adds to it a partitioned topic and a consumer group that reads it. Five numbers are printed for each setup: the row counts of the most loaded and the least loaded consumer, the number of idle consumers, the number of subscribers spread over more than one partition, and the consumption rows collected from the partitions. ```python # FICTIONAL. No real messaging system, stream processor, or orchestrator # is called; the meter readings of the measurement network are fictional, seed 20260218. 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(): # one reading per subscriber per period 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 # ---- PIPELINE: five stages. Each stage counts the rows it drops. 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): # per-region mean over the periods 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) # ---- FICTIONAL topic: partitions , partition key , consumer group. def route(key, p): """Partition number from the key. The same key always falls into the same partition.""" x = 0 for c in key: x = (x * 131 + ord(c)) & M32 return int(rng(SEED + x)() * p) def partition_records(records, p, key): """Writing to the topic goes period by period ; each record is written to a partition by its key. The offset within the partition preserves the write order.""" partitions = {b: [] for b in range(p)} for k in sorted(records, key=lambda k: (k["period"], k["subscriber"])): value = {"subscriber": k["subscriber"], "record": k["subscriber"] + k["period"], "period": k["period"]}[key] b = route(value, p) partitions[b].append(dict(k, offset=len(partitions[b]))) return partitions def consumer_group(partitions, c): """A partition is read by a single consumer within the group ; the partition count is the ceiling.""" assignment = {t: [] for t in range(c)} for b in partitions: assignment[b % c].append(b) return assignment raw = generate_source() correct_serving, correct_n = read_result(raw) print("consumer group 4 consumers , consumption on the healthy pipeline", correct_n, "rows") print("key partitions max load min load idle split subscribers computed consumption") for key in ("subscriber", "record", "period"): for p in (1, 2, 4, 8): parts = partition_records(raw, p, key) assign = consumer_group(parts, 4) loads = [sum(len(parts[b]) for b in assign[t]) for t in assign] placed = {} for b in parts: for k in parts[b]: placed.setdefault(k["subscriber"], set()).add(b) consumed = sum(len(s3_consumption(parts[b])[0]) for b in parts) print(f"{key:<10} {p:>10} {max(loads):>8} {min(loads):>8} " f"{sum(1 for y in loads if y == 0):>5} " f"{sum(1 for v in placed.values() if len(v) > 1):>17} {consumed:>20}") ``` ``` consumer group 4 consumers , consumption on the healthy pipeline 2200 rows key partitions max load min load idle split subscribers computed consumption subscriber 1 2400 0 3 0 2200 subscriber 2 1200 0 2 0 2200 subscriber 4 708 492 0 0 2200 subscriber 8 696 528 0 0 2200 record 1 2400 0 3 0 2200 record 2 1211 0 2 200 1076 record 4 621 583 0 200 527 record 8 624 572 0 200 268 period 1 2400 0 3 0 2200 period 2 1600 0 2 200 1000 period 4 800 400 0 200 200 period 8 1000 200 0 200 200 ``` #### The Work the Partition Count Distributes The first pair of columns writes a fact independent of the key: in a topic with a single partition, **three of the four consumers sit idle**, with two partitions two sit idle, and when the count goes up to four none is left idle. The partition count is not a preference but a **ceiling**; adding consumers to the group gains nothing beyond the point where they exceed the partition count. At eight partitions four consumers work as well, but each reads two partitions. The second observation is how the load is divided. With the subscriber key, four partitions skew between 708 and 492 rows, that is the most loaded consumer does 1.44 times the work of the emptiest one. With the record key the same setup is squeezed into 621 to 583: this is **the most balanced of the three keys**, because a key that combines subscriber and period produces 2400 distinct values and spreads almost evenly over the partitions. The period key is the worst; distributing twelve distinct values over four partitions opens the load to between 800 and 400, and at eight partitions to between 1000 and 200. Someone looking only at these three rows would choose the record key. #### Order Preserved per Partition The last two columns reverse the choice. With the subscriber key **no subscriber is split** and consumption stays at **2200 rows** at every partition count. With the record key **all 200 subscribers** are split, and consumption falls to 527 rows at four partitions and to 268 at eight. The mechanism is a single sentence: a partition preserves order within itself, there is no order across partitions, and consumption is **the difference of two consecutive periods** of a subscriber. If the two periods fall into different partitions, there is no consumer to compute that row. ```python print("key partitions computed consumption serving deviation") for key, p in (("subscriber", 4), ("record", 4), ("period", 2), ("period", 4)): parts = partition_records(raw, p, key) summ, consumed = {}, 0 for b in parts: # each consumer reads its own partition v, _ = s3_consumption(parts[b]) consumed += len(v) o, _ = s4_summary(v) for name, s in o.items(): summ[name] = summ.get(name, 0) + s s, _ = s5_serve(summ) print(f"{key:<10} {p:>10} {consumed:>20} ", {b: round(s.get(b, 0) - correct_serving[b], 2) for b in correct_serving}) parts = partition_records(raw, 4, "subscriber") print() print("subscriber-keyed partition 0 :", len(parts[0]), "records , last offset", parts[0][-1]["offset"]) print("first three records :", [(k["subscriber"], k["period"], k["offset"]) for k in parts[0][:3]]) sub = parts[0][0]["subscriber"] print(sub, "partitions of its records :", sorted(set(b for b in parts for k in parts[b] if k["subscriber"] == sub))) rec_parts = partition_records(raw, 4, "record") print(sub, "under the record key :", sorted(set(b for b in rec_parts for k in rec_parts[b] if k["subscriber"] == sub))) ``` ``` key partitions computed consumption serving deviation subscriber 4 2200 {'central': 0.0, 'north': 0.0, 'east': 0.0, 'west': 0.0, 'south': 0.0} record 4 527 {'central': -977.27, 'north': -960.55, 'east': -686.91, 'west': -616.0, 'south': -801.1} period 2 1000 {'central': 16.15, 'north': -7.35, 'east': 12.84, 'west': -15.29, 'south': 4.65} period 4 200 {'central': 180.55, 'north': 57.45, 'east': 124.64, 'west': -24.09, 'south': 25.45} subscriber-keyed partition 0 : 492 records , last offset 491 first three records : [('A005', '2026-01', 0), ('A007', '2026-01', 1), ('A014', '2026-01', 2)] A005 partitions of its records : [0] A005 under the record key : [0, 1, 2, 3] ``` The last four lines show the mechanism on a single subscriber. With the subscriber key, all twelve records of A005 are in partition 0, and the consumer that reads that partition sees all twelve in order. With the record key, the records of the same subscriber are spread over all four partitions; no consumer sees two consecutive periods of that subscriber together. The offset within the partition keeps increasing flawlessly — partition 0 numbers its own 492 records from 0 to 491. The order has not been broken; it has been **preserved in the wrong place**. The second job of the offset touches the rule of the course. A consumer group keeps a single number per partition: how far it has read in that partition. When the group fails or is set up again, it continues reading from that offset, and when the offset is moved back the same records are read **again**. This is the basis of making the ingestion step rerunnable; to be able to do the same with a pulling connector, the source must still be serving the old rows. Since the last offset of partition 0 is 491, reading that partition from the start brings back 492 records, and as long as the target write is idempotent, the second run gives the same result as the first. The rule that follows is that the most balanced key may be the most wrong key. The partition key looks like a load-distribution setting, whereas it determines **which records the consumer can see together**. The right key is the grouping field of the computation: because consumption is computed per subscriber, the key should be the subscriber. The price of this is a load skew of 1.44 times, and it is cheap against the whole of the 2200 rows. #### The Served Number Misleading The two rows of the period key are the quietest result of the lesson. At two partitions consumption falls from 2200 to **1000 rows**, that is 1,200 rows disappear, yet the serving deviation stays between **−15.29 and +16.15**: a movement not even reaching 2% on values of a thousand-odd units. The four-partition setup with the record key, on the other hand, does not look like it loses fewer rows — with 527 rows it loses more — and it brings the serving down by **−616 to −977**. The difference lies not in the amount of the loss but in the loss's **shape**. The summary layer takes a total for each region and period pair, and the serving layer takes the period mean of those totals. The period key drops a whole period at once: the cell of that period is never formed and the mean is taken over the remaining periods, each correct on its own. The record key, however, drops random rows from inside every period: the cells are formed but they total short, and the mean goes down across the board. A larger loss from the same mechanism can produce a smaller deviation. The correctness of a partitioning decision cannot be measured by looking at the served number; its measure is the consumption rows computed. #### Summary - The partition count is the ceiling of the number of active consumers in the same group: with a single partition three of the four consumers, with two partitions two of them stay idle; with four partitions all work. - The partition key is not a load-distribution setting but a decision that determines which records the consumer can see together. A partition preserves order within itself, there is no order across partitions. - The subscriber key splits no subscriber and keeps consumption at 2200 rows at every partition count; its price is a load skew between 708 and 492, that is 1.44 times. - The most balanced key may be the most wrong key: the record key squeezes the load into 621 to 583 but splits all 200 subscribers and lowers consumption to 527 rows at four partitions. - The period key moves the serving only between −15.29 and +16.15 while losing 1,200 of the 2200 rows at two partitions, because dropping a whole period does not disturb the mean. The measure of the partitioning decision is not the served number but the consumption rows computed. #### Next Step In these three lessons the data was **ingested** into the pipeline: the method was chosen, the constraint of the source was counted, and the topic was partitioned and distributed to a consumer group. In all three, the period was treated as a closed box — all the records of a period were counted as in hand while that period was being processed. A record can arrive **after** its own period has closed, and then which window will own the result becomes a decision. The next lesson measures stream processing with this question: how the window decision changes the result for a late-arriving record, and what the cost of **carrying state** between windows is. The common question of the remaining lessons was never asked here: whether data **is lost** while being ingested into the pipeline. ### Stream Processing Source: https://academia.sh/en/courses/data-pipelines/stream-processing Turning the window decision and the carried state into numbers: when the period window is closed the moment the period ends, 40 records that arrive one period late are dropped and 10 of the 55 summary cells are polluted; when the window is kept open one period longer the dropped rows fall to 0, but the 2026-07 cell is written in 2026-08 and the carried state grows from 400 entries to 600. Tolerance is a bet on delay: when the same records are two periods late, tolerance 1 gives exactly the same result as tolerance 0. In none of the nine setups do the five served values change, only the number of polluted cells does. If the window never closes the state grows to 2400 and the mode turns back into batch processing; with the state lost without a checkpoint, 200 rows cannot be computed, the summary falls to 50 cells, and the serving shifts in every region. In the previous three lessons the data was ingested into the pipeline: the method of ingestion, how to ingest under the constraint of the source, and which partition the messages would come from were written down. Whether the ingested row arrives **completely** in the pipeline was not asked. This lesson measures the first form of loss, and that form is not a transport error: it is **a record that arrives late to a window that has closed**. Batch processing has no window question, because the period is looked at after it has closed. In stream processing the source never ends; to take a total, the stream has to be cut somewhere. The cutting decision determines three things at once: which record enters the computation, in which period the result is written, and how much **state** the run carries. This lesson puts those three numbers side by side. - **DI41.** The code is **fictional**; no real stream processor, messaging system, or orchestrator is called. The source is the fictional meter readings of the Data Engineering and MLOps curriculum: 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DI42. The comparison of batch and stream mode is not repeated**; it was measured in the Batch and Stream Processing lesson of the Introduction to Data Engineering course. The baseline here is the stream mode of that lesson; the only thing added is **how long the window stays open**. - **DI43.** The window is one period. The **watermark** points to the latest period counted as closed; **tolerance** is how many more periods the window stays open before it closes. - **DI44.** A late-arriving record is counted not as "never arrived" but as **arrived one period later**: the forty readings of period 2026-07 arrive in 2026-08. The corruption class is the `late_arrival` class of the common definition. - **DI45.** The carried state has two parts: the rows waiting in open windows and the last index per subscriber. The measured size is the **largest value** the sum of the two reaches during the run. - **DI46.** In stream mode consumption is computed from the state: the consumption of a row is subtracted from the last index of the same subscriber that sits in the state. A row with no counterpart in the state cannot be computed. - **DI47.** Delay is written as **periods**, not as a raw duration. The step of the summary cell ratio is 1/55, that is 0.0182; no ratio is printed on the serving, the value itself is written. - **DI48.** A checkpoint is the storage of the carried state outside the run. Its procedure is not written in this lesson; only the **cost of losing it** is counted. #### Window, Tolerance, and State The block below sets up the common definition and adds a single runner to it. `stream_window` takes in each period the rows that **arrive** in that period, drops those that belong to a window the watermark has closed, keeps the rest waiting in the open windows, and closes the window whose turn has come and writes its summary cells. ```python # FICTIONAL. No real stream processor , messaging system or orchestrator is called; # the meter readings of the measurement network are fictional , seed 20260218. 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(): # one reading per subscriber per period 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): # batch mode: consumption from the period difference 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 out.append(dict(k, m3=int(k["index"]) - int(prev["index"]))) 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): # per-region mean over the periods 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 with_arrival(records, delay): """Gives each row an arrival period. late_arrival class: the A001..A040 readings of period 2026-07 arrive `delay` periods later.""" data = [] for k in records: i = PERIODS.index(k["period"]) g = delay if k["period"] == "2026-07" and k["subscriber"] <= "A040" else 0 data.append(dict(k, arrival=PERIODS[i + g])) return data def stream_window(data, tolerance, reset_at=None): """The period window stays open for `tolerance` more periods , then closes. reset_at: the carried state is lost at the start of this period (no checkpoint).""" buffer, last_index, cells, written = {}, {}, {}, {} dropped, state, uncomputable = 0, 0, 0 def close(p, i): nonlocal uncomputable v, _ = s1_screen(buffer.pop(p, [])) v, _ = s2_dedupe(v) c = [] for k in v: if k["subscriber"] in last_index: c.append(dict(k, m3=k["index"] - last_index[k["subscriber"]])) elif PERIODS.index(p) > 0: uncomputable += 1 for k in v: last_index[k["subscriber"]] = k["index"] # state: last index per subscriber o, _ = s4_summary(c) if o: cells.update(o) written[p] = PERIODS[i] for i, d in enumerate(PERIODS): if d == reset_at: last_index.clear() for k in data: if k["arrival"] != d: continue if PERIODS.index(k["period"]) < i - tolerance: dropped += 1 # the watermark closed the window continue buffer.setdefault(k["period"], []).append(k) state = max(state, sum(len(v) for v in buffer.values()) + len(last_index)) if i - tolerance >= 0: close(PERIODS[i - tolerance], i) for j in range(max(0, len(PERIODS) - tolerance), len(PERIODS)): close(PERIODS[j], len(PERIODS) - 1) return {"dropped": dropped, "state": state, "cells": cells, "written": written, "uncomputable": uncomputable} raw = generate_source() v, _ = s1_screen(raw) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) BASELINE, _ = s4_summary(v) def dirty(h): return sum(1 for name in set(h) | set(BASELINE) if h.get(name) != BASELINE.get(name)) def diff(h): s, t = s5_serve(h)[0], s5_serve(BASELINE)[0] d = {b: round(s[b] - t[b], 2) for b in t if round(s.get(b, 0) - t[b], 2)} return d if d else "empty" print("baseline: consumption rows", len(v), ", summary cells", len(BASELINE)) print("baseline serving", s5_serve(BASELINE)[0]) print() print("40-row delay tolerance dropped state dirty/55 07 written serving diff") for delay in (0, 1, 2): for t in (0, 1, 2): r = stream_window(with_arrival(raw, delay), t) print(f"{delay:>10} period {t:>9} {r['dropped']:>7} {r['state']:>6}" f" {dirty(r['cells']):>9} {r['written']['2026-07']:>10} {diff(r['cells'])}") ``` ``` baseline: consumption rows 2200 , summary cells 55 baseline serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} 40-row delay tolerance dropped state dirty/55 07 written serving diff 0 period 0 0 400 0 2026-07 empty 0 period 1 0 600 0 2026-08 empty 0 period 2 0 800 0 2026-09 empty 1 period 0 40 400 10 2026-07 empty 1 period 1 0 600 0 2026-08 empty 1 period 2 0 800 0 2026-09 empty 2 period 0 40 400 10 2026-07 empty 2 period 1 40 600 10 2026-08 empty 2 period 2 0 800 0 2026-09 empty ``` #### The Result the Window Decision Changes The first three rows show the **cost** of the decision without any loss. Even when no record arrives late, tolerance 1 raises the carried state from 400 entries to 600, tolerance 2 raises it to 800, and the 2026-07 cell is written one or two periods late. This cost is paid **in every run**; whether a late record arrives or not does not change it. Keeping a window open is reserving room for a record that has not arrived, and the room is reserved even if that record never comes. The middle three rows give the return. When the forty readings arrive one period late, tolerance 0 **drops** them: the window has closed and the watermark is not taken back. The result is that **10** of the 55 summary cells are polluted — the 2026-07 cells are left short, and the 2026-08 cells are inflated, because the consumption of those forty subscribers is computed from a two-period difference. A record dropped from one window breaks not only its own cell but **also the cell of the period that follows**; the reason is that consumption is computed from the state. At tolerance 1 the same records still find the window open: dropped rows **0**, polluted cells **0**. The last three rows define what tolerance is: a guess. When the same forty records are two periods late, tolerance 1 drops them too, and the result is **exactly the same** as tolerance 0 — 40 dropped rows, 10 polluted cells. Tolerance is a bet on how long the delay will be; when the bet does not hold, the price paid cannot be taken back, and moreover it was paid before the right answer could be obtained. Tolerance 2 wins this time, but in all of the tolerance 2 rows the state has risen to 800 entries. The last column is the warning of the lesson. In all nine rows the **serving difference is empty**. Despite ten polluted cells, the five region values served are exactly the same as the baseline, because what falls short in 2026-07 is in excess in 2026-08 and the mean of the eleven periods does not move. The only evidence that the window decision is wrong is **the comparison at cell level**; no observation that looks at the output can question this decision. The rows that give the right answer also have a cost, and that cost sits in the same table: with tolerance the 2026-07 cell is written one or two periods late. Keeping the window open is delaying the result. #### The Two Costs of State Enlarging the tolerance buys correctness up to a point, and that has a limit. The first table of the block below shows the limit: what happens if the window never closes. The second table measures the real nature of the state — the carried state is not a memory size but the **condition for the pipeline to be able to produce the right answer**. In stream mode consumption is computed from the last index in the state; if the state is lost, the row cannot be computed. The second table runs the tolerance 1 setup twice: with the state kept between runs and with the state lost at the beginning of 2026-08. ```python print("window tolerance state dropped dirty/55 07 written") for name, t in (("closes each period", 1), ("never closes", 11)): r = stream_window(with_arrival(raw, 1), t) print(f"{name:<18} {t:>9} {r['state']:>6} {r['dropped']:>8} {dirty(r['cells']):>9}" f" {r['written']['2026-07']:>10}") print() print("state uncomputable cells dirty/55") for name, rs in (("kept", None), ("lost", "2026-08")): r = stream_window(with_arrival(raw, 1), 1, reset_at=rs) print(f"{name:<14} {r['uncomputable']:>13} {len(r['cells']):>6} {dirty(r['cells']):>9}") print(" serving", s5_serve(r["cells"])[0]) ``` ``` window tolerance state dropped dirty/55 07 written closes each period 1 600 0 0 2026-08 never closes 11 2400 0 0 2026-12 state uncomputable cells dirty/55 kept 0 55 0 serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} lost 200 50 5 serving {'central': 1325.6, 'north': 1208.7, 'east': 873.7, 'west': 835.2, 'south': 1068.2} ``` The first table gives the upper limit of the tolerance. If the window never closes, no record is dropped and no cell is polluted, but the carried state rises to **2400 entries**, that is the whole of the source, and the 2026-07 cell is written in the last period, 2026-12. This row is no longer stream processing; it is a batch run that accumulates all the data and processes it once at the end. **The window is what makes stream processing stream**: a window that never closes gives back all of the mode's freshness gain. The second table measures the fragility of the state. When the state is lost, **none of the 200 rows** in the 2026-08 window can be computed, because no subscriber's previous index is at hand. The summary falls from 55 cells to **50**: one period is not written at all. The effect of this on the serving is visible this time — all five region values shift, and in two directions at once, with central rising while north falls. The reason is that the region mean is now taken over **ten** periods, not eleven; how the dropped period differs from the mean determines the sign. When the two costs are put side by side, a strange balance of the stream mode appears. The late record corrupted ten cells without moving the serving at all; the lost state moved the serving, but by leaving five cells **missing**, that is by dropping a countable trace. A missing cell is a question that can be asked in an audit; a wrong cell is not. Storing the state outside the run — a **checkpoint** — removes this second cost; it does not remove the first, that is the window decision. What the two costs have in common is that both are **a consequence of the window**. There is no such question in a batch run: every run reads the whole of the source again, a late row enters the computation by itself in the next run, and no state is left to carry. Because the stream mode drops the row it has read, it loses this free correction; in its place it leaves two decisions — how long the window will stay open and where the state will be stored. Neither of these two decisions can be made by looking at the data; they are made by looking at **the distribution of the delay** and the frequency with which the run is interrupted. A pipeline's number is therefore not read from its code alone. #### Summary - In stream processing the window decision determines three things at once: which record enters the computation, in which period the result is written, and how much state the run carries. - At tolerance 0, 40 records that arrive one period late are dropped and 10 of the 55 summary cells are polluted; at tolerance 1 the dropped rows and polluted cells fall to 0, and the price is that the 2026-07 cell is written one period late. - Tolerance is a bet on delay: when the same records are two periods late, tolerance 1 gives exactly the same result as tolerance 0, moreover with a state of 600 entries already paid. - The cost of the tolerance does not depend on the run: the carried state is 400, 600, and 800 entries whether or not a record arrives late; if the window never closes the state rises to 2400, that is the whole of the source. - In none of the nine setups do the five served values change; the only witness of the ten polluted cells is the comparison at cell level. - State lost without a checkpoint makes 200 rows impossible to compute, lowers the summary from 55 cells to 50, and moves the serving in every region. #### Next Step In this lesson the source of the loss was the pipeline's own decision: the record that arrived after the window closed was dropped on purpose, how many were dropped was counted, and when the decision was changed the number fell to zero. The record had **reached** the pipeline; it had only reached it late. What the number would be if a record never arrived, or arrived twice, was not asked. The delivery between the ingestion and the pipeline was assumed to be flawless throughout the lesson: every row sent was processed as if it had arrived exactly once. The next lesson removes this assumption and puts the three delivery modes side by side: the setups in which a row can be lost, can arrive twice, and is written with a key. The question to be measured is this: of loss and duplication, which corrupts the served number, which is caught by a check, and why the **repairability** of the two is not the same. ### Delivery Guarantees Source: https://academia.sh/en/courses/data-pipelines/delivery-guarantees Turning the three delivery modes into numbers on the same pipeline: with at-most-once delivery 19 rows of period 2026-07 are lost, consumption falls from 2200 to 2162, 10 of the 55 summary cells are polluted, and the serving shifts in every region (−9.18 … −28.36); with at-least-once delivery 19 rows arrive twice but consumption stays at 2200 and the serving difference comes out empty, because the dedupe stage swallows the duplication; with exactly-once delivery the keyed write zeroes out both. The real finding is in the check column: the uniqueness check catches the duplication at stage 0, while the loss slips past every check — the row ratio of 0.0079 stays below the 0.05 threshold and the largest cell deviation of 0.2488 stays below the 0.25 threshold. Loss is silent, duplication is repairable. The previous lesson measured one form of loss, but there the record had **reached** the pipeline: it had only reached a window that had closed late, and it was dropped by a decision. This lesson removes that assumption: a row is sent and **never arrives**, or it is sent once and **arrives twice**. These two events are the two unavoidable ends of transport. When the sender gets no acknowledgement it either gives up or tries again; if it gives up the result is loss, if it tries again the result is duplication. The promise that transport makes on this point is called a **delivery guarantee**: **at-most-once** is the mode in which a row can be lost, **at-least-once** the mode in which it can arrive twice, and **exactly-once** the mode in which neither happens. The third often sits not in transport but in the **write**: the row is written with a key, and when the same key comes a second time the write is not made. The lesson runs the three modes on the same pipeline and asks: which one corrupts the served number, and which one is caught by a check. - **DI49.** The code is **fictional**; no messaging system, stream processor, or orchestrator is called. The source is the fictional meter readings of the Data Engineering and MLOps curriculum: 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DI50. The procedure of delivery semantics is not repeated**; it was measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum. The only thing added is what the mode does to **the pipeline's serving**. - **DI51.** Loss and duplication are modeled in a single period, 2026-07, with a share of 0.10 per row. The generator is seeded; 19 of the 200 rows of that period are affected. - **DI52.** The delivery mode is a property of **transport**; the five stages of the pipeline are the same in all three modes. - **DI53. Exactly-once** is modeled here not as a transport property but as a **keyed write**: if the (subscriber, period) pair has been written before, the second write is not made at all. - **DI54.** The check set is taken from the common definition; the row count tolerance is 0.05 and the distribution tolerance is 0.25. - **DI55.** The dedupe stage of the pipeline is **an option** in this lesson; where the duplication goes when it is turned off is measured. - **DI56.** The units counted are rows, lost rows, duplicated rows, polluted cells, and the serving deviation. The step of the row ratio is 1/2400, that is 0.0004; no ratio is printed on the serving. #### Three Modes, One Pipeline The block below sets up the common definition and adds the `deliver` function and the common check set to it. `deliver` touches only period 2026-07: it drops a row, adds a copy of it, or does not write a row whose key has been seen at all. ```python # FICTIONAL. No real messaging system , stream processor or orchestrator is called; # the meter readings of the measurement network are fictional , seed 20260218. 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(): # one reading per subscriber per period 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 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 out.append(dict(k, m3=int(k["index"]) - int(prev["index"]))) 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): # per-region mean over the periods 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 deliver(records, mode, period="2026-07", share=0.10): """At most once: loss. At least once: duplication. Exactly once: keyed write.""" r = rng(SEED + 7) out, lost, duplicates = [], 0, 0 seen = set() for k in records: if k["period"] != period: out.append(k) continue if mode == "at_most_once": if r() < share: lost += 1 continue out.append(k) elif mode == "at_least_once": out.append(k) if r() < share: out.append(dict(k)) duplicates += 1 else: # exactly_once: keyed write key = (k["subscriber"], k["period"]) if r() < share and key in seen: duplicates += 0 continue seen.add(key) out.append(k) return out, lost, duplicates def c_schema(data, fields=("subscriber", "period", "region", "index", "unit")): return any(any(a not in k for a in fields) for k in data) def c_uniqueness(data): key = [(k["subscriber"], k["period"]) for k in data] return len(key) != len(set(key)) def c_row_count(data, expected, tolerance=0.05): return abs(len(data) - expected) / expected > tolerance def c_partition_completeness_period(data): seen = set((k["region"], k["period"]) for k in data) return seen != set((b, d) for b in REGION for d in PERIODS) def c_distribution(summary, baseline, tolerance=0.25): for key, v in summary.items(): t = baseline.get(key) if t and abs(v - t) / max(abs(t), 1) > tolerance: return True return False raw = generate_source() def run(records, dedupe=True): v, _ = s1_screen(records) if dedupe: v, _ = s2_dedupe(v) v, _ = s3_consumption(v) summ, _ = s4_summary(v) serving, _ = s5_serve(summ) return len(v), summ, serving N, BASELINE, BASE_SERVING = run(raw) def dirty(summ): return sum(1 for name in set(summ) | set(BASELINE) if summ.get(name) != BASELINE.get(name)) def diff(serving): d = {b: round(serving[b] - BASE_SERVING[b], 2) for b in BASE_SERVING if round(serving.get(b, 0) - BASE_SERVING[b], 2)} return d if d else "empty" CHECKS = [("schema", lambda c, o: c_schema(c)), ("uniqueness", lambda c, o: c_uniqueness(c)), ("row", lambda c, o: c_row_count(c, 2400)), ("partition_period", lambda c, o: c_partition_completeness_period(c)), ("distribution", lambda c, o: c_distribution(o, BASELINE))] print("baseline: rows 2400 , consumption", N, ", summary cells", len(BASELINE)) print("baseline serving", BASE_SERVING) print() print("mode rows lost duplicates consumption dirty/55 firing check") for mode in ("at_most_once", "at_least_once", "exactly_once"): c, lost, dup = deliver(raw, mode) n, summ, serving = run(c) fired = [name for name, f in CHECKS if f(c, summ)] or ["none"] print(f"{mode:<13} {len(c):>5} {lost:>5} {dup:>10} {n:>11} {dirty(summ):>9} {','.join(fired)}") print(" serving diff", diff(serving)) print() print("keys held by the keyed write: per period", len({(k["subscriber"], k["period"]) for k in raw if k["period"] == "2026-07"}), ", for the whole run", len({(k["subscriber"], k["period"]) for k in raw})) ``` ``` baseline: rows 2400 , consumption 2200 , summary cells 55 baseline serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} mode rows lost duplicates consumption dirty/55 firing check at_most_once 2381 19 0 2162 10 none serving diff {'central': -9.18, 'north': -22.64, 'east': -9.81, 'west': -28.36, 'south': -20.28} at_least_once 2419 0 19 2200 0 uniqueness serving diff empty exactly_once 2400 0 0 2200 0 none serving diff empty keys held by the keyed write: per period 200 , for the whole run 2400 ``` #### The Asymmetry of Loss and Duplication Two rows of the table are mirror images of each other: one removes 19 rows, the other multiplies 19 rows. Their results are nothing alike. In **at-most-once** delivery, 19 rows belonging to period 2026-07 never arrive. The consumption rows fall from 2200 to **2162**, that is the loss costs **38** rows: a missing reading lowers the consumption of its own period, and because it removes the previous index it also makes the consumption of the following period impossible to compute. **10** of the 55 summary cells are polluted and **all** five region values served shift: −28.36 in the west, −9.18 in the center. In the previous lesson exactly 10 cells were polluted too, but the serving did not move at all; the difference is that there what fell short in July was in excess in August, while here **both periods fall short**. In **at-least-once** delivery the same 19 rows arrive twice. The dedupe stage of the pipeline drops the second copy; consumption **stays at 2200**, the polluted cells are **0**, and the serving difference is **empty**. In **exactly-once** mode the keyed write does not make the second write of the same key at all. Since every (subscriber, period) pair is in the source once, the output stays at 2400 rows and there is no duplication to count — the duplication counter of the mode is therefore zero. This mode is a property not of transport but of the **write**; transport can still deliver twice, the write makes it invisible. That it is not free is written on the last line. For the write to recognize the same key a second time, it has to **remember** it: **200** keys for one period, **2400** for the whole run. Exactly-once is not a transport assurance but an assurance **bought by keeping state**; for a key that is not kept there is no guarantee. #### Why the Loss Is Silent The real finding is in the last column. The mode that shifts the five values served in every region is **caught by no check**; the mode that changes no value fires the **uniqueness check**. The block below writes under which thresholds the loss slips through. ```python c, lost, _ = deliver(raw, "at_most_once") n, summ, serving = run(c) print("lost", lost, "rows ; row ratio", round(abs(len(c) - 2400) / 2400, 4), "< threshold 0.05") print("largest summary cell deviation", round(max(abs(summ[a] - BASELINE[a]) / abs(BASELINE[a]) for a in BASELINE if a in summ), 4), "< threshold 0.25") print("region set complete:", not c_partition_completeness_period(c)) print("same check at tolerance 0.005:", c_row_count(c, 2400, tolerance=0.005)) print() print("loss share lost consumption largest cell deviation row ratio firing") for p in (0.02, 0.05, 0.10, 0.20, 1.00): c, lost, _ = deliver(raw, "at_most_once", share=p) n, summ, serving = run(c) mx = round(max(abs(summ[a] - BASELINE[a]) / abs(BASELINE[a]) for a in BASELINE if a in summ), 4) fired = [name for name, f in CHECKS if f(c, summ)] or ["none"] print(f"{p:<11} {lost:>5} {n:>12} {mx:>23} {round(abs(len(c) - 2400) / 2400, 4):>10}" f" {','.join(fired)}") print() print("dedupe stage off:") print("mode consumption dirty/55 serving diff") for mode in ("at_most_once", "at_least_once", "exactly_once"): c, _, _ = deliver(raw, mode) n, summ, serving = run(c, dedupe=False) print(f"{mode:<13} {n:>11} {dirty(summ):>9} {diff(serving)}") ``` ``` lost 19 rows ; row ratio 0.0079 < threshold 0.05 largest summary cell deviation 0.2488 < threshold 0.25 region set complete: True same check at tolerance 0.005: True loss share lost consumption largest cell deviation row ratio firing 0.02 4 2192 0.1441 0.0017 none 0.05 10 2180 0.2094 0.0042 none 0.1 19 2162 0.2488 0.0079 none 0.2 39 2122 0.3953 0.0163 distribution 1.0 200 1800 0.0 0.0833 row,partition_period dedupe stage off: mode consumption dirty/55 serving diff at_most_once 2162 10 {'central': -9.18, 'north': -22.64, 'east': -9.81, 'west': -28.36, 'south': -20.28} at_least_once 2219 5 {'central': 6.19, 'north': 12.27, 'east': 3.64, 'west': 18.36, 'south': 6.45} exactly_once 2200 0 empty ``` The first four lines show how the silence is built. The row count check expects 2400 and finds 2381; the ratio is **0.0079** against a threshold of 0.05, that is the check stays six times below the threshold. The distribution check sees a deviation of **0.2488** in the most corrupted cell and stays **by a hair** below the threshold of 0.25. Partition completeness can see nothing, because the lost rows are spread over the five regions. The loss deletes no field, breaks no type, duplicates no key; it only leaves **fewer** rows — and the checks look at the shape of the row. The flaw is not in the code but in the **tolerance**: the fourth line shows the same check firing at tolerance 0.005. The loss is not unmeasurable, it is unmeasured. The second table sweeps the loss share and shows that the silence is not a point but a **band**. As the share goes from 0.02 to 0.20, the lost rows climb from 4 to 39 and the largest cell deviation from 0.1441 to 0.3953: the effect grows steadily with the share, but the distribution check sees it only at **0.20**. The row count check fires **nowhere** in the band; the reason is arithmetic, the loss is confined to a single period and that period is 200/2400 of the source, that is 0.0833. The last row is the most anomalous finding of the lesson. When 2026-07 never arrives at all, two checks fire **at once** and the deviation of the remaining cells is **0.00**: the cells of 2026-07 and 2026-08 have not been computed wrongly, they have not been written at all, and the remaining 45 cells are flawless. **Total loss is noisier than partial loss.** #### The Place and Cost of Repair The third table explains why the duplication stayed harmless. When the dedupe is turned off, at-least-once delivery is no longer absorbed: consumption rises to **2219**, 5 cells are polluted, and the serving shifts **upward** in every region. Duplication is not harmless by itself; it is the stage that makes it harmless, and the cost of that is again a set of keys. The first row of the same table completes the asymmetry. With the dedupe off, the result of at-most-once delivery **does not change at all**: 2162 rows, 10 polluted cells, the same deviations. No stage of the pipeline repairs the loss, because there is nothing to repair — the lost row leaves no trace anywhere. A duplicated row, on the other hand, does leave a trace: it is an extra record, it has a key, it can be counted and deleted. The backbone of the lesson is this sentence: **loss is silent, duplication is repairable**. Duplication is noisy; it both fires a check and can be undone by a stage. Loss is noiseless and cannot be undone; the only way to see it is **to know the number of rows sent**, that is to keep a counter outside the pipeline. This is why at-least-once is preferred: accepting duplication and making the write resistant to repetition is cheaper than accepting loss and searching for it. #### Summary - In at-most-once delivery 19 rows of period 2026-07 are lost, consumption falls from 2200 to 2162, 10 of the 55 cells are polluted, and all five values served shift (−9.18 … −28.36). - In at-least-once delivery 19 rows arrive twice, the dedupe stage drops the second copy, and the consumption, the polluted cells, and the serving difference stay at their baseline values. - No check catches the loss: the row ratio of 0.0079 stays below the 0.05 tolerance and the largest cell deviation of 0.2488 stays below the 0.25 tolerance. The uniqueness check catches the duplication at the entrance of the pipeline. - When the loss share is swept, the silence turns out to be a band: the distribution check fires only at 0.20, and the row count check fires at no share, because one period is only 0.0833 of the source. When the whole period is lost two checks fire at once — total loss is noisier than partial loss. - When the dedupe is turned off, the duplication stops being harmless (consumption 2219, serving upward in every region); the result of the loss does not change at all. The cost of exactly-once sits in the same place: 200 keys per period, 2400 per run. - Loss is silent, duplication is repairable: the duplication has a trace and can be deleted, the loss has no trace and can be seen only with a counter kept outside the pipeline. #### Next Step In five lessons the data always flowed in **one direction**: from the source system to the pipeline, from the pipeline to the summary, from the summary to the serving. Every measure assumed this direction — the ingestion method read the source, the delivery mode carried the row, the window decision gathered it in a period. One question was not asked: what happens when the value the pipeline produces **returns**. A number in the serving layer often does not stay there; it is written to an operational system or to an application that reads in the field, because that is where it does its real work. At the moment that write touches the pipeline's own input, the arrow closes into a circle. The next lesson measures the circle: after how many stages the value written back returns to its own input, and what it does to the number in the next run. ### Reverse ETL Source: https://academia.sh/en/courses/data-pipelines/reverse-etl Counting the circle that opens when processed data is written back to an operational system: when the regional threshold computed in the serving layer is written back, the chain returns to its own input after seven stages through ingest, screen, dedupe, consumption, summary, serve and write_back. What the written-back value does in the next run is measured: the threshold written by the first run clips 1129 readings in the second run, the serving shifts in every run (the largest difference between runs is 47.00, then 10.90 and 14.36), and the loop stops at the fourth run. Where it stops is not the baseline: while the consumption rows stay at 2200, 55 of the 55 summary cells are polluted, the serving deviates by −19.27 … −59.46, and because the largest cell deviation of 0.1788 stays below the 0.25 threshold, no check fires. When the written-back value is kept apart from the pipeline's input, every run gives the same result. In the previous five lessons the data always flowed in one direction: from the source system to the pipeline, from the pipeline to the summary, from the summary to the serving. This direction was the silent assumption of every measure. Yet a number in the serving layer often does not stay there. The expected consumption per region, the risk class of a subscriber, or the threshold of a region does its real work not on a dashboard but in the **field**: the application that takes the readings uses it, billing applies it, the alerting mechanism looks at it. Writing processed data back to an operational system is called **reverse ETL**, and it is one of the most legitimate outputs of a pipeline. Being legitimate does not mean being harmless. If the written-back value changes the behavior of the operational system, the record that system produces changes too; and that record is the pipeline's **input**. At this point the arrow closes into a circle. This lesson measures the circle with two numbers: after how many stages it closes, and what the next run gives after it has closed. - **DI57.** The code is **fictional**; no operational system, cloud service, vendor, or orchestrator is called. The source is the fictional meter readings of the Data Engineering and MLOps curriculum: 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DI58.** Reverse ETL is modeled with a single value: the **expected consumption threshold per region**. It is computed in the serving layer, written to the operational system, and its factor is 1.2. - **DI59.** What the operational system does with that threshold is fictional but ordinary: the field application **does not record** an increment above the threshold, adds the threshold to the previous index, and leaves the reading to be verified. This is the value that gets recorded. - **DI60.** The chain is written as a graph. **The graph algorithm is not repeated**; it was measured in the Graphs topic of the Data Structures course. Here the chain is only **walked** and its length is counted. - **DI61.** The source itself does not change between runs. The only thing that changes is the index that the operational system **records**; the real meter readings are fixed. - **DI62.** The check thresholds are the same as before: 0.05 for the row count tolerance and 0.25 for the distribution tolerance. - **DI63.** The units counted are the stages in the cycle, clipped readings, consumption rows, polluted summary cells, the serving deviation, and the **difference between runs**. No duration is written. - **DI64.** The rule of the course is tested directly in this lesson: the number of a pipeline is that it gives **the same result** when it is run again. #### The Length of the Circle The block below sets up the common definition and adds three things to it: `cycle`, which walks the chain, `ingest`, which produces the record of the operational system, and `write_back`, which writes the threshold from the serving layer. Then it runs the pipeline five times; each run starts with the threshold written by the previous one. ```python # FICTIONAL. No real operational system , cloud service or orchestrator # is called; the meter readings of the measurement network are fictional , seed 20260218. 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(): # one reading per subscriber per period 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 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 out.append(dict(k, m3=int(k["index"]) - int(prev["index"]))) 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): # per-region mean over the periods 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 CHAIN = {"ingest": "screen", "screen": "dedupe", "dedupe": "consumption", "consumption": "summary", "summary": "serve", "serve": "write_back", "write_back": "ingest"} def cycle(start): """Walks the chain and collects the stages passed until it returns to the start.""" path, nxt = [start], CHAIN[start] while nxt != start: path.append(nxt) nxt = CHAIN[nxt] return path def ingest(raw, threshold): """FICTIONAL operational system: does not record an increment above the threshold , adds the threshold instead.""" records, last, clipped = [], {}, 0 for k in raw: e, prev, s = k["index"], last.get(k["subscriber"]), threshold.get(k["region"]) if prev is not None and s is not None and e - prev > s: e = prev + s clipped += 1 last[k["subscriber"]] = e records.append(dict(k, index=e)) return records, clipped def pipeline(records): v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) summ, _ = s4_summary(v) serving, _ = s5_serve(summ) return v, summ, serving def write_back(v, factor=1.2): """Reverse ETL: the expected consumption threshold per region is written to the operational system.""" totals, count = {}, {} for k in v: totals[k["region"]] = totals.get(k["region"], 0) + k["m3"] count[k["region"]] = count.get(k["region"], 0) + 1 return {b: int(round(factor * totals[b] / count[b])) for b in totals} raw = generate_source() path = cycle("ingest") _, BASELINE, BASE_SERVING = pipeline(ingest(raw, {})[0]) print("cycle:", " -> ".join(path), "-> ingest") print("stages in the cycle", len(path), ", baseline consumption 2200 , summary cells", len(BASELINE)) print() print(" run clipped east thr central north east west south vs previous") threshold, prev = {}, None for i in range(1, 6): records, clip = ingest(raw, threshold) v, summ, serving = pipeline(records) f = "-" if prev is None else max(abs(round(serving[b] - prev[b], 2)) for b in serving) print(f"{i:>5} {clip:>9} {threshold.get('east', 0):>10} " + " ".join(f"{serving[b]:>8}" for b in ("central", "north", "east", "west", "south")) + f" {f:>12}") prev, threshold = serving, write_back(v) ``` ``` cycle: ingest -> screen -> dedupe -> consumption -> summary -> serve -> write_back -> ingest stages in the cycle 7 , baseline consumption 2200 , summary cells 55 run clipped east thr central north east west south vs previous 1 0 0 1321.45 1212.55 881.36 833.09 1065.55 - 2 1129 31 1274.45 1177.73 853.45 813.82 1033.09 47.0 3 1207 30 1263.55 1167.45 845.55 813.82 1025.36 10.9 4 1225 30 1263.55 1153.09 845.55 813.82 1025.36 14.36 5 1225 30 1263.55 1153.09 845.55 813.82 1025.36 0.0 ``` #### After Seven Stages, Its Own Input The first line gives the circle. The chain leaves the `ingest` node, passes the five stages of the pipeline, enters the reverse ETL write, and from there returns **to `ingest` again**: **seven stages**. The number itself is not a warning; what is a warning is that the chain is now a **graph cycle**. Before reverse ETL was added, the chain was six nodes and five edges, with a clear start and end. The moment the seventh edge is added, no node is "run first" any more; every node has a predecessor. In practice this means that the pipeline no longer has a definition **independent of the run**. The answer to the question "what is the output of the pipeline" is not a number but a **run number**. The last column of the table counts this: the second run deviates from the first by 47.00, the third from the second by 10.90, the fourth from the third by 14.36. Run with the same source, the same code, and the same seed, the pipeline gives three different answers. The mechanism can be read in the middle columns. The first run goes without a threshold and gives the right answer; at the end it writes a threshold of 31 for the east region to the operational system. In the second run the field application uses this threshold and clips **1129 readings**: increments above the threshold are not recorded. The mean computed from the clipped readings falls, the lower mean produces a smaller threshold, and the smaller threshold clips **1207** readings in the third run. The loop feeds itself and narrows the input a little more with every turn. Pay attention to where the seventh edge sits: not in the code of the pipeline. The pipeline only writes a value and ends there; the clipping decision is inside the operational system. The circle is built between two systems and **appears in the code of neither**. Someone who looks at the graph of the pipeline sees a neat chain of six nodes; someone who looks at the field application sees a reasonable validation rule. Only someone who draws the two at the same time can see the circle. #### Where the Loop Stops The fifth row says that the loop has stopped: the clipped readings settle at 1225 and the difference between runs becomes **0.00**. After the fourth run, when the pipeline is run again, it gives **the same result**. The block below measures what the settled run is and runs the same pipeline in a setup where the write-back does not touch the input. ```python records, clip = ingest(raw, threshold) v, summ, serving = pipeline(records) print("settled run: consumption", len(v), ", dirty cells", sum(1 for a in set(summ) | set(BASELINE) if summ.get(a) != BASELINE.get(a))) print("serving deviation from baseline", {b: round(serving[b] - BASE_SERVING[b], 2) for b in BASE_SERVING}) print("largest summary cell deviation", round(max(abs(summ[a] - BASELINE[a]) / abs(BASELINE[a]) for a in BASELINE), 4), "< threshold 0.25") print() print("the written-back threshold does not touch the input:") print(" run clipped central north east west south vs previous") threshold, prev = {}, None for i in range(1, 4): records, clip = ingest(raw, {}) v, summ, serving = pipeline(records) f = "-" if prev is None else max(abs(round(serving[b] - prev[b], 2)) for b in serving) print(f"{i:>5} {clip:>9} " + " ".join(f"{serving[b]:>8}" for b in ("central", "north", "east", "west", "south")) + f" {f:>12}") prev, threshold = serving, write_back(v) print("threshold written to the operational system", threshold) ``` ``` settled run: consumption 2200 , dirty cells 55 serving deviation from baseline {'central': -57.9, 'north': -59.46, 'east': -35.81, 'west': -19.27, 'south': -40.19} largest summary cell deviation 0.1788 < threshold 0.25 the written-back threshold does not touch the input: run clipped central north east west south vs previous 1 0 1321.45 1212.55 881.36 833.09 1065.55 - 2 0 1321.45 1212.55 881.36 833.09 1065.55 0.0 3 0 1321.45 1212.55 881.36 833.09 1065.55 0.0 threshold written to the operational system {'central': 32, 'north': 32, 'east': 31, 'west': 32, 'south': 32} ``` The first three lines say what the settled run is, and the answer is disturbing. The consumption rows are **2200**, exactly the baseline value; no row has been dropped and none has multiplied. Yet **55** of the 55 summary cells are dirty: every period of every region has been computed wrongly. The serving deviates between −19.27 and −59.46. And no check fires: the row count is complete, the schema is complete, the keys are unique, every region is present in every period, and the largest cell deviation of **0.1788** stays below the 0.25 tolerance. This table also corrects how the rule of the course should be read. After the fourth run, when the pipeline is run again, it gives the same result, so the rule is satisfied — but the result is wrong. The rule is a **necessary** condition, not a sufficient one: a pipeline that cannot be re-run has not been measured, while a pipeline that can be re-run has been measured only **if it does not re-read its own output**. In a looped pipeline, stability is a sign not of correctness but of **convergence**; the pipeline has come to rest at a constraint that it produced itself. The only check that could see this corruption is already in the last column of the first table: a check that compares the serving value of the same period with that of the previous run. That check would fire in the second, third, and fourth runs, because it would see differences of 47.00, 10.90, and 14.36 respectively. After the fifth run it would go **silent**. That is, comparing runs catches a loop only **before it settles**; after it has settled, all the indicators of the pipeline are green and the served value is permanently wrong. A loop is the only form of corruption whose catch window closes by itself. The second table shows the solution, and the solution is not to give up reverse ETL. The threshold is still computed and still written to the operational system — central 32, north 32, east 31, west 32, south 32 — but the field application uses it only to call the verification, not **to change the record**. The raw reading is stored as it is. The clipped readings are **0**, the difference between runs is **0.00**, and all three runs give the baseline. Reverse ETL is an output; the moment it becomes the pipeline's input it ceases to be an output. What makes the distinction is not whether the value is written, but **whether the written-back field is the field the pipeline reads**. #### Summary - Reverse ETL is writing processed data back to an operational system, and if the written-back value affects the record that system produces, the chain of the pipeline closes into a circle: the path leaving the `ingest` node returns to its own input after seven stages. - The threshold written by the first run clips 1129 readings in the second run, 1207 in the third and 1225 in the fourth; with the same source and the same code, the difference between runs comes out as 47.00, 10.90, and 14.36. - The loop stops at the fourth run but the stopping point is not the baseline: while the consumption rows stay at 2200, 55 of the 55 summary cells are polluted and the serving deviates by −19.27 … −59.46. - No check catches this corruption; the row count, schema, uniqueness, and partition completeness are all intact, and the largest cell deviation of 0.1788 stays below the 0.25 tolerance. - A check that compares runs would see the loop in the second, third, and fourth runs and would go silent after the fifth. The catch window of a loop closes by itself. - When the same threshold is computed and written but the raw record is not touched, the clipped readings are 0 and the difference between runs is 0.00: what decides is not that the value is written, but whether the pipeline reads its own output. #### Next Step In six lessons the data was **ingested** into the pipeline. The rows the ingestion method carried and the changes it missed were counted, the call completed under the constraint of the source was measured, the partition a message would come from and what the number of partitions does to ordering were written down, the result a window decision changes for a late-arriving record and the cost of carrying state were derived, the loss and duplication of the three delivery modes were counted separately, and the circle that the write-back opens was measured. What they all have in common is this: **the row entered the pipeline**. What was done to it after it entered was never asked. In all of these lessons the five stages of the pipeline were called like a ready-made box; where which transformation would be done, whether the same computation would run at the source or at the target, how far down the layers the raw row would be kept, and how to write whether a transformation is **correct** were not defined. The next topic does exactly that: it chooses the place of the transformation, separates the layers, and counts which corruption each layer can stop. ## Transformation Counting where, in which layer and with which guarantee the transformation is done: the effect of running the transformation at the source or at the target on the transported rows and the cells read, the tasks that dependency resolution orders in query-based modeling and what happens when a cycle slips through, the rows each of the raw, refined and serving layers keeps and which layer can stop a corruption, the corruption classes caught by data tests and expectation definitions against the residue that none of them catches, and the way applying a rule change to only two periods shifts the report by 16.98 to 29.57 from full application while both runs look internally consistent and trip no check. ### ETL and ELT Source: https://academia.sh/en/courses/data-pipelines/etl-and-elt Counting whether the transformation runs at the source or at the target: when the boundary is moved to the source, the rows transported to the target fall from 2400 to 55 and the transported cells from 14400 to 165, but the total number of cells read by all three setups stays fixed at 33110 and only the identity of the reader changes; the hybrid boundary lowers the transported rows to 2200 while raising the transported cells to 15400; when a reporting rule changes, the number of rules that can be recomputed at the target is 3 for the setup that keeps the raw data, 1 for the hybrid setup and 0 for the setup that transforms at the source, and the last setup has to go back to the source and can recompute only three periods. The first six lessons of this course **ingested** data into the pipeline: it was pulled from the source, queued, delivered and returned in processed form. Their shared assumption can be written in one sentence: data is taken from one place and put in another. None of them asked **what state the data was brought to**. This lesson opens that question with the first decision: **where** the transformation will run. Next to the source, or at the target after the data has been moved there. The two answers are the names of two patterns, and the difference between them is a trade-off: transforming at the source lowers the **transported rows**, transforming at the target **keeps the raw data** and makes recomputation possible. In this lesson the two run side by side. - **TF1.** The source, the pipeline and the split are **fictional**; no real store, query engine or transformation framework is called. The seed is **20260218**, and the measurement does not depend on the run. - **TF2.** **ETL** means extract, transform, load; **ELT** means extract, load, transform. Both are pattern names, not product names; the only difference between them is which side the transformation runs on. - **TF3.** The five stages of the pipeline are split in two by a single **boundary**. The stages before the boundary run on the source side, the rest run at the target. The ETL boundary is **4**, the ELT boundary is **0**. - **TF4.** A cell read is a model: each stage reads the stated number of fields of every row in its input. The number is not a hardware measurement but a measure of the stage's **workload**. - **TF5.** The baseline is the **unsplit pipeline**: there is no boundary decision, and recomputation has not been asked about. - **TF6.** The source is an operational system and keeps only the **last three periods**; the target keeps all twelve periods. - **TF7.** Storage layout, partitioning and cost were measured in the Data Storage Architectures course and are **not repeated**. The question here is not where the row sits but where the transformation runs. - **TF8.** Resolution: the step of the row ratio is **0.0004**. **No ratio is printed over the serving output**, the value itself is written. #### Where the Boundary Sits The five stages of the pipeline are fixed: screening, deduplication, consumption calculation, summary and serving. The only thing that changes is **where this chain is cut**. If the boundary is zero, no stage runs at the source, the raw rows are moved to the target as they are, and all five stages run there. If the boundary is four, four stages run next to the source and only the summary table goes to the target. Choosing the boundary is not an architectural preference but a trade-off between two quantities. The code below counts those two quantities: the rows that pass through the boundary and the cells read by each side. ```python # FICTIONAL. Meter readings, the five-stage pipeline and the split are defined # in the lesson; no real store, query engine or transformation framework 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 # ---- PIPELINE: five stages. Each stage counts the rows it drops. 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 STAGES = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption), ("summary", s4_summary), ("serve", s5_serve)] # ---- Cells read: each stage reads this many fields of every row in its input. FIELDS_READ = {"screen": 5, "dedupe": 2, "consumption": 4, "summary": 3, "serve": 2} def cells(data): """A summary dict carries three values per cell , a record carries its own fields.""" if isinstance(data, dict): return len(data) * 3 return sum(len(k) for k in data) def split_pipeline(records, boundary): """The first `boundary` stages run on the source side , the rest at the target. What crosses the boundary is the output of stage number `boundary`.""" cells_read, data, transported = {"source": 0, "target": 0}, records, records for i, (name, f) in enumerate(STAGES): cells_read["source" if i < boundary else "target"] += len(data) * FIELDS_READ[name] data, _ = f(data) if i + 1 == boundary: transported = data return data, cells_read, transported ``` #### Transported Rows and Cells Read The second block runs the same source with three boundaries. `transported` gives the rows that pass through the boundary, `t.cells` the number of fields those rows carry, and `source` and `target` the cells read by the two sides. The last column states whether the five served values stay the same as the baseline. ```python SOURCE_RECORDS = generate_source() SETUPS = [("target (ELT)", 0), ("hybrid", 3), ("source (ETL)", 4)] BASELINE, _, _ = split_pipeline(SOURCE_RECORDS, 0) print("baseline serving:", BASELINE) print(f"{'setup':<16}{'boundary':>10}{'transported':>13}{'t.cells':>11}" f"{'source':>8}{'target':>7}{'serving':>8}") for name, n in SETUPS: serving, ok, moved = split_pipeline(SOURCE_RECORDS, n) print(f"{name:<16}{n:>10}{len(moved):>13}{cells(moved):>11}{ok['source']:>8}{ok['target']:>7}" f"{'same' if serving == BASELINE else 'changed':>8}") ``` ``` baseline serving: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} setup boundary transported t.cells source target serving target (ELT) 0 2400 14400 0 33110 same hybrid 3 2200 15400 26400 6710 same source (ETL) 4 55 165 33000 110 same ``` The last column is **the same** for all three setups: wherever the boundary is put, the five served values come out identical. The decision has no effect on the output; its whole effect is on **how** the output is produced. The first two columns give the expected trade-off. The setup that transforms at the source transports **55 rows** and **165 cells** to the target; the setup that transforms at the target transports **2400 rows** and **14400 cells**. The ratio in transported rows is 43.6, and this is the only measurable quantity of data carried over the network. The hybrid row shows that this trade-off is not a straight line. When the boundary is moved to three, the transported rows fall from 2400 to **2200**, but the transported cells **rise** from 14400 to **15400**. The reason is that the consumption stage adds the `m3` field to every row: the row count shrinks while the number of fields per row grows. **Lowering the transported rows does not mean lowering the transported data.** The last two columns carry the most important number of the lesson. The total of cells read by the three setups is, in order, 33110, 33110 which is the sum of 26400 and 6710, and 33110 which is the sum of 33000 and 110. **All three are the same.** The boundary decision does not remove the computation work, it only changes **who does it**. Transforming at the source drops a read of 33000 cells onto the operational system; transforming at the target moves the same load to the target. Choosing a setup because it is "cheap" means not seeing where the load lands. The way the two sides absorb this load is not the same either. The source is an operational system and its main job is not running transformations but serving its own application; a read of 33000 cells runs there, sharing resources with that service. The target, by contrast, is set aside only for this job. So two numbers that look equal are not equal: the cells read at the source are stolen from another job, the cells read at the target are not. This difference grows every time the transformation chain grows, because every stage added raises the load of whichever side of the boundary it sits on. Writing the transformation in SQL does not change this decision: the language is one choice, the boundary is a separate one, and the same query can be written on either side. #### The Cost of Recomputation Up to this point there is no measurable advantage between the two setups. The difference appears when the pipeline runs a **second time**. A reporting rule changes: commercial consumption will from now on be reported with a factor of 1.05. Whether the rule change can be applied at the target depends on **which stage's output** sits at the target. ```python RULES = {"commercial factor 1.05": 3, "dedupe key": 1, "unit correction": 0} SOURCE_RETENTION = 3 # the operational source keeps only the last three periods print(f"{'rule':<24}{'required boundary':>18}" + "".join(f"{a:>16}" for a, _ in SETUPS)) for rule, need in RULES.items(): print(f"{rule:<24}{need:>18}" + "".join(f"{('yes' if n <= need else 'no'):>16}" for _, n in SETUPS)) print(f"{'recomputable at target':<24}{'':>18}" + "".join(f"{sum(1 for g in RULES.values() if n <= g):>16}" for _, n in SETUPS)) print() def commercial_rule(boundary): """The target tries to recompute using only the data it already holds.""" _, _, moved = split_pipeline(SOURCE_RECORDS, boundary) if boundary > RULES["commercial factor 1.05"]: return None, "yes", len(SOURCE_RECORDS), SOURCE_RETENTION v = moved if not any("m3" in k for k in v): v, _ = s1_screen(v) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) new = [dict(k, m3=k["m3"] * (1.05 if k["type"] == "commercial" else 1.0)) for k in v] summ, _ = s4_summary(new) serving, _ = s5_serve(summ) return serving, "no", 0, len(PERIODS) print(f"{'setup':<16}{'return to source':>18}{'re-read':>16}{'periods':>8} serving") for name, n in SETUPS: serving, back, reread, periods = commercial_rule(n) print(f"{name:<16}{back:>18}{reread:>16}{periods:>8} {serving if serving else 'not at target'}") print() TRANSFERABLE = len(SOURCE_RECORDS) // len(PERIODS) * SOURCE_RETENTION print("moving the boundary from 4 to 0: history to transport", len(SOURCE_RECORDS), " available from source", TRANSFERABLE, " permanently lost", len(SOURCE_RECORDS) - TRANSFERABLE) ``` ``` rule required boundary target (ELT) hybrid source (ETL) commercial factor 1.05 3 yes yes no dedupe key 1 yes no no unit correction 0 yes no no recomputable at target 3 1 0 setup return to source re-read periods serving target (ELT) no 0 12 {'central': 1357.32, 'north': 1235.57, 'east': 902.13, 'west': 859.22, 'south': 1095.81} hybrid no 0 12 {'central': 1357.32, 'north': 1235.57, 'east': 902.13, 'west': 859.22, 'south': 1095.81} source (ETL) yes 2400 3 not at target moving the boundary from 4 to 0: history to transport 2400 available from source 600 permanently lost 1800 ``` Three rules ask for three different boundaries. The commercial factor touches the consumption row, changing the deduplication key touches the row before deduplication, and the unit correction touches the raw index itself. A target can apply a rule only if the output it holds sits **before** that boundary. The number is this: the rules that can be recomputed at the target are **3** in the ELT setup, **1** in the hybrid setup and **0** in the ETL setup. The second table writes what that means. The first two setups recompute all twelve periods and produce the new serving output without ever going back to the source, transporting **0 rows** again. The third setup holds only 55 summary cells at the target, so it cannot see the commercial and residential distinction; the only way to apply the rule is to go back to the source and re-read **2400 rows**. The source is an operational system and keeps the last three periods: the periods that can be recomputed are **3**, and the remaining **nine periods** cannot be brought back in any form. This is the first concrete counterpart of the rule of the course: a pipeline's number gives the same result when it is rerun, and **a pipeline that cannot be rerun counts as unmeasured**. The setup that transforms at the source works flawlessly in its first run, lowers the transported rows by a factor of 43.6 and produces the correct serving output. The moment it cannot recompute the nine periods, none of those numbers is defensible any longer. The other side of the cost must also be written. Keeping the raw data at the target means holding 2400 rows in two places at once, and the cells read at the target rise from 110 to 33110. The ELT setup does not get the right to recompute for free; it buys it by **reading 301 times more cells at the target**. #### Moving the Boundary Later The assumption that the boundary is chosen once and left alone is also wrong. As the target's capacity grows, or as the need for recomputation appears, the boundary is wanted not toward the source but toward the target. The last line at the end of the block writes the cost of this move. Taking the boundary from four to zero requires bringing to the target the raw rows that have never sat there: **2400 rows**. Of these, the ones that can come from the source are only the rows of the three retained periods, that is **600**. The remaining **1800 rows** sit nowhere and cannot be brought back in any form. After the boundary is moved, the pipeline can run forward with the new rules, but recomputing the nine periods of history is now permanently impossible. The conclusion that follows is sharper than where the boundary should go: **the boundary decision is the one decision that cannot be corrected retroactively.** A wrong factor can be corrected, a dropped row can be taken again, a faulty check can be replaced. A raw row that was not stored does not come back, and with it goes the rerunnability of every number belonging to those periods. #### Summary - A transformation decision consists of **where the chain of five stages is cut**. The ETL boundary is 4, the ELT boundary is 0, and all three setups produce exactly the same serving output. - Transforming at the source carries **55 rows and 165 cells** to the target, transforming at the target carries **2400 rows and 14400 cells**; the ratio in transported rows is 43.6. - Lowering the transported rows does not lower the transported data: the hybrid boundary brings the rows down to 2200 while raising the transported cells to 15400, because the consumption stage adds a field to every row. - The total of cells read by the three setups is **fixed at 33110**. The boundary decision does not remove the work, it only moves it between the source and the target. - When a rule changes, the number of rules that can be recomputed at the target is **3, 1 and 0**; the setup that transforms at the source has to go back to the source, and because the source keeps three periods, **nine periods** cannot be brought back. - The cost of keeping the raw data at the target is the cells read at the target rising from 110 to 33110; the right to recompute is bought with that cost. #### Next Step All three setups in this lesson shared a single assumption: the transformation is a chain of five stages whose order is fixed and which can be written by hand. A real transformation layer is not a single chain but dozens of separate outputs that refer to one another, and which of them runs before which is written in a hand-kept list. That list breaks every time a new output is added. The next lesson writes the transformations as **models** that refer to one another, derives the dependency from the reference itself and puts two numbers side by side: the tasks that dependency resolution orders and the **dependencies missed** by the hand-kept order. ### Transformation Frameworks Source: https://academia.sh/en/courses/data-pipelines/transformation-frameworks Counting how transformations are written as models that refer to one another and where the order comes from: over eleven models and thirteen declared dependencies the resolver order gives 0 violations, while the hand-kept write order violates 1 dependency and runs 7 models with missing input, and the fixed list violates 5 dependencies and runs 10 models with missing input; the violating run produces no error and lowers the serving output in every region by 17.46 to 28.37; when one more reference is added to a query text, the resolver orders only 3 of the 11 models and does not start the run, while the manual order runs all eleven and silently breaks one edge of the cycle; because a single undeclared read never appears in the graph, the resolver reports 0 violations while the actual violation is 1. The previous lesson asked where the transformation would run and put three setups side by side. All three shared a single assumption: the transformation is a chain of five stages, its order is fixed, and one person can keep it in mind. A transformation layer that accumulates at a target is not a single chain; it is dozens of separate outputs that refer to one another, and some of them never connect to the serving output at all. This lesson writes those outputs as **models**. A model is a single output that has a name, and its definition is a query. Two things are measured: the tasks that dependency resolution orders and the **dependencies missed by a hand-kept order**. In this lesson the cost of the second shows up not as an error but as a wrong number produced without any error. - **TF9.** The model registry, the query texts and the resolver are **fictional**; no transformation framework, query engine or data warehouse product is called. The seed is **20260218**. - **TF10.** A model's dependency is **derived** from the `ref(...)` calls in its query text; it is not kept in a separate list. Writing the reference and declaring the dependency are one and the same act. - **TF11.** The models are the nodes of a **directed acyclic graph**. The topological sorting algorithm was measured in the Graphs topic of the Data Structures course in the Computer Science curriculum and is **not repeated**; here the graph is built and used. - **TF12.** The measured run is **the first run**: no table has been written before. A model that is read before its turn reads an **empty table** and raises no error. - **TF13.** The baseline is **the hand-kept order**: no dependency resolution, no violation counter, no cycle check. - **TF14.** The producer has re-read the commercial readings of the last period and found the indexes 12 too high. This is not a corruption but an ordinary **producer correction**. - **TF15.** The units counted are the task, the violated dependency and the **affected model**. An affected model is the model that runs with missing input together with every model reachable from it. - **TF16.** Resolution: models **11**, declared dependencies **13**, serving **5 values**. **No ratio is printed over the serving output**, the value itself is written. #### A Model's Dependency Lives in Its Own Text The difference between a hand-kept order list and a resolved graph is **where the dependency is written**. In a list, the order is a separate document: the query sits in one place, the order in another, and keeping the two consistent is a human job. In reference-based writing the dependency is **inside** the query: a model reading another model is the same thing as the model naming it with `ref(...)`. The graph is derived from these names, so the moment the query changes, the graph changes with it. The first block sets up the fictional pipeline of the previous lessons and adds ordering on top of it. ```python # FICTIONAL. Meter readings, the five-stage pipeline and the model registry are defined # in the lesson; no real transformation framework, query engine or data warehouse is called. import re 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 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 ``` Eleven models sit in a dictionary. `correction` is the newest addition: the producer has re-read the commercial readings of the last period, published the corrected indexes as a separate output, and `consumption` now reads it as well. ```python QUERY = { "raw_reading": "select * from source", "screened": "select * from ref('raw_reading') where required fields are filled", "deduplicated": "select * from ref('screened') group by subscriber , period", "correction": "select subscriber , period , new_index from ref('screened')", "consumption": "select * from ref('deduplicated') left join ref('correction')", "region_period": "select region , period , sum(m3) from ref('consumption')", "region_serving": "select region , avg(total) from ref('region_period')", "type_period": "select type , period , sum(m3) from ref('consumption')", "region_type": "select * from ref('region_period') join ref('type_period')", "subscriber_yearly": "select subscriber , sum(m3) from ref('consumption')", "freshness": "select count(*) from ref('raw_reading') , ref('region_serving')", } REF = re.compile(r"ref\('(\w+)'\)") def build_graph(query): """Edges are derived from the references in the query text , not from a separate list.""" nodes = list(query) edges = [(o, m) for m, s in query.items() for o in dict.fromkeys(REF.findall(s))] return nodes, edges def violations(order, edges): """Which model that must run first runs later in the given order.""" position = {u: i for i, u in enumerate(order)} return [(a, b) for a, b in edges if position[a] > position[b]] def affected_models(edges, broken): """The model that runs with missing input: the target of the violation and everything reachable from it.""" successors = {} for a, b in edges: successors.setdefault(a, []).append(b) stack = [b for _, b in broken] closed = set(stack) while stack: u = stack.pop() for v in successors.get(u, []): if v not in closed: closed.add(v) stack.append(v) return closed NODES, EDGES = build_graph(QUERY) ORDER, _ = order_tasks(NODES, EDGES) WRITE_ORDER = ["raw_reading", "screened", "deduplicated", "consumption", "region_period", "region_serving", "freshness", "subscriber_yearly", "type_period", "region_type", "correction"] # a fixed hand-kept list that has no relation to the dependencies FIXED_LIST = ["subscriber_yearly", "screened", "region_period", "region_serving", "region_type", "correction", "raw_reading", "freshness", "deduplicated", "type_period", "consumption"] ORDERS = [("resolver", ORDER), ("manual , write order", WRITE_ORDER), ("manual , fixed list", FIXED_LIST)] print("model:", len(NODES), " declared dependencies:", len(EDGES)) print("resolver order:", ORDER) print() print(f"{'order':<24}{'model':>7}{'broken':>7}{'affected':>9} model causing the violation") for name, s in ORDERS: iv = violations(s, EDGES) print(f"{name:<24}{len(s):>7}{len(iv):>7}{len(affected_models(EDGES, iv)):>9} {[a for a, _ in iv]}") ``` ``` model: 11 declared dependencies: 13 resolver order: ['raw_reading', 'screened', 'deduplicated', 'correction', 'consumption', 'region_period', 'type_period', 'subscriber_yearly', 'region_serving', 'region_type', 'freshness'] order model broken affected model causing the violation resolver 11 0 0 [] manual , write order 11 1 7 ['correction'] manual , fixed list 11 5 10 ['raw_reading', 'consumption', 'consumption', 'type_period', 'consumption'] ``` The resolver orders the eleven models and violates none of the thirteen dependencies. The interesting part is the other two rows. The write order is the order in which the models were added to the repository, and until `correction` was added it was **flawless**: the order of ten models was correct. The new model was written to the end of the list, because new things are written there. `consumption`, which reads it, sits in fourth place in the list. That is **a single violation**, and that single violation makes **seven** of the eleven models run with missing input. The fixed list is worse and more honest: it violates five dependencies and breaks ten models. Because the source read itself runs after `screened`, which falls to second place, the whole chain is left idle. The difference between the two rows is not the number of violations but the **visibility** of the violation. #### What Happens When the Order Breaks The violation count is a counter; the real question is what the run produces. The third block actually runs every order. The table of a model whose turn has not come has not been written yet, so the model that reads it reads an **empty table** and raises no error. ```python def total(records, field, paired=True): t = {} for k in records: key = (k[field], k["period"]) if paired else k[field] t[key] = t.get(key, 0) + k["m3"] return t def share(region_period, type_period): """The region's share of the period total.""" period_total = {} for (t, p), v in type_period.items(): period_total[p] = period_total.get(p, 0) + v return {(b, p): round(v / period_total[p], 4) for (b, p), v in region_period.items() if period_total.get(p)} # The producer re-read the last period's commercial readings: the index came out 12 higher. FUNCS = { "raw_reading": lambda d: generate_source(), "screened": lambda d: s1_screen(d["raw_reading"])[0], "deduplicated": lambda d: s2_dedupe(d["screened"])[0], "correction": lambda d: {(k["subscriber"], k["period"]): k["index"] + 12 for k in d["screened"] if k["period"] == "2026-12" and k["type"] == "commercial"}, "consumption": lambda d: s3_consumption([dict(k, index=d["correction"].get( (k["subscriber"], k["period"]), k["index"])) for k in d["deduplicated"]])[0], "region_period": lambda d: s4_summary(d["consumption"])[0], "region_serving": lambda d: s5_serve(d["region_period"])[0], "type_period": lambda d: total(d["consumption"], "type"), "subscriber_yearly": lambda d: total(d["consumption"], "subscriber", paired=False), "region_type": lambda d: share(d["region_period"], d["type_period"]), "freshness": lambda d: {"raw": len(d["raw_reading"]), "region": len(d["region_serving"])}, } LISTS = ("raw_reading", "screened", "deduplicated", "consumption") EMPTY = {name: ([] if name in LISTS else {}) for name in QUERY} def run(order): """First run: a model read before its turn reads an empty table.""" store = dict(EMPTY) for name in order: store[name] = FUNCS[name](store) return store for name, s in ORDERS: print(f"{name:<24}{run(s)['region_serving']}") ``` ``` resolver {'central': 1349.82, 'north': 1232.18, 'east': 898.82, 'west': 853.82, 'south': 1090.64} manual , write order {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} manual , fixed list {} ``` The third row is the expected failure: the fixed list produces no serving output at all. An empty dictionary misleads no one; the first person to look sees that something went wrong. The second row is the real finding of the lesson. The write order produces a serving output that is **complete for all five regions**. The values are in a plausible range, the relations between regions are right, no field is empty, and the run completed without error. The only problem is that the numbers are wrong: because the `correction` table is written after `consumption`, the corrected indexes never enter the calculation. The difference from the correct serving output is between 17.46 and 28.37 in every region. This is the form the course's third claim takes in this lesson: **looking at the output is not enough.** The wrong order did not crash, did not warn, did not leave a gap; it silently used the old data. The source of the difference is a single model, but its effect is not limited to a single model. The number of affected models is seven: because `consumption` ran with missing input, the region summary, the region serving output, the type summary, the region-type share, the subscriber yearly and the freshness report that are fed from it also carry the old number. The cost of missing one dependency is **everything that sits below the missed dependency**. #### Two Things the Resolver Cannot See The resolver derives the dependency from the query, so it cannot know more than the query says. Two cases show the limit: the case where the query says **too much** and the case where it says **too little**. ```python CYCLIC = dict(QUERY, correction=QUERY["correction"] + " where region in ref('region_serving')") D2, K2 = build_graph(CYCLIC) S2, _ = order_tasks(D2, K2) print("cyclic record model:", len(D2), " dependencies:", len(K2)) print("resolver ordered:", len(S2), " unordered:", len(D2) - len(S2), " ordered tasks:", S2) print("manual order running models:", len(WRITE_ORDER), " broken edges:", len(violations(WRITE_ORDER, K2))) print() ACTUAL = EDGES + [("region_serving", "subscriber_yearly")] print("declared dependencies:", len(EDGES), " actual dependencies:", len(ACTUAL)) print("violation reported by the resolver:", len(violations(ORDER, EDGES)), " actual violation:", len(violations(ORDER, ACTUAL)), " affected model:", len(affected_models(ACTUAL, violations(ORDER, ACTUAL)))) ``` ``` cyclic record model: 11 dependencies: 14 resolver ordered: 3 unordered: 8 ordered tasks: ['raw_reading', 'screened', 'deduplicated'] manual order running models: 11 broken edges: 1 declared dependencies: 13 actual dependencies: 14 violation reported by the resolver: 0 actual violation: 1 affected model: 1 ``` In the first case, a condition that looks at the published region serving output has been added to the `correction` model. This is an ordinary request and a single reference, but it puts a cycle into the graph: the region serving output comes from consumption, consumption comes from the correction, and the correction comes from the region serving output. The resolver can order only **three** of the eleven models, **eight** cannot be ordered, and the run never starts. The hand-kept order, on the other hand, runs all eleven models and passes over one edge of the cycle as **a broken edge count of 1**, writing it nowhere. **A linear list cannot show a cycle as a cycle**; it merely picks one edge of the cycle and breaks it silently. The resolver's refusal to run is not a failure; it is the only visible proof of the cycle. The second case is the resolver's own cost. If a model reads another model's output by the physical name of the table instead of `ref(...)`, that read does **not appear** as a reference in the query text. The actual dependencies are fourteen, the declared ones thirteen. The resolver orders its own graph flawlessly and reports **0 violations**; the actual violation is **1** and the affected model is **1**. The guarantee the resolver produces rests not on the correctness of the graph but on **the completeness of the declaration**. Every undeclared read hides under a green report. #### Summary - A model's dependency lives not in a separate order list but in the `ref(...)` calls in its own query text; because the graph is derived from that text, the graph changes when the query changes. - Over eleven models and thirteen dependencies the resolver gives **0 violations**; the write order violates **1** and the fixed list **5** dependencies, and they run **7** and **10** models with missing input respectively. - A single model added later is enough to break the hand-kept order, because the new model is written to the end of the list but sits higher up in the graph. - A violating run raises no error: it produces a complete serving output for five regions and deviates from the correct serving output by 17.46 to 28.37 in every region. The empty fixed-list run is less dangerous than the run that produces a wrong number. - Adding a single reference to a query can put a cycle into the graph: the resolver orders 3 of the 11 models and does not start the run, while the manual order runs all eleven and silently breaks one edge of the cycle. - The resolver's guarantee rests on the completeness of the declaration: a single undeclared read raises the actual dependencies to 14, the resolver still reports **0 violations**, and the actual violation is 1. #### Next Step The eleven models of this lesson sat on a single plane: all equally durable, all equally queryable. In a real transformation layer these models are split into **layers**, and each layer makes a different promise. The raw layer keeps the source as it is, the refined layer applies the rules, and the serving layer carries only what will be published. The cost of this separation is that the same data is stored more than once; in return, a corruption can be **stopped before it travels downstream**. The next lesson counts the rows each layer keeps and distributes the six corruption classes of the previous course over these three layers to measure in which layer each of them can be stopped. ### Staged Layers Source: https://academia.sh/en/courses/data-pipelines/staged-layers Counting the rows each of the raw, refined and serving layers keeps and the layer in which a corruption can be stopped: the sound pipeline leaves 2400 rows in the raw layer, 2200 rows in the refined layer and 55 summary cells in the serving layer; of the six corruption classes four stop in the raw layer, one in the refined layer and one only in the serving layer, and three of them do not change the served value at all; the three-layer setup stores 4655 rows and stops all six classes, while the two-layer setup stops the same six with 2455 rows but catches the unit change one layer late, and the single-layer setup stores 55 cells, stops only two classes and never sees four. The previous lesson kept eleven models on a single plane: all equally durable, all equally queryable. On that plane, someone looking at the output of a model cannot tell whether the number came from the source or from after four transformations. This is why, in a durable transformation layer, models do not sit on a single plane; they are split into **staging layers** stacked on one another, and each layer makes a different promise. This lesson builds three layers and measures two things: the **rows each layer stores** and **the layer in which a corruption can be stopped**. The second question depends on the first, because a rule can look only at the table it can read. The six corruption classes defined in the previous course are distributed over these three layers. - **TF17.** The layers, the rules and the corruption classes are **fictional**; no real store, table format or data test library is called. The seed is **20260218**, and the measurement does not depend on the run. - **TF18.** Three layers: **raw** stores the source as it is, **refined** applies screening, deduplication and the consumption calculation, **serving** carries the summary and the published value. Each layer leaves its own table, and the next layer reads the previous one. - **TF19.** A rule can run only **if the table of the layer it reads is stored**. When a layer is removed, the rules at its boundary go with it. - **TF20.** The six corruption classes were defined in the Introduction to Data Engineering course and are **not repeated**; here they are used as input. None of them is an attack, all of them are accidents on the producer side, and each corrupts a single period. - **TF21.** The routine that measures detection delay in stages was also measured in that course and is not repeated. The question here is not after how many stages but **in which layer** the corruption is stopped. - **TF22.** The baseline is **the single-layer pipeline**: only the serving layer is stored, no intermediate table, no raw row. - **TF23.** The units counted are the stored row, the summary cell and the **stopped corruption class**. - **TF24.** Resolution: the step of the row ratio is **0.0004**, that of the summary cell **0.0182**. **No ratio is printed over the serving output**, the value itself is written. #### Three Layers, Three Separate Promises The layers are not copies of one another; each becomes able to answer a different question and loses something in return. The **raw layer** stores what comes from the source without changing it: field names, types and duplicated rows stay as they are. In this layer one can look at the name of a field or the type of a value, but nothing can be said about consumption, because consumption has not been calculated yet. The **refined layer** applies screening, deduplication and the consumption calculation. Here there is now one consumption value per row, and one can look at that value's sign, size and distribution. In return, the dropped rows are gone: someone looking at the refined layer cannot see what was dropped and why. The **serving layer** carries only what will be published. Five values and the fifty-five summary cells that produce them. The only thing that can be done in this layer is to compare today's number with yesterday's. ```python # FICTIONAL. Meter readings, layers and rules are defined in the lesson; # no real store, table format or data test library 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 CORRUPTIONS = ["late_arrival", "duplicated", "schema_changed", "unit_changed", "missing_partition", "type_drift"] def corrupt(records, cls, period="2026-07"): """Corrupts a single period , the remaining periods stay intact.""" out, counter = [], 0 for k in records: if k["period"] != period: out.append(k) continue if cls == "late_arrival": if k["subscriber"] <= "A040": counter += 1 continue out.append(k) elif cls == "duplicated": out.append(k) if k["subscriber"] <= "A060": out.append(dict(k)) counter += 1 elif cls == "schema_changed": y = dict(k) y["index_value"] = y.pop("index") counter += 1 out.append(y) elif cls == "unit_changed": y = dict(k) y["index"] = k["index"] * 1000 y["unit"] = "m3" counter += 1 out.append(y) elif cls == "missing_partition": if k["region"] == "north": counter += 1 continue out.append(k) elif cls == "type_drift": y = dict(k) y["index"] = str(k["index"]) counter += 1 out.append(y) return out, counter 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 # ---- THREE LAYERS: each leaves its own table LAYERS = ["raw", "refined", "serving"] def build_layers(records): v = records for f in (s1_screen, s2_dedupe, s3_consumption): v, _ = f(v) summary, _ = s4_summary(v) serving, _ = s5_serve(summary) return {"raw": records, "refined": v, "summary": summary, "serving": serving} ``` The rules sit on the **boundaries** of the layers, and each reads only the table of its own layer. The field, type and key rules read the raw layer. The row count and the sign of consumption read the refined layer; the second **cannot be written** in the raw layer, because there is no consumption there. The distribution rule reads the summary. ```python def d_schema(v, fields=("subscriber", "period", "region", "index", "unit")): return any(any(a not in k for a in fields) for k in v) def d_type(v): return any(not isinstance(k.get("index"), int) for k in v) def d_uniqueness(v): key = [(k["subscriber"], k["period"]) for k in v] return len(key) != len(set(key)) def d_partition_period(v): seen = set((k["region"], k["period"]) for k in v) return seen != set((b, p) for b in REGION for p in PERIODS) def d_negative(v): """Since the index only increases , consumption is never negative on a sound pipeline.""" return any(k["m3"] < 0 for k in v) def d_row_count(v, expected, tolerance=0.05): return abs(len(v) - expected) / expected > tolerance def d_distribution(summary, baseline, tolerance=0.25): for key, v in summary.items(): t = baseline.get(key) if t and abs(v - t) / max(abs(t), 1) > tolerance: return True return False RULES = [ ("schema", "raw", lambda lay, base: d_schema(lay["raw"])), ("type", "raw", lambda lay, base: d_type(lay["raw"])), ("uniqueness", "raw", lambda lay, base: d_uniqueness(lay["raw"])), ("partition_period", "raw", lambda lay, base: d_partition_period(lay["raw"])), ("row_count", "refined", lambda lay, base: d_row_count(lay["refined"], base["rows"])), ("negative", "refined", lambda lay, base: d_negative(lay["refined"])), ("distribution", "serving", lambda lay, base: d_distribution(lay["summary"], base["summary"])), ] def stopper(lay, base, open_layers=LAYERS): """The first layer to stop the corruption and the rule that fires in that layer.""" for layer in LAYERS: if layer not in open_layers: continue for name, k, f in RULES: if k == layer and f(lay, base): return layer, name return "-", "-" SOURCE_RECORDS = generate_source() T = build_layers(SOURCE_RECORDS) BASELINE = {"rows": len(T["refined"]), "summary": T["summary"], "serving": T["serving"]} print("sound pipeline raw:", len(T["raw"]), " refined:", len(T["refined"]), " summary cells:", len(T["summary"]), " serving:", len(T["serving"])) print("rule that fires on the sound pipeline:", stopper(T, BASELINE)[1]) print() print(f"{'corruption':<19}{'raw':>6}{'refined':>11}{'summary':>9}{'serving':>9}" f"{'stopping layer':>16}{'rule':>17}") for s in CORRUPTIONS: corrupted, _ = corrupt(SOURCE_RECORDS, s) lay = build_layers(corrupted) layer, rule = stopper(lay, BASELINE) print(f"{s:<19}{len(lay['raw']):>6}{len(lay['refined']):>11}{len(lay['summary']):>9}" f"{('same' if lay['serving'] == BASELINE['serving'] else 'changed'):>9}{layer:>16}{rule:>17}") ``` ``` sound pipeline raw: 2400 refined: 2200 summary cells: 55 serving: 5 rule that fires on the sound pipeline: - corruption raw refined summary serving stopping layer rule late_arrival 2360 2120 55 changed serving distribution duplicated 2460 2200 55 same raw uniqueness schema_changed 2400 1800 45 changed raw schema unit_changed 2400 2200 55 same refined negative missing_partition 2354 2108 53 changed raw partition_period type_drift 2400 2200 55 same raw type ``` #### In Which Layer the Corruption Stops The sound pipeline leaves **2400 rows**, **2200 rows** and **55 summary cells** in the three layers respectively; no rule fires. The distribution of the corrupted runs, on the other hand, is **four, one and one**. The raw layer stops four classes, and all four for the same reason: `duplicated`, `schema_changed`, `missing_partition` and `type_drift` are visible **in the row itself**. A field is missing, a type has drifted, a key is duplicated, or a region-period pair is absent altogether. Because the raw layer stores the row as it is, it can look at these four. Had a transformation been applied, most of them would have become **invisible**: the evidence of the `type_drift` row is not in the refined layer, because the consumption calculation silently turns the number that became text back into a number and the row is not dropped. The refined layer stops a single class: `unit_changed`. The rule that does this says that consumption cannot be **negative**; because the index only increases, it never fires on the sound pipeline. The rule could not have been written in the raw layer, because in the raw layer there is no such thing as consumption, only the index. This is the clearest proof of why the layer exists: **the refined layer produces a new field, and a new rule can be written over that field.** The serving layer also stops a single class: `late_arrival`. Forty rows never arrived and the refined layer fell to 2120, but the drop did not exceed the five percent tolerance of the row count rule. Only after the corruption has passed both layers and entered the summary does it become visible, and only through the distribution comparison. **By that point the corruption has already contaminated two tables**, and the repair has to cover not only the serving layer but the two layers beneath it. The last columns carry the quiet finding of the lesson. Of the six classes caught, **three** do **not change** the five served values at all: `duplicated`, `unit_changed` and `type_drift`. All three were caught, but none of them could have been noticed by looking at the output. `unit_changed` is the clearest example: 2400 rows in the raw layer, 2200 rows in the refined layer, 55 cells in the summary, and in serving the same five values as the baseline. Only the sign of consumption in the middle layer says that the corruption exists. #### When a Layer Is Removed Three layers are not free. The same data is stored in three different forms, and each layer brings a separate write job, a separate freshness question and a separate ownership. The last block removes the layers one by one and puts the number of stopped corruption classes next to the stored rows. ```python SETUPS = [("three layers", LAYERS), ("two layers", ["raw", "serving"]), ("one layer", ["serving"])] print(f"{'setup':<12}{'rows stored':>16}{'stopped':>12}" f"{'at serving layer':>18} not caught") for name, open_layers in SETUPS: stored = sum(len(T[k]) if k != "serving" else len(T["summary"]) for k in open_layers) stopped, late, missed = 0, 0, [] for s in CORRUPTIONS: corrupted, _ = corrupt(SOURCE_RECORDS, s) layer, _ = stopper(build_layers(corrupted), BASELINE, open_layers) if layer == "-": missed.append(s) else: stopped += 1 late += layer == "serving" print(f"{name:<12}{stored:>16}{stopped:>12}{late:>18} {missed}") ``` ``` setup rows stored stopped at serving layer not caught three layers 4655 6 1 [] two layers 2455 6 2 [] one layer 55 2 2 ['duplicated', 'schema_changed', 'missing_partition', 'type_drift'] ``` The second row gives the real contribution of the refined layer. While the raw and serving layers are stored, all six of the six classes are stopped; the only difference is that `unit_changed` is now left to the distribution rule, that is, it is caught **one layer late**. That is what the refined layer buys, and its cost is **2200 rows**. The value of a layer is not always catching a new corruption class; most of the time it is catching the same class **higher up**. The third row is the baseline and carries the harshest number of the lesson. When only the serving layer is stored, total storage falls to 55 cells, but the stopped corruption classes fall **from six to two**. The four classes that are not caught are written by name in the list. Two of them (`schema_changed`, `missing_partition`) change the served value, so there is a chance that an attentive reader might notice one day. The other two (`duplicated`, `type_drift`) **do not change** the served value at all: in a single-layer pipeline there is nowhere to look for these two corruption classes, because the table that carries the evidence is not stored. This is the layer-level counterpart of the rule of the course. The single-layer pipeline runs, produces numbers, and says nothing in four of the six corruptions. **A pipeline that does not store the table that could show the corruption counts as unmeasured for that corruption.** The cost of the layers, 4655 rows against 55, is exactly the price of this measurability. #### Summary - Three layers make three separate promises: the raw layer stores the row as it is, the refined layer produces the consumption value, the serving layer carries only what will be published. On the sound pipeline they leave **2400 rows, 2200 rows and 55 summary cells** respectively. - A rule can run only if the table of the layer it reads is stored; the rule that looks at the sign of consumption **cannot be written** in the raw layer, because there is no consumption there. - **Four** of the six corruption classes stop in the raw layer, **one** in the refined layer and **one** only in the serving layer; when the last one is stopped, the two tables below it have long been contaminated. - Of the six classes caught, **three** do not change the five served values at all; the only evidence of `unit_changed` is the negative consumption in the refined layer. - When the refined layer is removed, the number of stopped classes does not fall, but `unit_changed` is caught **one layer late**; the 2200-row cost of the layer buys that one layer. - When only the serving layer is stored, storage falls to 55 cells and the stopped classes fall **from six to two**; because two of the four classes that are not caught do not change the served value either, there is nowhere left to look. #### Next Step The rules of this lesson looked at **the rows the pipeline carries**: is there a field, is the type right, is the key unique, is consumption negative. All of them were about the data itself, and none said anything about the transformation itself. A transformation can take correct data and calculate wrongly; an aggregation can be grouped by the wrong key, a factor can be applied to the wrong type, a join can duplicated rows. None of these cases is visible in the input, and none of them trips the rules at the layer boundaries. The next lesson carries the rule from above the data **to above the output of the transformation**: an expectation is written on the output of each transformation, the number of corruption classes it catches is counted, and the residue that no expectation catches is measured. ### Testable Transformations Source: https://academia.sh/en/courses/data-pipelines/testable-transformations Counting what expectations placed on the transformation's own output catch and what they do not: six expectations never fire on the sound pipeline and catch four of the six corruption classes, while duplication and type drift are swallowed by the transformation itself and never appear in the output. The unit change is caught by an expectation on the output at the third stage, and no baseline is needed for it, although the serving stays exactly the same. A rule error that shifts the period label back by one period leaves 2200 rows, 55 cells and five serving values exactly the same, so it trips none of the six expectations; only a unit test on a fixed input catches it. Three lessons wrote down the **place** of the transformation: whether it is done at the source or at the target, in which order it runs, and in which layer it stops. The answer to all three questions was an arrangement, and all three assumed the same thing — that the rule itself is correct and the data arrives in good shape. At no stage was it tested that the transformation **produces the correct result**. This lesson asks that question and puts two separate answer instruments side by side: a unit test that runs on a hand-written **fixed input**, and an expectation placed on the **actual output** of the run. What they catch is different and, more importantly, there is a residue that neither of them can see. - **TF41.** The source, the five-stage pipeline and the corruption classes are **fictional**; no real store, transformation framework or data test library is called. The seed is **20260218**, and the output does not depend on the run. - **TF42.** In this lesson the **transformation** is the composition of the first three stages: `s1_screen`, `s2_dedupe` and consumption from the period difference. Its input is the raw row, its output the consumption row. - **TF43.** An **expectation** is a routine that returns, as true or false, an invariant the output must carry. Here `True` means **broken**. - **TF44.** The unit test routine was measured in the Unit Testing and Test-Driven Development course of the Software Quality and Testing curriculum, and the translation of quality dimensions into checks in the Introduction to Data Engineering course; neither **is repeated**. The difference here is at a single point: those checks looked at the **source**, these look at the **output of the transformation**. - **TF45.** The baseline is **the transformation without expectations**: broken expectations **0**, because no expectation has been written. - **TF46.** The corruption classes are not attacks but accidents on the producer side, and each corrupts **a single period** (`2026-07`). - **TF47.** The consumption ceiling **200** is a physical choice; it is not the 49 known from the source's generator. The one who writes the expectation knows the meter, not the generator. - **TF48.** Resolution: the step of the consumption row ratio is **1/2200 = 0.0005**, the summary **55 cells**, the serving **5 values**. **No ratio is printed over the serving output**, the value itself is written. #### The Condition of a Testable Transformation A transformation is testable only when it is a **pure function**: for the same input it gives the same output on every run, carries no hidden state, does not look at the clock, reads nothing from outside. This condition is the direct counterpart of the rule of the course — a transformation that does not give the same result when rerun has no result to test, because every run makes a different answer defensible. Once the condition is met, two separate kinds of tests become possible, and they look at different questions: | Type | Input | Oracle: where the expected result comes from | Catches | |---|---|---|---| | unit test | hand-written fixed input | hand-computed output | an error in the **rule** | | data test (expectation) | actual output of the run | invariant the output must carry | a corruption of the **data** | The distinction lives in the oracle column. In a unit test the correct answer comes from the person writing the test, and since the input is kept small it can be computed by hand. An expectation has no such answer; nobody can know by hand the correct total of two thousand small rows. An expectation can only write what **must be true of the output itself on every run**: the key is unique, consumption cannot be negative, the cell count is complete. The block below sets up the source, the five stages and the six corruption classes. It is the definition used throughout the lesson. ```python # FICTIONAL. The meter readings of the measurement network, the five-stage pipeline and the corruption # classes are defined in the lesson; no real store, framework or orchestrator 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 CORRUPTIONS = ["late_arrival", "duplicated", "schema_changed", "unit_changed", "missing_partition", "type_drift"] def corrupt(records, cls, period="2026-07"): """Corrupts a single period , the remaining periods stay intact.""" out, counter = [], 0 for k in records: if k["period"] != period: out.append(k) continue if cls == "late_arrival": if k["subscriber"] <= "A040": # 40 subscribers never arrived in this period counter += 1 continue out.append(k) elif cls == "duplicated": out.append(k) if k["subscriber"] <= "A060": # 60 records arrived twice out.append(dict(k)) counter += 1 elif cls == "schema_changed": y = dict(k) y["index_value"] = y.pop("index") # the field name changed counter += 1 out.append(y) elif cls == "unit_changed": y = dict(k) y["index"] = k["index"] * 1000 # liters instead of m3 y["unit"] = "m3" # THE LABEL DID NOT CHANGE , silent corruption counter += 1 out.append(y) elif cls == "missing_partition": if k["region"] == "north": # one region's partition was never written counter += 1 continue out.append(k) elif cls == "type_drift": y = dict(k) y["index"] = str(k["index"]) # the number turned into text counter += 1 out.append(y) return out, counter def s1_screen(records): """A row that does not carry the required fields is dropped.""" 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. If the previous period does not exist , 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 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): """Serving: the period average per region.""" 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 ``` #### Unit Test on a Fixed Input The second block writes an eight-row input and computes the correct output by hand. The input has two subscribers, three periods each, one duplicated row and one row missing a required field; therefore all three stages have to do their work. Next to it a rule is placed that does the same calculation but writes the result with **the label of the previous period**. ```python def transform(records): """From raw row to consumption row: screen , dedupe , period difference.""" v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return v def transform_shifted(records): """THE SAME calculation , the only difference: the result is written with the previous period's label.""" return [dict(k, period=PERIODS[PERIODS.index(k["period"]) - 1]) for k in transform(records)] def row(sub, p, e, type="residential"): return {"subscriber": sub, "period": p, "region": "east", "type": type, "index": e, "unit": "m3"} # Fixed input: eight rows , one duplicate and one row with a missing field. FIXED = [row("T1", "2026-01", 100), row("T1", "2026-02", 112), row("T1", "2026-02", 112), row("T1", "2026-03", 130), row("T2", "2026-01", 500, "commercial"), row("T2", "2026-02", 540, "commercial"), row("T2", "2026-03", 545, "commercial"), {"subscriber": "T3", "period": "2026-02"}] # Hand-computed correct output: the oracle of the test. EXPECTED = [("T1", "2026-02", 12), ("T1", "2026-03", 18), ("T2", "2026-02", 40), ("T2", "2026-03", 5)] def essence(v): return sorted((k["subscriber"], k["period"], k["m3"]) for k in v) def report(name, v): print(f"{name:<11}", " ".join(f"{a}/{d[-2:]}={m}" for a, d, m in essence(v)), " passed:", essence(v) == sorted(EXPECTED)) print("fixed input:", len(FIXED), "rows , expected output:", len(EXPECTED), "rows") report("correct", transform(FIXED)) report("shifted", transform_shifted(FIXED)) print("same input run twice , is the output the same:", essence(transform(FIXED)) == essence(transform(FIXED))) ``` ``` fixed input: 8 rows , expected output: 4 rows correct T1/02=12 T1/03=18 T2/02=40 T2/03=5 passed: True shifted T1/01=12 T1/02=18 T2/01=40 T2/02=5 passed: False same input run twice , is the output the same: True ``` Eight rows fall to four: the duplicated row in the `s2_dedupe` stage, the row with the missing field in the `s1_screen` stage, and the first period of each subscriber by the definition of consumption. All four numbers can be verified by hand: for T1, 112 − 100 = 12 and 130 − 112 = 18; for T2, 540 − 500 = 40 and 545 − 540 = 5. The shifted rule produces **the same four numbers** and fails the test immediately, because the labels have slipped back by one period. This is the strength of the test: the calculation itself is right, only the period it is written to is wrong, and this distinction is visible to the eye on a fixed input. The last line verifies the rule of the course on the fixed input — when the same input is run twice, the output is exactly the same. #### Expectations on the Output The third block defines six expectations. All of them look at the **transformation's own output**; none goes back to the source and none needs a correct result known in advance. The expectations are run on the six corruption classes one by one. ```python import statistics # All the expectations look at the transformation's OWN output , not at the source. ROWS, CELLS, CEILING = 2200, 55, 200 def e_key_unique(v, sm): key = [(k["subscriber"], k["period"]) for k in v] return len(key) != len(set(key)) def e_no_negative(v, sm): return any(k["m3"] < 0 for k in v) def e_ceiling(v, sm): return any(k["m3"] > CEILING for k in v) def e_row_ratio(v, sm): return abs(len(v) - ROWS) / ROWS > 0.02 def e_cell_completeness(v, sm): return len(sm) != CELLS def e_volatility(v, sm): """A region's period cells must not stray far from their own median.""" for b in REGION: vals = [x for (bb, _), x in sm.items() if bb == b] median = statistics.median(vals) if vals else 0 if not vals or any(abs(x - median) / abs(median) > 0.30 for x in vals): return True return False EXPECTATIONS = [("key_unique", e_key_unique), ("no_negative", e_no_negative), ("ceiling", e_ceiling), ("row_ratio", e_row_ratio), ("cell_completeness", e_cell_completeness), ("volatility", e_volatility)] def run_checks(v): """Names of the broken expectations , summary and serving.""" sm = s4_summary(v)[0] return [a for a, f in EXPECTATIONS if f(v, sm)], sm, s5_serve(sm)[0] SOURCE = generate_source() CLEAN, _, SERVING = run_checks(transform(SOURCE)) print("expectations:", len(EXPECTATIONS), " broken on the sound pipeline:", len(CLEAN)) print(f"{'corruption':<19}{'rows':>6}{'cells':>7}{'broken expectations':>34}{'serving':>9}") for s in CORRUPTIONS: v = transform(corrupt(SOURCE, s)[0]) bz, sm, sn = run_checks(v) print(f"{s:<19}{len(v):>6}{len(sm):>7}{','.join(bz) or 'none':>34}" f"{'same' if sn == SERVING else 'changed':>9}") ``` ``` expectations: 6 broken on the sound pipeline: 0 corruption rows cells broken expectations serving late_arrival 2120 55 row_ratio,volatility changed duplicated 2200 55 none same schema_changed 1800 45 row_ratio,cell_completeness changed unit_changed 2200 55 no_negative,ceiling,volatility same missing_partition 2108 53 row_ratio,cell_completeness changed type_drift 2200 55 none same ``` The first line shows there is no cost: on the sound pipeline none of the six expectations fires, so no good row is wrongly blocked. **Six expectations catch four of the six classes.** The `unit_changed` row is the most important number of this lesson. In the Introduction to Data Engineering course the same corruption was visible only through the distribution check that looks at the summary of the fourth stage; here the `no_negative` and `ceiling` expectations catch it **in the output of the transformation**, at the third stage. The difference is not merely one stage: the distribution check needs **a sound baseline summary** to work, these three expectations need no baseline at all. A newly built pipeline has no history to compare against, whereas an invariant can be written on the first run. The serving column, on the other hand, still says **same** for `unit_changed` — although the five served values stay exactly in place, two periods are garbage, and the only thing that says so is the expectation. The remaining three catches are ordinary: `late_arrival` and `missing_partition` lower the row count and break the cell band, while `schema_changed` makes the whole period vanish at the `s1_screen` stage and leaves 1800 rows and 45 cells. #### The Residue No Expectation Sees Two lines in the table are empty: `duplicated` and `type_drift`. The fourth block writes two rules next to them — one faulty, one deliberate — and runs the same six expectations on them. ```python SHIFTED = transform_shifted(SOURCE) FACTORED = [dict(k, m3=k["m3"] * (1.05 if k["type"] == "commercial" else 1.0)) for k in transform(SOURCE)] for name, v in (("shifted rule", SHIFTED), ("factor rule", FACTORED)): bz, sm, sn = run_checks(v) print(f"{name:<17} rows={len(v)} cells={len(sm)} broken={','.join(bz) or 'none'}") print(" serving:", sn) print(" baseline:", SERVING) ``` ``` shifted rule rows=2200 cells=55 broken=none serving: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} factor rule rows=2200 cells=55 broken=none serving: {'central': 1357.32, 'north': 1235.57, 'east': 902.13, 'west': 859.22, 'south': 1095.81} baseline: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} ``` The residue has three parts, and the mechanism of each is separate. **The two classes the transformation swallows.** `duplicated` is repaired in the `s2_dedupe` stage and `type_drift` in the type conversion; an expectation looking at the output cannot tell a repaired corruption from a corruption that never happened. The emptiness of these two lines is proof of the transformation's soundness, but nobody learns that the producer is still sending text, and the text of a coming period may not be convertible to a number. **An expectation placed on the output does not take the place of a check placed on the input**; the two look at different questions. **The label-shifting rule.** The shifted rule produces 2200 rows and 55 cells, breaks no expectation and leaves the serving **exactly the same**. The reason is structural: the set of calculated numbers has not changed at all, only the periods they are written to have slipped; the average of the eleven cells per region is the average of the same numbers. No invariant that can be written about the output itself can see this, because the output itself is flawless — what is wrong is **which fact it corresponds to**. The only thing that catches this error is the fixed-input test at the start of the lesson. **The deliberate rule change.** The factor rule reports commercial consumption multiplied by 1.05. None of the six expectations breaks, because there is no corruption; the serving, however, moves up in all five values. Expectations cannot say whether this difference is **a wanted change or an accident**. The only thing that makes the distinction is writing the rule version to the record. #### Summary - A transformation is testable only if it is a pure function; a rule that does not give the same output for the same input on every run has no result to test. - A unit test catches an error in the rule, an expectation catches a corruption of the data; the distinction lies in where the oracle comes from, and the two do not take each other's place. - On the sound pipeline none of the six expectations fires, and they catch four of the six corruption classes; `duplicated` and `type_drift` are swallowed by the transformation itself and never appear in the output. - `unit_changed` is caught by an expectation on the output at the third stage and **without any baseline**, although the five served values stay exactly the same. - The rule that shifts the period label back by one period produces 2200 rows, 55 cells and the same five serving values, so it trips none of the six expectations; only the unit test with a fixed input sees it. #### Next Step The factor rule of the last block opens the question this lesson cannot close. The rule changed and the new runs will produce the correct result — but eleven of the twelve periods stand computed with the **old rule**. Recomputing all the periods means run cost, while recomputing only a few periods leaves one part of the report with the new rule and the rest with the old one. The next lesson measures backfill over these two options: how many periods are recomputed, how far the mixed report deviates from the full application, and which checks that report passes. ### Backfill Source: https://academia.sh/en/courses/data-pipelines/backfill Counting the application of a changed rule to past periods: when the rule is applied to all twelve periods, 12 periods and 2200 rows are recomputed, while applying it to only two periods takes 2 periods and 400 rows, but the report deviates from the full application by 16.98 to 29.57. The mixed report breaks none of the six expectations, comes out internally consistent, and trips no check, because the 2.75% jump the rule creates in the central cells stays below the median of the natural period volatility. The source of the deviation is not the data but which periods were recomputed; when the row is stamped with the rule version, the mixing becomes visible with a single comparison. The last block of the previous lesson wrote a rule that reports commercial consumption with a factor of 1.05, and none of the six expectations broke. The expectations were right, because there was no corruption: the rule had changed on purpose. The question left open there was this — if the rule is valid from today, what happens to **the eleven periods computed with the old rule**. This lesson measures that question. Recomputing all the periods means run cost; recomputing only a few periods leaves one part of the report with the new rule and the rest with the old one. The second option is cheap, fast, and the report it produces **looks internally consistent**. This is the clearest example of the second claim of this course. - **TF49.** The source and the five-stage pipeline are **fictional**; no real store, transformation framework or orchestrator is called. The seed is **20260218**, and the report does not depend on the run. - **TF50.** **Backfill** is applying a changed rule or corrected data to **past periods** and recomputing those periods. What is measured is which periods are recomputed. - **TF51.** The rule change is not a corruption but a **decision**: commercial consumption will be reported with a factor of 1.05 instead of 1.00. The new rule is taken as correct; what is debated is its scope. - **TF52.** The baseline is **the run without a backfill**: no past period is recomputed, the report stands entirely on the old rule. - **TF53.** Reprocessing and idempotence were measured at the **row level** in the Introduction to Data Engineering course and are **not repeated**; what is measured here is **at the period level**. - **TF54.** The cost is written as **periods** and **rows**, raw time is not written. - **TF55.** Since consumption is computed from the period difference, recomputing a period requires reading also the source of **the previous period**; the window is one period wider at the source. - **TF56.** Resolution: consumption rows **2200**, summary **55 cells**, serving **5 values**. **No ratio is printed over the serving output**, the value itself is written. #### The Scope Question of Backfill When a rule changes there are three options and all three are defensible. **Not backfilling at all** leaves the report consistent but describes everything before today with the wrong rule. **Backfilling everything** carries the whole report to the new rule and has all twelve periods recomputed. **Backfilling partially** is the cheapest and usually has a reason too: only the audited periods, only the quarter still open, only the window for which the data is still at hand. The problem with the partial option is not its cost but its **invisibility**. The resulting report is a combination of two rules, yet none of the rows carries which rule it was computed with. The block below sets up the source, the five stages and the backfill function that applies the rule to the chosen periods. ```python # FICTIONAL. The meter readings of the measurement network and the five-stage pipeline are # defined in the lesson; no real store, framework or orchestrator 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): """A row that does not carry the required fields is dropped.""" 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. If the previous period does not exist, 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 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): """Serving: the period average per region.""" 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 backfill(records, periods, new_rule=True): """The rule changed: commercial consumption will be reported with a factor of 1.05 instead of 1.00. Only the chosen periods are recomputed.""" v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) out = [] for k in v: factor = 1.05 if (new_rule and k["period"] in periods and k["type"] == "commercial") else 1.0 out.append(dict(k, m3=k["m3"] * factor)) sm, _ = s4_summary(out) served, _ = s5_serve(sm) return served ``` #### Three Reports and Two Costs The second block produces three reports from the same source: the rule never applied, applied to all twelve periods, and applied only to the periods `2026-07` and `2026-08`. Then the cost of the two backfill windows is counted. ```python WINDOW = ["2026-07", "2026-08"] SOURCE = generate_source() def diff(a, b): return {k: round(a[k] - b[k], 2) for k in a} OLD = backfill(SOURCE, [], new_rule=False) FULL = backfill(SOURCE, PERIODS) MIXED = backfill(SOURCE, WINDOW) print("old rule :", OLD) print("full application :", FULL) print("two periods :", MIXED) print("mixed - full :", diff(MIXED, FULL)) print("mixed - old :", diff(MIXED, OLD)) def cost(periods): """Recomputed periods , consumption rows written , source periods read and rows.""" i = min(PERIODS.index(d) for d in periods) read = sorted(set(periods) | {PERIODS[max(i - 1, 0)]}) written = 200 * sum(1 for d in periods if d != PERIODS[0]) return len(periods), written, len(read), 200 * len(read) print() print(f"{'window':<13}{'periods':>7}{'written':>9}{'periods read':>14}{'rows read':>14}") for name, pl in (("all periods", PERIODS), ("two periods", WINDOW)): print(f"{name:<13}" + "".join(f"{x:>{w}}" for x, w in zip(cost(pl), (7, 9, 14, 14)))) ``` ``` old rule : {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} full application : {'central': 1357.32, 'north': 1235.57, 'east': 902.13, 'west': 859.22, 'south': 1095.81} two periods : {'central': 1327.75, 'north': 1216.36, 'east': 885.15, 'west': 837.77, 'south': 1070.65} mixed - full : {'central': -29.57, 'north': -19.21, 'east': -16.98, 'west': -21.45, 'south': -25.16} mixed - old : {'central': 6.3, 'north': 3.81, 'east': 3.79, 'west': 4.68, 'south': 5.1} window periods written periods read rows read all periods 12 2200 12 2400 two periods 2 400 3 600 ``` The cost table shows why the partial option is chosen: the two-period window recomputes **2 periods** and **400 rows**, while the full application computes **12 periods** and **2200 rows**. The source-read column is the cost of the dependency between periods: to recompute two periods the source of **three periods** is read, because the `2026-07` consumption rests on the `2026-06` index. The window is always one period wider at the source, and this width is paid at the start of the window. The deviation line is the main number of the lesson. The mixed report deviates from the full application **between −16.98 and −29.57**; none of the five regions stands at the value the full application gives. In return, the mixed report also differs from the old rule — all five values are **between 3.79 and 6.30** higher. So the resulting report is **neither the report of the old rule nor of the new rule**; it is a third thing, and no rule in the real world corresponds to that third thing. The source of the deviation is not the data. All three reports were produced from the same 2400 source rows with the same five stages; no row was lost, no row was corrupted. The only thing that creates the difference is **which periods were recomputed**. #### Which Check the Mixed Report Passes The third block tests the mixed report with the six expectations of the previous lesson, checks whether the report is internally consistent, and compares the jump the rule creates in a cell with the region's **natural period volatility**. This last comparison represents the most optimistic case: what an observer who inspects the report period by period would see if they looked for the jump. ```python import statistics def recompute(records, periods): """The same rule , two differences: the rows are returned and every row is stamped with the rule version.""" v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return [dict(k, rule="v2" if k["period"] in periods else "v1", m3=k["m3"] * (1.05 if (k["period"] in periods and k["type"] == "commercial") else 1.0)) for k in v] def expectations(v): """The six expectations of the previous lesson , on the transformation's own output.""" sm = s4_summary(v)[0] key = [(k["subscriber"], k["period"]) for k in v] d = {"key_unique": len(key) != len(set(key)), "no_negative": any(k["m3"] < 0 for k in v), "ceiling": any(k["m3"] > 200 for k in v), "row_ratio": abs(len(v) - 2200) / 2200 > 0.02, "cell_completeness": len(sm) != 55, "volatility": False} for b in REGION: vals = [x for (bb, _), x in sm.items() if bb == b] if any(abs(x - statistics.median(vals)) / statistics.median(vals) > 0.30 for x in vals): d["volatility"] = True return [a for a, s in d.items() if s], sm BROKEN, SM_MIXED = expectations(recompute(SOURCE, WINDOW)) SM_OLD = s4_summary(recompute(SOURCE, []))[0] print("broken expectations in the mixed report:", BROKEN or "none") print("is the report internally consistent:", all(abs(MIXED[b] - round(sum(x for (bb, _), x in SM_MIXED.items() if bb == b) / 11, 2)) < 0.01 for b in REGION)) for b in ("central", "east"): natural = [abs(SM_OLD[(b, PERIODS[i])] - SM_OLD[(b, PERIODS[i - 1])]) / SM_OLD[(b, PERIODS[i - 1])] for i in range(2, 12)] jump = [(SM_MIXED[(b, d)] - SM_OLD[(b, d)]) / SM_OLD[(b, d)] for d in WINDOW] print(f"{b:<7} natural change median={statistics.median(natural):.4f} " f"largest={max(natural):.4f} rule jump={max(jump):.4f}") ``` ``` broken expectations in the mixed report: none is the report internally consistent: True central natural change median=0.0883 largest=0.1874 rule jump=0.0275 east natural change median=0.0909 largest=0.1541 rule jump=0.0234 ``` Three lines say the same thing: **no check reports this.** None of the six expectations breaks, because there is nothing to break — the row count is 2200, the cell count 55, the keys are unique, no value is negative or above the ceiling. The report is internally consistent: each region's serving value equals the average of its own eleven summary cells. The checks that look at the source also stay silent, because the source has not changed at all. The last two lines eliminate even the most optimistic observer. The largest jump the rule creates in a cell is **2.75%** in the central region and **2.34%** in the east; yet the **median** of the natural period change of the same regions is 8.83% and 9.09%, and the largest 18.74% and 15.41%. The trace of the rule is about a third of even the median of the natural volatility. An observer who looks period by period cannot find the jump even if they look for it, because the jump is inside the noise. That both runs look internally consistent is the core of this lesson. Neither the full application nor the partial application produces an error; both run to the end, both print a clean report. The only difference between them is a list — which periods were recomputed — and that list is written nowhere in the report. #### The Rule Version Stamp The fourth block adds what is missing. `recompute` writes into every row the rule it was computed with; looking at how many versions the report contains is enough. The same block also tests the rule of the course: what happens when the backfill is run twice. ```python V_FULL = recompute(SOURCE, PERIODS) V_MIXED = recompute(SOURCE, WINDOW) for name, v in (("full", V_FULL), ("mixed", V_MIXED)): s = sorted(set(k["rule"] for k in v)) print(f"{name:<8} rule version={str(s):<14} single version in the report:", len(s) == 1) print("run twice from the source:", backfill(SOURCE, WINDOW) == backfill(SOURCE, WINDOW)) patched = [dict(k, m3=k["m3"] * (1.05 if k["period"] in WINDOW and k["type"] == "commercial" else 1.0)) for k in V_MIXED] print("second time on the output:", diff(s5_serve(s4_summary(patched)[0])[0], MIXED)) ``` ``` full rule version=['v2'] single version in the report: True mixed rule version=['v1', 'v2'] single version in the report: False run twice from the source: True second time on the output: {'central': 6.62, 'north': 4.0, 'east': 3.98, 'west': 4.92, 'south': 5.35} ``` The stamp is a single field and reduces the mixing to a single comparison: if the report has two versions, the report is mixed. Its cost is also a single field, a string written to each of the 2200 rows. The moment the stamp is written matters; after the calculation is done, which row was produced with which rule cannot be worked out retroactively, because the only thing left in the row is the result itself. The last two lines show the rule of the course from two sides. When the backfill is computed **from the source**, running it twice does not change the result; the function is **idempotent** and the report is independent of the number of runs. When the same correction is applied **as a patch over the output**, the second run multiplies the factor once more, and the report shifts once more by between 3.98 and 6.62. Both paths aim at the same result, one can be rerun, the other is correct **only once** — and there is no way to tell from the report which of the two was run. #### Summary - Backfill is a scope decision: the full application has 12 periods and 2200 rows recomputed, while the two-period window makes do with 2 periods and 400 rows, but **three periods** are read at the source, because consumption rests on the index of the previous period. - The report of the partial backfill deviates from the full application **between −16.98 and −29.57** and also differs from the old rule by between 3.79 and 6.30; the resulting report belongs to neither of the two rules. - The source of the deviation is not the data but **which periods were recomputed**; all three reports are produced from the same 2400 source rows without losing a single row. - The mixed report breaks none of the six expectations, comes out internally consistent, and the largest jump the rule creates in the central cells (2.75%) stays far below the median of the natural period volatility (8.83%), so it is invisible even to an observer who looks period by period. - When the row is stamped with the rule version, the mixing becomes visible with a single comparison; when the backfill is computed from the source it is idempotent, and when it is applied as a patch over the output the second run shifts the report once more by between 3.98 and 6.62. #### Next Step Eleven lessons built the pipeline: data was ingested through six separate paths, what loss and duplication do to the serving was counted, and where the transformation is done, in which order it runs, in which layer it stops, how it is tested and how it is applied to the past were written. In all of these eleven lessons there was one thing that ran the pipeline: **a start decision a person gave by hand**. Which task will run when, what the tasks behind it will do when a task falls, whether the same job was triggered twice, and **who will write** the rule version stamp of this lesson were never asked. The next topic turns the pipeline into a task graph and measures how far the shape of the graph carries a failure. ## Orchestration and Observability Counting how a workflow with dependencies is built as a graph and what the rest does when one task fails: in a graph of 48 nodes and 47 edges, the consumption task also depending on the screening of the previous period and five tasks being unable to run at all when the ingestion task of the first period fails; how many times and when the time, event and data readiness triggers run the same job; the rows a retry multiplies in a pipeline that is not idempotent and the failure an alert threshold misses; a half-written partition making serving wrong in every region whereas an atomic write never writes the partition and so is caught by the period-aware completeness check; how many periods later the freshness, volume and schema evolution indicators show a corruption; and, in the environment and version management of the pipeline code, the different result the same code produces in two environments. ### Workflow Schedulers Source: https://academia.sh/en/courses/data-pipelines/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. ### Dependency and Triggering Source: https://academia.sh/en/courses/data-pipelines/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. ### Retry and Alerting Source: https://academia.sh/en/courses/data-pipelines/retry-and-alerting Measuring what a task failing in the middle does to serving and whether retrying repairs the pipeline: when the partition of the seventh period is written halfway at 120 rows, the raw rows stay at 2320, both the row count and the period-aware completeness check stay green, yet serving is wrong in every region (between −49.09 and −121.82); with an atomic write the same task leaves no rows, consumption falls from 2200 to 1800, the deviation stays between −3.25 and +12.80 and both checks fire; the write with a three-try retry that appends duplicates 240 rows and lifts the raw count of the store to 2640, with the keyed write the duplication is zero, and when the retry succeeds the task alert goes silent. All the runs in the previous 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. The partial run was harmless there, because the write was keyed and the missing part was completed in the next run. This lesson removes that guarantee. In a real pipeline there is a third outcome: the task runs, starts writing and **fails in the middle.** What is left behind is neither an empty partition nor a complete one. This lesson measures two questions — what a half-written partition does to serving, and whether retrying the failed task repairs the pipeline or duplicates rows. The answer to the first question is the strongest example of the third claim of the course. - **OO17.** The source, the pipeline and the task run are **fictional**; no real orchestrator, store or alerting service is called. The seed is **20260218**, and the measurement does not depend on the run. - **OO18.** An **atomic write** means that a task either writes all of its rows or leaves none behind. In a write that is not atomic, the rows left halfway stay readable. - **OO19.** The task fails at **row 120** while writing the `2026-07` partition. This number is a choice of place; what is measured is not the number itself but the difference between half and nothing. - **OO20.** The retry, backoff, circuit breaker and graceful degradation procedure was measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum and in the Resilience and Reliability course of the System Design and Distributed Systems curriculum and is **not repeated**. The only question here is **whether a retry duplicates rows**. - **OO21.** The term for being **safe to repeat** is `idempotent`. Reprocessing duplicating rows was measured in the Introduction to Data Engineering course; the measurement here shows the same fact at **task level**. - **OO22.** Two alert sources are separated: the **task alert** goes out if the task is still failing after the last try, the **data alert** comes from the checks. - **OO23.** The baseline is the **run without orchestration**: no atomic write, no retry, no alert. - **OO24.** The counted units: rows written, duplicated rows, consumption rows, serving deviation, tries. **No raw duration is written.** No ratio is printed on serving, the value itself is written. #### Failing in the Middle of a Task The task that writes a partition is not a single operation; it writes the rows in order and can fail after any row. What it leaves behind depends on whether the write is atomic. In an atomic write the half-finished work is **never visible** — the task fails, the partition stays empty, and the pipeline reads that period as if it did not exist. In a write that is not atomic, the first 120 rows stay where they are, and as they pass through the pipeline they cannot be told apart from complete rows. The measurement is made with two checks. The **row count check** looks at whether the raw row count deviates from the expected one by more than five percent. The **period-aware completeness check** looks at whether every pair of region and period exists in the data; the Introduction to Data Engineering course had separated this check from its flawed sibling, and the correct one is used here. ```python # FICTIONAL. The meter readings, the five-stage pipeline and the task run are # defined inside the lesson; no real store, orchestrator or alerting service 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) def write_partition(records, period, cutoff=None, atomic=False): """cutoff: the task fails after this many rows. atomic: a half-written partition is not visible.""" target = [k for k in records if k["period"] == period] if cutoff is None or cutoff >= len(target): return target, False partial = target[:cutoff] return ([], True) if atomic else (partial, True) def check_row_count(data, expected=2400, margin=0.05): return abs(len(data) - expected) / expected > margin def check_period_aware_completeness(data): """Every region exists in every period.""" seen = set((k["region"], k["period"]) for k in data) return seen != set((b, d) for b in REGION for d in PERIODS) ``` #### Half Versus Nothing The second block runs the same failure in two write modes and puts four things side by side: the rows written to the partition, the raw rows of the store, the consumption rows and the deviation of serving from the baseline. ```python SOURCE = generate_source() BASELINE_SERVING, BASELINE_N = read_result(SOURCE) OTHERS = [k for k in SOURCE if k["period"] != "2026-07"] print("baseline consumption:", BASELINE_N) print("baseline serving:", BASELINE_SERVING) print(f"{'setup':<16}{'written':>8}{'raw':>6}{'consumption':>13}{'row check':>11}{'completeness':>14}") for name, atomic in (("not atomic", False), ("atomic", True)): written, failed = write_partition(SOURCE, "2026-07", 120, atomic) store = OTHERS + written serving, n = read_result(store) print(f"{name:<16}{len(written):>8}{len(store):>6}{n:>13}" f"{str(check_row_count(store)):>11}{str(check_period_aware_completeness(store)):>14}") print(" deviation:", {b: round(serving[b] - BASELINE_SERVING[b], 2) for b in BASELINE_SERVING}) ``` ``` baseline consumption: 2200 baseline serving: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} setup written raw consumption row check completeness not atomic 120 2320 2040 False False deviation: {'central': -92.63, 'north': -121.82, 'east': -49.09, 'west': -49.27, 'south': -61.55} atomic 0 2200 1800 True True deviation: {'central': -2.89, 'north': 6.12, 'east': -3.25, 'west': 12.8, 'south': 7.01} ``` The first row is the strongest example of the third claim of the course. The half-written partition makes serving wrong in **all five regions**: the deviation is between −49.09 and −121.82, and none of it is small. In return, **both checks are green.** The raw rows are 2320, which deviates from the expected 2400 by 3.33% and stays inside the band of 5%. The period-aware completeness check passes too, because the 120 rows written carry subscribers from all five regions — the `2026-07` period exists in the data for every region, it is only incomplete. **The number looks reasonable, serving is wrong, and no check says so.** Narrowing the band is the first fix that comes to mind, and it brings its own cost. To see this half-write, the band has to go below 3.33%. In the previous lesson four subscribers ended their subscriptions, which means 16 rows over four periods, that is 0.67% of the raw rows; for the band not to fire on that change, it has to stay above 0.67%. **The threshold can be tuned, but only between 0.67% and 3.33%; choosing within that range requires knowing in advance both the size of the error to be caught and the size of the movement to be ignored.** The second row looks worse at first glance. In the atomic write the partition is never written; the store drops to 2200 rows and consumption falls to **1800** instead of 2040, that is, **240 rows fewer** are computed than in the half-write. The reason the loss grows is the graph of the earlier lesson: if `2026-07` does not exist at all, the consumption of `2026-08` cannot be computed either, because there is no index of the previous period. Looking at serving, though, the order reverses. The setup that loses more rows has a deviation between **−3.25 and +12.80**; the one that loses fewer falls as far as −121.82. The reason is this: serving is the **period average** per region. When the partition is never written, the cell of that period does not enter the average at all, and the remaining eleven periods give the correct average. When it is written halfway, the cell enters the average, but it holds one hundred and twenty subscribers instead of two hundred — a low number joins the average as if it were a correct one. Moreover, in the atomic setup **both checks fire.** The raw rows are 2200 and exceed the band of five percent; and because the `2026-07` period is absent from the data for every region, the period-aware completeness check sees the corruption too. The sentence that comes out of this is the backbone of this lesson: **atomicity does not make the answer correct; it makes the failure visible and bounded.** The result is still incomplete and the deviation is still not zero; what is gained is that the incompleteness can be **caught** and its effect does not go beyond one period. #### Retry and Alerting The failed task is retried. The third block tries the task three times — the first two tries fail again at row 120, the third completes — and counts what it changes whether the write appends or is keyed. The `task alert` column carries the task alert: it fires if the task is still failing after the last try. ```python def task_run(records, period, cutoffs, atomic=False, keyed=False): """The task is tried in order; cutoffs give the row at which each attempt fails. None = the attempt succeeds. The write either appends or is keyed.""" store, written, tries, failed = [k for k in records if k["period"] != period], [], 0, False for cutoff in cutoffs: tries += 1 part, failed = write_partition(records, period, cutoff, atomic) if keyed: table = {(k["subscriber"], k["period"]): k for k in written} for k in part: table[(k["subscriber"], k["period"])] = k written = list(table.values()) else: written = written + part if not failed: break return store + written, tries, len(written), failed print(f"{'setup':<24}{'tries':>7}{'written':>8}{'raw':>6}{'consumption':>13}{'north':>9}" f"{'task alert':>12}{'row check':>11}{'completeness':>14}") for name, cutoffs, atomic, keyed in ( ("no retry , not atomic", (120,), False, False), ("no retry , atomic", (120,), True, False), ("three tries , appending", (120, 120, None), False, False), ("three tries , keyed", (120, 120, None), False, True)): store, tries, written, failed = task_run(SOURCE, "2026-07", cutoffs, atomic, keyed) serving, n = read_result(store) print(f"{name:<24}{tries:>7}{written:>8}{len(store):>6}{n:>13}" f"{round(serving['north'] - BASELINE_SERVING['north'], 2):>9}{str(failed):>12}" f"{str(check_row_count(store)):>11}{str(check_period_aware_completeness(store)):>14}") def without_dedupe(records): """The pipeline without a deduplication stage.""" v, _ = s1_screen(records) v, _ = s3_consumption(v) summ, _ = s4_summary(v) return s5_serve(summ)[0], len(v) store, _, _, _ = task_run(SOURCE, "2026-07", (120, 120, None)) print() print("three tries , appending , pipeline without dedupe:", without_dedupe(store)) ``` ``` setup tries written raw consumption north task alert row check completeness no retry , not atomic 1 120 2320 2040 -121.82 True False False no retry , atomic 1 0 2200 1800 6.12 True True True three tries , appending 3 440 2640 2200 0.0 False True False three tries , keyed 3 200 2400 2200 0.0 False False False three tries , appending , pipeline without dedupe: ({'central': 1470.73, 'north': 1320.73, 'east': 1002.64, 'west': 928.55, 'south': 1189.73}, 2440) ``` The third row gives the cost of the retry. The task was tried three times, the two failed tries left 120 rows each in the partition, the successful try added 200 more rows: the partition holds **440 rows** instead of 200, that is, **240 rows were duplicated**. Serving is still correct and the deviation of the north is zero, because the deduplication stage swallows the excess. This behavior itself was measured in the Introduction to Data Engineering course; the new information here is that the source of the duplication is not a corruption but **the own mechanism of orchestration**. The retry repairs the pipeline and damages the store. The last row shows the proof of that mechanism. In the keyed write each try overwrites the same keys; after three tries the partition holds 200 rows, the duplicated rows are **zero**, and the raw rows stay at 2400. What makes the retry safe is not the try itself but **the write being idempotent**. The last output line says how much load the deduplication carries. When the same appending store is passed through a pipeline without a deduplication stage, the consumption becomes 2440 rows instead of 2200 and serving rises in every region: the north is **1320.73** instead of 1212.55, the central is **1470.73** instead of 1321.45. That is, the zero deviation of the third row comes not from the correctness of the pipeline but from the existence of a single stage. The `task alert` column shows where the alert goes silent. In the first two rows the last try fails and the task alert goes out — but that alert only says "the task failed"; it says nothing about the deviation of −121.82 in the first row, because what produces the alert is the state of the task, not the data. In the last two rows the retry succeeds, so the **task alert never goes out.** In the third row one sign is left: the row count check sees 2640 rows and fires. Duplication is noisy and repairable; the half-write is silent. **The most dangerous setup is not the one that raises an alert but the one that stops raising it.** #### Summary - The half-written partition makes serving wrong in **all five regions** (between −49.09 and −121.82), yet the raw rows stay at 2320: both the row count check and the period-aware completeness check are **green**. - In the atomic write the partition is never written, consumption falls to **1800** instead of 2040, but the deviation stays between **−3.25 and +12.80** and **both checks fire**. Atomicity does not make the answer correct; it makes the failure **visible and bounded**. - The reason the order reverses is that serving is a period average: the cell that is not written does not enter the average, the cell that is written halfway enters it with a low number. - The write with a three-try retry that appends leaves **440 rows** in the partition instead of 200, that is, it duplicates **240 rows**; in the keyed write the duplicated rows are **zero**. What makes the retry safe is not the try but **the write being idempotent**. - That the duplicated rows do not damage serving is the work of the deduplication stage: the same store without that stage carries the consumption to 2440 rows and the north to 1320.73. - When the retry succeeds, the **task alert goes silent**; only the data alerts remain, and they never see the half-write. The place where the alert goes silent is the least measured place of the pipeline. #### Next Step In three lessons the pipeline was run: its order was read from the graph, its trigger was chosen, the failed task was retried and what the failure does to serving was counted. The common gap of these three lessons is this — every measurement was made **afterwards**. The table that shows the half-write was built after the run was over and the result was read; the pipeline itself said nothing at that moment. What is watched while a pipeline runs, how many periods later an indicator shows a corruption and which corruption **no indicator** sees were never asked. The next lesson runs the freshness, volume and schema change indicators on the six corruption classes of the Introduction to Data Engineering course and counts the catch lag of each in periods. ### Pipeline Observability Source: https://academia.sh/en/courses/data-pipelines/pipeline-observability Counting what three indicators placed on top of a running pipeline see: the freshness, volume and schema indicators look at a total of 14 observation points and none of them reads the value of a cell, so of the six corruption classes only the unit change, which alters the value, stays invisible to all three. Volume shows three classes in the same period, schema two classes in the same period, freshness the field rename one period later; while an indicator fires for at most one period the corruption lasts two periods, and in two classes serving does not change at all although the indicator fires. As the band widens from 0.05 to 0.30, the number of classes volume sees falls from 3 to 0 and the largest loss the band hides rises from 10 rows to 60 rows; when the freshness tolerance is 0 the lag falls from 1 period to 0, but a false alarm arises on a healthy pipeline. The previous lesson dealt with the failing task: a retry duplicates rows in a pipeline that is not idempotent, a half-written partition makes serving wrong in every region, and an atomic write brought the deviation down to between −2.89 and +12.80 by not writing the partition at all. In three lessons the pipeline was run and what happens when it fails was written down. A failing task announces itself. The subject of this lesson is the run that **does not fail**: all tasks finish green, the graph completes and the pipeline carries wrong data. There the only evidence is the **indicator** placed on top of the pipeline. The lesson builds three of them — freshness, volume and schema change — and asks a single question: **how many periods later** do they show the six corruption classes of the Introduction to Data Engineering course, and which one do they never show. - **OO41.** The source, the five-stage pipeline, the six corruption classes and the three indicators are **fictional** and are generated inside the lesson. No orchestrator, observability product or data store is called. The seed is 20260218. - **OO42.** The **observability** procedure — metric collection, tracing, logging, alert design — was measured in the Observability and Reliability course of the Backend Development curriculum, the Observability and Operations course of the DevOps and Platform Engineering curriculum and the AI Agents course of the AI Engineering curriculum; it is **not repeated**. The only measure here is the lag of three indicators specific to the data pipeline. - **OO43.** The six corruption classes are taken as **input** from the Introduction to Data Engineering course. Each of them corrupts only the `2026-07` period; the remaining eleven periods are healthy. - **OO44.** **Freshness** means that the newest period in serving is at most one period behind the period being worked on; **volume** means that the row count of the period does not deviate from the median of the previous three periods by more than the band; **schema evolution** means that the field name and type pairs stay the same as the registered signature. The tolerance, the band and the signature are each a **choice**; the last section counts them. - **OO45.** **Drift** is the change of the distribution of the input over time and was measured in the Model Evaluation, Interpretability and Ethics course of the Machine Learning and Data Science curriculum. The three indicators do not see drift; the first section counts the reason. - **OO46.** The baseline is the run without orchestration: no indicator, classes seen **0**, lag **undefined**. The lag is written in **periods**. #### The Surface the Indicators Look At The block below sets up the shared definition and prints a single thing: the **run-independent** measure of the three indicators, that is, the number of attributes each looks at. This number is known without running the pipeline at all. ```python # FICTIONAL. The meter readings of the measurement network, the five-stage pipeline, the six corruption # classes and the three indicators are generated inside the lesson; no orchestrator, observability # product or data store is called. Seed 20260218. 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 CORRUPTIONS = ["late_arrival", "duplicated", "schema_changed", "unit_changed", "missing_partition", "type_drift"] def corrupt(records, cls, period="2026-07"): """Corrupts a single period , the remaining periods stay healthy.""" new, count = [], 0 for k in records: if k["period"] != period: new.append(k) continue if cls == "late_arrival": if k["subscriber"] <= "A040": count += 1 continue new.append(k) elif cls == "duplicated": new.append(k) if k["subscriber"] <= "A060": new.append(dict(k)) count += 1 elif cls == "schema_changed": y = dict(k) y["index_value"] = y.pop("index") count += 1 new.append(y) elif cls == "unit_changed": y = dict(k) y["index"] = k["index"] * 1000 y["unit"] = "m3" # LABEL UNCHANGED count += 1 new.append(y) elif cls == "missing_partition": if k["region"] == "north": count += 1 continue new.append(k) elif cls == "type_drift": y = dict(k) y["index"] = str(k["index"]) count += 1 new.append(y) return new, count 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 not in seen: 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 # ---- THREE INDICATORS. Each looks at one projection of the record. REGISTERED = {("subscriber", "str"), ("period", "str"), ("region", "str"), ("type", "str"), ("index", "int"), ("unit", "str")} ATTRIBUTES = {"freshness": 1, "volume": 1, "schema": 2 * len(REGISTERED)} print(f"source {200 * len(PERIODS)} rows x {len(REGISTERED)} fields =" f" {200 * len(PERIODS) * len(REGISTERED)} cells") print("indicator what it looks at attributes") print(f"freshness newest period in serving {ATTRIBUTES['freshness']:>10}") print(f"volume rows per period {ATTRIBUTES['volume']:>10}") print(f"schema field name and type pairs {ATTRIBUTES['schema']:>10}") print(f"total observation points {sum(ATTRIBUTES.values())} ," f" cells whose value is read 0") ``` ``` source 2400 rows x 6 fields = 14400 cells indicator what it looks at attributes freshness newest period in serving 1 volume rows per period 1 schema field name and type pairs 12 total observation points 14 , cells whose value is read 0 ``` The source carries 14400 cells and the three indicators look at 14 points. What matters is the last line: **cells whose value is read: zero**. Freshness asks whether a period exists, volume asks how many rows there are, schema asks the name and type of the fields. None of them looks at **how many cubic meters** a meter wrote; this is also the reason drift is not seen. This line gives a prediction before the run: if among the six classes there is one that changes only the **value**, that class will be invisible to all three indicators. There is one class in the list that fits the definition. `unit_changed` multiplies the index by a thousand, leaves the unit label as `m3`, keeps the field name and type, and does not change the row count. That is the prediction; the run tests it. #### How Many Periods Later The block below runs the pipeline period by period over twelve periods and evaluates the three indicators after each period. For each class it prints the three lags, the first indicator that shows, **how many periods the corruption lasts** and whether serving changed. ```python # This block continues the generate_source , corrupt , s1..s5 and REGISTERED definitions of the first block. SOURCE = generate_source() INDICATORS = ("freshness", "volume", "schema") CORRUPT_PERIOD = "2026-07" def signature(records): return set((a, type(v).__name__) for k in records for a, v in k.items()) def median(x): y = sorted(x) return y[len(y) // 2] def summarize(records): v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) summary, _ = s4_summary(v) return summary def watch(records, band=0.05, tolerance=1): """The pipeline is run each period and three indicators are evaluated. Freshness carries a tolerance of tolerance periods; volume looks at the deviation from the median of the last three periods.""" fired, history = {g: [] for g in INDICATORS}, [] for i, d in enumerate(PERIODS): current = [k for k in records if k.get("period") == d] summary = summarize([k for k in records if PERIODS.index(k["period"]) <= i]) if i - max([PERIODS.index(x) for _, x in summary], default=-1) > tolerance: fired["freshness"].append(d) if len(history) >= 3: t = median(history[-3:]) if abs(len(current) - t) / t > band: fired["volume"].append(d) history.append(len(current)) if signature(current) != REGISTERED: fired["schema"].append(d) return fired def lag(periods): """How many periods later it showed. A dash if it never showed.""" after = [x for x in periods if x >= CORRUPT_PERIOD] return PERIODS.index(after[0]) - PERIODS.index(CORRUPT_PERIOD) if after else "-" BASELINE = summarize(SOURCE) BASELINE_SERVING, _ = s5_serve(BASELINE) print(f"healthy pipeline: no indicator fires , serving {BASELINE_SERVING}") print("\ncorruption freshness volume schema first indicator affected periods serving") for c in CORRUPTIONS: v, _ = corrupt(SOURCE, c) a, summary = watch(v), summarize(v) serving, _ = s5_serve(summary) lags = {x: lag(a[x]) for x in INDICATORS} seen = [(v2, x) for x, v2 in lags.items() if v2 != "-"] impact = [d for (b, d) in set(BASELINE) | set(summary) if BASELINE.get((b, d)) != summary.get((b, d))] print(f"{c:<18}{lags['freshness']:>10}{lags['volume']:>8}{lags['schema']:>8}" f" {(min(seen)[1] if seen else 'none'):<17}{len(set(impact)):>17}" f" {'changed' if serving != BASELINE_SERVING else 'same'}") print("\nband largest hidden loss classes seen by volume") for band in (0.05, 0.10, 0.20, 0.25, 0.30): seen = [c for c in CORRUPTIONS if watch(corrupt(SOURCE, c)[0], band=band)["volume"]] print(f"{band:<5} {int(band * 200):<21}{len(seen)} {seen}") print("\ntolerance false alarms on healthy pipeline lag of schema_changed") for tolerance in (0, 1, 2): false_alarms = watch(SOURCE, tolerance=tolerance)["freshness"] corrupted = watch(corrupt(SOURCE, "schema_changed")[0], tolerance=tolerance)["freshness"] print(f"{tolerance:<11}{len(false_alarms):<34}{lag(corrupted)}") ``` ``` healthy pipeline: no indicator fires , serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} corruption freshness volume schema first indicator affected periods serving late_arrival - 0 - volume 2 changed duplicated - 0 - volume 0 same schema_changed 1 - 0 schema 2 changed unit_changed - - - none 2 same missing_partition - 0 - volume 2 changed type_drift - - 0 schema 0 same band largest hidden loss classes seen by volume 0.05 10 3 ['late_arrival', 'duplicated', 'missing_partition'] 0.1 20 3 ['late_arrival', 'duplicated', 'missing_partition'] 0.2 40 2 ['duplicated', 'missing_partition'] 0.25 50 1 ['duplicated'] 0.3 60 0 [] tolerance false alarms on healthy pipeline lag of schema_changed 0 1 0 1 0 1 2 0 - ``` On the healthy pipeline no indicator fires; false alarms are zero and serving is the serving of the shared definition. The volume column shows three classes **in the same period**: `late_arrival` removes 40 rows, `missing_partition` 46, `duplicated` adds 60, and all three exceed the band of 0.05. Volume does not say what happened, it only says that the number has shifted; it does not show from which region the 46 rows dropped, because it carries no region breakdown. The period-aware completeness check of the Introduction to Data Engineering course says that. `schema_changed` is shown by two indicators **in two separate periods**. Schema fires in the same period: the signature has `index_value` instead of `index`. Freshness fires one period later: the 200 rows whose field name changed drop at screening, the serving cell of that period is never formed, and because freshness carries a tolerance of one period it stays silent at `2026-07` and speaks at `2026-08`. A single event, two indicators, a difference of one period. `type_drift` is visible only to schema, and serving **does not change**: the index turned into text can be converted back to an integer in the consumption computation. The indicator fires, there is no damage. `duplicated` is the same: volume fires, the deduplication stage swallows the 60 copies, and the affected periods are **0**. The `unit_changed` row has three dashes and **none**. The prediction held. Serving does not change either; the Introduction to Data Engineering course measured the mechanism — in the intermediate layer two periods move by about 1.8 billion and cancel each other, and the average of the twelve periods stays the same. The affected periods column still writes **2**: two summary cells are garbage. The only thing that sees this is the distribution check of the Introduction to Data Engineering course, and that is a data quality check, not a pipeline indicator. The difference is said in a single word: **it looks at the value**. The hardest reading is across the columns. The indicator fires for at most **one period**, the corruption lasts **two periods**. The reason is the edge of the graph between periods: because consumption depends on the reading of the previous period, the `2026-08` rows of the 40 subscribers lost at `2026-07` cannot be computed either. One period after the alert goes silent, the pipeline is still corrupted. The opposite direction is even worse: in `duplicated` and `type_drift` the indicator fires and the damage is zero; in `unit_changed` the indicator is silent and two periods are garbage. **There is no link between firing and damage.** #### Band and Tolerance Are Choices The last two tables count the cost of the indicator itself. The number of classes volume sees is three at the bands 0.05 and 0.10, two at 0.20, one at 0.25 and zero at 0.30. The second column is the real meaning of the band: because a period carries 200 rows, the band is **the largest loss it hides** — 0.05 stays silent up to ten rows, 0.30 up to sixty rows. The resolution of the set is one row, that is 0.005; everything under the band is invisible, and this is not a flaw but the definition itself. In this fictional source a narrow band is free, because every period carries exactly 200 rows and false alarms on the healthy pipeline are zero. In a real source the row count of a period varies, and the band is widened precisely to reduce the false alarm (false positive). As it widens, the three classes disappear in order of size: first the 40-row `late_arrival`, then the 46-row `missing_partition`, last the 60-row `duplicated`. The band is where false alarms and blindness are traded off, and the currency is the row. The tolerance table shows the same tradeoff in freshness, in a barer form. At tolerance 0 the lag of `schema_changed` falls from 1 period to **0**; the cost is **one false alarm** on the healthy pipeline, because the first period has no period before it and produces no consumption, and freshness without tolerance counts this as an empty serving. At tolerance 2 the false alarm stays zero but freshness **sees nothing**, the column is a dash. Three settings, three separate indicators. The schema indicator has no such row. It cannot be tuned because it has no threshold: the signature either holds or does not. In return it sees only what is written in the signature, and it is blind to `unit_changed` because that does not break the signature. **A tunable indicator chooses what it will miss; an untunable indicator fixes in advance, by its definition, what it will see.** #### Summary - The three indicators look at a total of **14 attributes** and none of them reads the **value** of a cell; this run-independent measure says in advance that the `unit_changed` class, which changes only the value, will stay invisible, and the run confirms it. - Volume shows `late_arrival`, `missing_partition` and `duplicated` **in the same period**; schema shows `schema_changed` and `type_drift` in the same period; freshness shows `schema_changed` **one period later**, because it carries a tolerance of one period. - The indicator fires for at most one period, the corruption lasts two periods: consumption depends on the reading of the previous period. In `duplicated` and `type_drift` the indicator fires while the damage is **0** periods, in `unit_changed` the indicator is silent while the damage is **2** periods. - As the band widens from 0.05 to 0.30, the number of classes volume sees falls from 3 to 0 and the largest hidden loss rises from 10 rows to 60 rows; the band is the tradeoff between false alarms and blindness, in rows. - When the freshness tolerance is lowered to 0, the lag falls from 1 period to 0 but **one false alarm** arises on the healthy pipeline; at tolerance 2 there is no false alarm and the indicator sees nothing. #### Next Step Every number in this lesson leaned on a single silent assumption: the code that runs the pipeline is always the same code, runs with the same setting and over the same dataset. The indicator band was 0.05, the tolerance was 1, the number of loaded periods was twelve — none of them was written in the code of the pipeline, they stood **around** the run. The next lesson runs the same pipeline code in two environments and measures how far serving deviates when the environment setting changes; then, when no version stamp is written on top of the run, it counts why it cannot be known with which code the report in your hands was produced. ### Environment and Version Management Source: https://academia.sh/en/courses/data-pipelines/environment-and-version-management Counting the different result the same pipeline code produces in three environments and the identity that is lost when no version stamp is written: the development, staging and production environments produce 400, 1000 and 2200 consumption rows respectively from the same 2400 rows and their reports differ from one another; the number of loaded partitions shifts serving between 20.41 and 62.55, the rule version between 20.77 and 35.87, and because the two effects are of the same order of magnitude they cannot be told apart by looking at serving, and in one region they cancel each other and bring the difference down to −5.72. In an archive of eleven runs, when the rule version changes after the seventh period, the share of the code in the observed jump is between 20.40 and 34.71 and stays within the range of 0.42 to 38.30 that the usual period movement would show if the code never changed; in four regions the data goes down while the report goes up. In the unstamped archive none of the 11 reports carries an identity and identification requires 22 runs. The lesson is the close of the course. The previous lesson placed three indicators on top of the pipeline and counted what each of them sees. Every number leaned on a single silent assumption: the code was always the same and always ran with the same setting. Yet the band was 0.05, the tolerance was 1, the number of loaded periods was twelve — all three stood **around** the run, not in the code of the pipeline. This lesson turns that surroundings into an object of measurement. A single pipeline code is deployed to three environments and two questions are asked: how different a result the same code gives in them, and whether a report in your hands shows **in which environment, with which code** it was produced. - **OO47.** The source, the five-stage pipeline and the three environment settings are **fictional** and are generated inside the lesson. No orchestrator, deployment tool or version management product is called. The seed is 20260218. - **OO48.** The procedure of continuous integration, delivery and environment promotion was measured in the Continuous Integration and Delivery and Infrastructure as Code courses of the DevOps and Platform Engineering curriculum; it is **not repeated**. The only measure here is the **different result** the same code produces across environments. - **OO49.** An environment carries three settings and all three are outside the code: the number of loaded **partitions**, the indicator **band** and the **version** of the reporting rule. `s1` reports commercial consumption with a coefficient of 1.00, `s2` with 1.05; the backfill procedure was measured in the last lesson of the Transformation topic. - **OO50.** A **version stamp** is the triple of period, code version and setting written to the output of a run. The baseline is the run without orchestration: a single environment is assumed, no stamp is written, and the number of reports that carry an identity is **0**. #### The Same Code, Three Environments The block below defines a single `report` function and runs it with three settings. From the environment the function takes only the set of periods to load and the rule coefficient; because the band tunes the indicator and not the pipeline, it enters the table with the measure of the previous lesson, the largest loss it hides. ```python # FICTIONAL. The meter readings of the measurement network, the five-stage pipeline and three # environment settings are generated inside the lesson; no orchestrator, deployment tool or # version management product is called. Seed 20260218. 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 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 out.append(dict(k, m3=int(k["index"]) - int(prev["index"]))) 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 # ---- A SINGLE pipeline code. From the environment: the loaded periods and the rule coefficient. # In a healthy source the screen and dedupe stages drop no rows, so consumption is # computed directly from the loaded periods. def report(records, periods, rule): v, _ = s3_consumption([k for k in records if k["period"] in periods]) v = [dict(k, m3=k["m3"] * (rule if k["type"] == "commercial" else 1.0)) for k in v] summary, _ = s4_summary(v) serving, _ = s5_serve(summary) return serving, len(v) def diff(a, b): return {x: round(a[x] - b[x], 2) for x in sorted(a)} SOURCE = generate_source() SETTINGS = {"development": (3, 0.30, 1.00), "staging": (6, 0.10, 1.00), "production": (12, 0.05, 1.05)} LAST = {name: set(PERIODS[len(PERIODS) - n:]) for name, (n, _, _) in SETTINGS.items()} print("environment partitions band rule consumption rows hidden by the band 2026-07 loaded") for name, (n, band, rule) in SETTINGS.items(): s, rows = report(SOURCE, LAST[name], rule) print(f"{name:<12} {n:>10} {band:<4} {rule:<5} {rows:>16} {int(band * 200):>18}" f" {'yes' if '2026-07' in LAST[name] else 'no':>14}") print() for name in SETTINGS: print(f"{name:<12} {report(SOURCE, LAST[name], SETTINGS[name][2])[0]}") ``` ``` environment partitions band rule consumption rows hidden by the band 2026-07 loaded development 3 0.3 1.0 400 60 no staging 6 0.1 1.0 1000 20 yes production 12 0.05 1.05 2200 10 yes development {'central': 1384.0, 'north': 1265.0, 'east': 928.5, 'west': 853.5, 'south': 1102.5} staging {'central': 1387.6, 'north': 1229.0, 'east': 897.8, 'west': 846.2, 'south': 1077.8} production {'central': 1357.32, 'north': 1235.57, 'east': 902.13, 'west': 859.22, 'south': 1095.81} ``` Three rows are three outputs of the same code and none is wrong; each is right for its own setting. `central` is reported as 1384.00 in development, 1387.60 in staging and 1357.32 in production, yet the source is the same 2400 rows in all three runs. The last column ties to the previous lesson. The development environment loads only the last three periods and `2026-07` falls outside that window, yet all six corruption classes of the Introduction to Data Engineering course corrupt exactly that period. The run that validates the pipeline **never sees** the period of the corruption; moreover, its band is 0.30 and already hides a loss of up to sixty rows. Validation ends green, production does not. #### Where the Deviation Came From The second block splits the deviation into two sources, then looks at the archive of eleven runs in which the rule version was deployed at the `2026-08` run. ```python # This block continues the report , diff , SOURCE , PERIODS , SETTINGS and LAST definitions of the first block. ALL = set(PERIODS) BASE, DEV, RULE = (report(SOURCE, ALL, 1.00)[0], report(SOURCE, LAST["development"], 1.00)[0], report(SOURCE, ALL, 1.05)[0]) print("region partition effect rule effect development minus production") for b in sorted(BASE): partition_effect, rule_effect = round(DEV[b] - BASE[b], 2), round(RULE[b] - BASE[b], 2) print(f"{b:<8} {partition_effect:>16} {rule_effect:>12} {round(partition_effect - rule_effect, 2):>28}") # ---- ARCHIVE: one run per period , all periods up to that period are loaded. # The rule version was s1 (1.00) ; s2 (1.05) was deployed in the 2026-08 run. CODE, DEPLOY = {"s1": 1.00, "s2": 1.05}, "2026-08" archive = {d: ("s2" if d >= DEPLOY else "s1", report(SOURCE, set(PERIODS[:i + 1]), CODE["s2" if d >= DEPLOY else "s1"])[0]) for i, d in enumerate(PERIODS) if i > 0} # Usual period movement: how much the report would move if the code never changed. # The first two runs are skipped , the average there carries fewer than two periods. usual = [abs(v) for i in range(3, len(PERIODS)) for v in diff(report(SOURCE, set(PERIODS[:i + 1]), 1.00)[0], report(SOURCE, set(PERIODS[:i]), 1.00)[0]).values()] y07, y08 = set(PERIODS[:7]), set(PERIODS[:8]) observed = diff(archive["2026-08"][1], archive["2026-07"][1]) from_data = diff(report(SOURCE, y08, 1.00)[0], report(SOURCE, y07, 1.00)[0]) from_code = diff(report(SOURCE, y08, 1.05)[0], report(SOURCE, y08, 1.00)[0]) print("\nregion observed jump from data from code") for b in sorted(observed): print(f"{b:<8} {observed[b]:>16} {from_data[b]:>9} {from_code[b]:>9}") print(f"usual period movement (code fixed) {round(min(usual), 2)} to {round(max(usual), 2)} ;" f" trace left by the code {min(from_code.values())} to {max(from_code.values())}") unstamped = {d: {"period": d, "serving": s} for d, (_, s) in archive.items()} stamped = {d: dict(x, code=archive[d][0], setting=f"production/{len(PERIODS)}/{CODE[archive[d][0]]:.2f}") for d, x in unstamped.items()} identity = [sum("code" in x and "setting" in x for x in a.values()) for a in (unstamped, stamped)] print(f"\narchive {len(archive)} reports ; carrying identity: unstamped {identity[0]} ," f" stamped {identity[1]} ; runs for identification: unstamped" f" {len(CODE) * len(archive)} , stamped 0") print(f"2026-08 stamp: {stamped['2026-08']['code']} / {stamped['2026-08']['setting']}") ``` ``` region partition effect rule effect development minus production central 62.55 35.87 26.68 east 47.14 20.77 26.37 north 52.45 23.02 29.43 south 36.95 30.26 6.69 west 20.41 26.13 -5.72 region observed jump from data from code central 52.24 17.53 34.71 east 15.59 -4.81 20.4 north 9.88 -11.4 21.28 south 25.1 -3.76 28.86 west 13.74 -11.88 25.62 usual period movement (code fixed) 0.42 to 38.3 ; trace left by the code 20.4 to 34.71 archive 11 reports ; carrying identity: unstamped 0 , stamped 11 ; runs for identification: unstamped 22 , stamped 0 2026-08 stamp: s2 / production/12/1.05 ``` The two effects in the first table are of the same order of magnitude: the number of partitions shifts serving between 20.41 and 62.55, the rule version between 20.77 and 35.87, both upward. Someone who sees the report rise by thirty units cannot tell **by looking at serving** whether it came from the rule version or from the number of loaded periods. The `west` row says something worse: the two effects are +20.41 and +26.13, and the total difference is **−5.72**. A number that says almost nothing has changed is two large effects cancelling each other. **A small difference is not proof of a small change.** In the second table the reader sees only the observed jump, between 9.88 and 52.24. The decomposition says what it is: in four of the five regions **the data went down and the report went up**; in `west`, consumption fell by 11.88 while the report rose by 13.74. The last line closes the door: the trace left by the code is 20.40 to 34.71, the usual period movement 0.42 to 38.30. The trace of the code is **inside the usual**; there is no way to tell from the jump that a deployment took place. What remains is the stamp. In the unstamped archive **none** of the eleven reports carries an identity; in the stamped one all eleven do. Finding out with which code a report in the unstamped archive was produced requires re-running two versions across eleven periods, that is **22 runs** — if every old version can still be installed. The scope is measured too: the rule version was in the environment setting, not in the code, so a stamp holding only the code fingerprint would give `s1` and `s2` **the same stamp**. **The boundary of the stamp is not the boundary of the code, it is the boundary of everything that determines the run.** #### Summary - The same pipeline code produces three separate reports in three environments and none of them is wrong: 400, 1000 and 2200 consumption rows come out of the same 2400 rows, and `central` is reported as 1384.00, 1387.60 and 1357.32. - The development environment never loads `2026-07` and its band hides a loss of up to sixty rows; the validating run ends green without seeing the period in which the corruption lies. - The effect of the environment setting (20.41 … 62.55) and the effect of the rule version (20.77 … 35.87) are of the same order of magnitude; in `west` the two cancel each other and bring the difference down to **−5.72**. - The trace left by the code (20.40 … 34.71) stays inside the usual period movement (0.42 … 38.30), and in four regions the report rises while the data goes down. - The number of reports that carry an identity is **0** in the unstamped archive and **11** in the stamped one; identification requires 22 runs, and the stamp covers not the code but the whole setting, including the rule version. #### Course Wrap-Up The rows share no common unit; **which measure** each row is read by is written next to the lesson name. The difference in parentheses is **the setup with orchestration minus the baseline without orchestration**; the sign is read from the measure name. | Lesson | Baseline without orchestration | Setup with orchestration | Cost | |---|---|---|---| | Ingestion Types — rows transported and changes missed | full load **15,560**, missed **0** | incremental load **2,360** (**−13,200**), missed **148** (**+148**) | only the distribution check reports the missed ones, at stage 4 | | Connectors and API Consumption — rows ingested, page 25 | without retry **2075** | with retry **2025** (**−50**) | 112 pages arrive late, `2026-12` is never ingested | | Messaging Systems — active consumers | single partition **1/4** | four partitions **4/4** (**+3**) | on the subscriber key the load skews between 708 and 492 | | Stream Processing — dropped rows and dirty cells | tolerance 0: **40** and **10/55** | tolerance 1: **0** and **0** (**−40**, **−10**) | the carried state grows from 400 to **600** entries | | Delivery Guarantees — consumption rows and deviation | at most once **2162**, −9.18 … −28.36 | at least once **2200** (**+38**), deviation **empty** | 19 duplicates arise, the loss is below both thresholds | | Reverse ETL — difference between runs | pipeline that does not write back **0.00** | reverse ETL **47.00 / 10.90 / 14.36** | **55** of 55 cells get corrupted, four checks are green | | ETL and ELT — rule that can be recomputed | ETL **0**, 55 rows to the target | ELT **3** (**+3**), 2400 rows to the target (**+2345**) | cells read in the target rise from 110 to **33110** | | Transformation Frameworks — violated dependency | manual order **1**, **7** models with missing inputs | resolver **0** and **0** (**−1**, **−7**) | the violating run raises no error, serving deviates by 17.46 … 28.37 | | Staged Layers — corruption class stopped | serving only **2/6**, 55 cells | three layers **6/6** (**+4**), 2400 + 2200 + 55 | **three** of the six caught do not change the served value | | Testable Transformations — corruption class caught | without expectations **0/6** | six expectations **4/6** (**+4**) | `duplicated` and `type_drift` are swallowed in the transformation | | Backfill — recomputed period | full apply **12** periods, 2200 rows | partial apply **2** periods, 400 rows (**−10**) | the report deviates by **−16.98 … −29.57**, six expectations are green | | Workflow Schedulers — tasks re-run per failed task | by hand **48** | with graph **5** (**−43**) | forty-seven edges are written by hand | | Dependency and Triggering — missing periods and consumption rows | fixed schedule **6** missing, **900** | data readiness trigger **0** and **2200** (**+1300**) | seven periods of total delay | | Retry and Alerting — deviation and the check that fires | half write **−49.09 … −121.82**, two checks green | atomic write **−3.25 … +12.80**, both fire | consumption falls to **1800** instead of 2040 | | Pipeline Observability — corruption class seen | no indicator **0/6**, lag undefined | three indicators **5/6** (**+5**), lag 0–1 periods | `unit_changed` stays invisible and leaves two periods of garbage | | Environment and Version Management — report carrying identity | unstamped archive **0/11** | stamped archive **11/11** (**+11**) | identification requires **22 runs** when unstamped | The table is a single rule written sixteen times. **A pipeline's number is not that it ran correctly once, but that it gives the same result when re-run; a pipeline that cannot be re-run counts as unmeasured.** The measure names are the rule itself: in Workflow Schedulers the re-run task falls from 48 to 5, in ETL and ELT the recomputable rule rises from 0 to 3, in Reverse ETL, as the pipeline reads its own output, the difference between runs leaves 0.00 and reaches 47.00. **The second claim is read from the same table: orchestration also corrupts.** In Connectors and API Consumption the retry eats the quota, drops the rows ingested from 2075 to 2025 and wipes out one period. In Reverse ETL, writing back raises the difference between runs to 47.00 and corrupts 55 of 55 cells, in Retry and Alerting a three-try retry duplicates 240 rows, in Backfill a partial backfill skews the report between −16.98 and −29.57, in Dependency and Triggering the safety window raises the runs to 24. In Pipeline Observability, as the band widens, the number of classes volume sees falls to zero; in this lesson the environment setting skews the same code by up to 62.55. **The third claim is quieter: the served number does not change or looks plausible.** In Stream Processing, in none of nine setups do the five served values change, and ten cells are dirty. In Delivery Guarantees nineteen duplicates leave the serving difference empty, and the loss is caught by no check. In Retry and Alerting the raw row count stays at 2320 in the half-written partition and two checks are green, in Testable Transformations the rule that shifts the period label leaves 2200 rows and five values the same, in Staged Layers three of the six stopped classes do not change the served value, in Pipeline Observability `unit_changed` is invisible to three indicators at once and in this lesson the trace left by the code falls inside the usual movement. In nine lessons the pipeline serves a wrong number and nothing says so. In sixteen lessons the pipeline was handled as if it ran on **a single machine**: tasks lined up, failed, were retried and forty-eight nodes were laid out in a single list. When the data does not fit on a single machine, **how the work will be split**, which partition will delay the finish when the partitions are unbalanced and the cost of **moving data over the network** were never asked. The next course measures not the order of a task but **into how many pieces a task is split** and how much data those pieces send to one another.