--- title: 'Introduction to Data Engineering' source: 'https://academia.sh/en/courses/introduction-to-data-engineering' language: en updated: '2026-10-21T05:03:00+00:00' license: 'CC BY-SA 4.0' lessons: 10 outcomes: - 'Builds the stages of the data engineering lifecycle on a fictional pipeline and counts the rows each stage drops' - 'Justifies the choice between batch and stream processing as a trade-off between freshness and detection lag' - "Turns data quality dimensions into working checks and measures the check's own defect" --- # Introduction to Data Engineering Building a fictional five-stage pipeline and counting how many stages later six corruption classes are caught: the pipeline reducing 2400 rows stage by stage to 5 served values and how many rows stand behind each served value, each stage's own corruption exposure, the same corruption reflecting differently on the data engineer's measure and the model builder's, and a unit change being visible in neither, the corruption profile and detection lag per source class, a corruption showing up one period later in batch processing and in the same period in streaming, along with the number of stages streaming enlarges, turning quality dimensions into checks and a partition check staying green because it does not see the period, the rows a schema change drops and the stage where a contract catches it, reprocessing duplicating rows in a pipeline that is not idempotent while the served number stays the same, lineage shortening the stages looked at to get from a corrupted value to its source, how many of the five stages carry an owner, and the number of periods a retention policy makes impossible to check retroactively. ## Domain and Lifecycle Building a fictional data pipeline with five stages and counting how much each stage carries: 2400 source rows passing through screening, deduplication, consumption, summary and serving stages down to 5 served values and how many rows stand behind each served value, the stage at which each of six corruption classes shows its effect and which of them never change the served number, the same corruption reflecting differently on the data engineer's measure and the model builder's, and a unit change being visible in neither, the corruption profile carried by the application database, log and external service source classes and the detection lag of each, a corruption showing up one period later in batch processing and in the same period in stream processing, along with the number of stages streaming enlarges and the state it carries. ### What Is Data Engineering Source: https://academia.sh/en/courses/introduction-to-data-engineering/what-is-data-engineering Passing 2400 meter readings of a fictional measurement network through a five-stage pipeline and counting the rows each stage carries: screening 2400, deduplication 2400, consumption 2200, summary 55 and serving 5. The consumption stage drops zero rows, yet its output is two hundred rows smaller, because the difference for the first period does not exist by definition. Behind the five served values stand 2400 source rows, an average of 480 per value; the largest value rests on 588 source rows, the smallest on 372. Because the pipeline is deterministic, all of these numbers can be derived without a run. In the uninstrumented pipeline the corruption caught is zero and the detection lag is undefined, because there is nothing that measures. The AI Engineering curriculum, opened by the Introduction to AI Engineering course, built the machinery around the model in one hundred thirty-two lessons and wrote down the cost of every part. Throughout those lessons the data entering the model was assumed to be **ready**: there was a table, it was read; there was a feature, it was computed. How that table got filled, what it takes to keep it running in production, and **who will know when it breaks** were never asked. The subject of this curriculum is that pipeline, and one peculiar flaw of the pipeline defines the whole measure: **a pipeline gives notice when it crashes, and says nothing when it carries wrong data.** A crashing pipeline leaves an error record, triggers an alert, piles up a queue; someone looks. A pipeline that carries wrong rows runs, finishes, writes the table, and the dashboard stays green. The job of this lesson is to build that pipeline and to **count**: how many rows go in, how many each stage carries, how many reach the served number. - **DL1.** The code is a **fiction**. No real data store, file format, orchestrator or runtime is called; the pipeline is written with the plain standard library. The source is the **fictional** meter readings of a measurement network, and the seed is **20260218**. - **DL2.** The source is **200 subscribers x 12 periods = 2400 readings**. Each reading carries an index value, the index only increases, and the unit is `m3`. Consumption is computed from the index difference. - **DL3.** The pipeline has **five stages**: `screen`, `dedupe`, `consumption`, `summary`, `serve`. Each stage **counts** the rows it drops; there is no uncounted drop. - **DL4. The procedure for collection, cleaning, missing values and outliers is not repeated.** It was measured in the Data Collection and Preparation course. The only thing added here is the **stage counter**: at which step and why a row dropped is written down. - **DL5. The procedure for schema, constraints and normalization is not repeated.** It was measured in the Data Modeling and Relational Theory course. In this lesson the schema is reduced to the single question "is the required field there". - **DL6.** The pipeline is **deterministic**. `random` is not used, the generator is written inside the lesson, and the same seed gives the same 2400 rows. For this reason the measures are independent of the run, and all of the numbers below can be derived without a run. - **DL7.** No product, vendor or format name is written; stores and processors are named by their **kind**. Time is not written as raw duration; lag is written as **stages** or **periods**. - **DL8.** The resolution of the set is the lower bound of the measurement band: over 2400 rows the smallest step of a row ratio is **0.0004**, over 55 summary cells **0.0182**, over 5 served values **0.2000**. **No ratio is printed on the serving layer**; the value itself is written. #### The Five Stages of the Pipeline The responsibility of data engineering fits in one sentence: to make data produced in one place usable in another. The word "usable" in that sentence hides a series of steps, and unless the steps are counted the responsibility cannot be measured. The pipeline below splits those steps into five parts: a row without a required field is **screened out**, the same subscriber-period pair is **deduplicated**, **consumption** is computed from the index difference, a **summary** is produced per region and period, and the period mean per region is **served**. The order of these five is not arbitrary, and the code itself enforces two of the ties. The `dedupe` stage reads the `subscriber` and `period` fields of the row directly; if one of these fields is missing, the stage stops with an error. So **screening has to come before deduplication**, or the pipeline crashes. The second tie is quieter: the `consumption` stage produces one output row for every input row, and the `summary` stage adds these rows up. If a double record entered consumption without being deduplicated, the same subscriber-period difference would produce two output rows and would be added to the summary cell **twice**. When the first tie is violated, the pipeline stops and gives notice; when the second tie is violated, the pipeline runs, finishes and **writes a wrong number**. The stage order of a pipeline is the first decision that determines which errors will be loud and which will be silent. ```python # FICTIONAL pipeline -- no real store, format or orchestrator is called. # The source is FICTIONAL meter readings of a measurement network, 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)} # One reading per subscriber per period. Unit m3 , the index only increases. 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 lacks a required field 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. 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 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: 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 STAGES = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption), ("summary", s4_summary), ("serve", s5_serve)] def pipeline(records): trace, data = [], records for name, f in STAGES: data, dropped = f(data) n = len(data) if hasattr(data, "__len__") else 0 trace.append({"stage": name, "remaining": n, "dropped": dropped}) return data, trace source = generate_source() served, trace = pipeline(source) print("source rows", len(source), " subscribers", len(SUBSCRIBERS), " periods", len(PERIODS)) print("stage remaining dropped") for a in trace: print(f"{a['stage']:<13}{a['remaining']:>10}{a['dropped']:>9}") print("served") for b, v in sorted(served.items(), key=lambda x: -x[1]): print(f" {b:<8}{v:>9.2f}") ``` ``` source rows 2400 subscribers 200 periods 12 stage remaining dropped screen 2400 0 dedupe 2400 0 consumption 2200 0 summary 55 0 serve 5 0 served central 1321.45 north 1212.55 south 1065.55 east 881.36 west 833.09 ``` The first two rows of the trace table pass with zero drops: because the source is intact, there is no row with a missing field to screen out and no double record to deduplicate. The third row is the key point of the reading. The `consumption` stage **drops zero rows**, yet its output is **two hundred rows smaller**. This difference is not a loss; it is the definition itself: consumption is the index difference of two periods, and the **difference for the first period of each subscriber does not exist**. The first period of two hundred subscribers cannot be computed, but these are not faulty rows. That is why the "remaining" and "dropped" columns of a pipeline trace are kept separate: **a row that does not come out by definition and a row removed because of an error cannot be written in the same column.** None of these numbers depends on the run. Because the pipeline is deterministic, the table can be derived without a run: 200 subscribers x 12 periods = 2400; each subscriber loses one period, so 200 x 11 = 2200; 5 regions x 11 periods = 55 cells; 5 regions = 5 values. A second run with the same seed gives the same table, so every difference seen in the comparisons below comes from the corruption itself, not from variability. #### The Rows Behind a Served Value The most looked-at place of a pipeline is its last stage, because the report is read from there. The last stage of this pipeline produces **five numbers**. How many rows stand behind those five numbers says how much of the pipeline they hide. ```python # How many rows stand behind each served value v1, _ = s1_screen(source) v2, _ = s2_dedupe(v1) v3, _ = s3_consumption(v2) summary, _ = s4_summary(v3) subs, rows, cells = {}, {}, {} for a in SUBSCRIBERS.values(): subs[a["region"]] = subs.get(a["region"], 0) + 1 for k in v3: rows[k["region"]] = rows.get(k["region"], 0) + 1 for b, d in summary: cells[b] = cells.get(b, 0) + 1 print("region served summary consumption source subscribers") for b, v in sorted(served.items(), key=lambda x: -x[1]): print(f"{b:<9}{v:>9.2f}{cells[b]:>9}{rows[b]:>13}{subs[b] * 12:>8}{subs[b]:>13}") print(f"{'total':<9}{len(served):>9}{len(summary):>9}{len(v3):>13}{len(source):>8}{len(SUBSCRIBERS):>13}") ``` ``` region served summary consumption source subscribers central 1321.45 11 539 588 49 north 1212.55 11 506 552 46 south 1065.55 11 440 480 40 east 881.36 11 374 408 34 west 833.09 11 341 372 31 total 5 55 2200 2400 200 ``` The table plainly writes the pipeline's **compression ratio**. The only number visible in the report for `central`, **1321.45**, stands on the **588 source rows** of forty-nine subscribers over twelve periods, on the **539 consumption rows** computed from them, and on eleven **summary cells**. The smallest value, `west`, stands on **372 source rows**. Together the five values represent **2400 rows**, that is, an average of **480 source rows** per value. This compression has two faces. The useful face is clear: nobody can decide by looking at two thousand four hundred rows, and that is exactly the job of a summary. Its cost is the subject of the measure. When a source row is corrupted, that corruption enters the served number **with a weight of one in 480**; for `central`, one in 588. The ratio of the consumption layer to the source is **0.9167**, and since the smallest step of this ratio is 0.0004, it can be read to three digits; but writing a ratio on the serving layer is meaningless, because only five values exist there and the proportional weight of one value is 0.2000. What is to be written on the serving layer is **the value itself**. From here comes the first structural result of the pipeline: **the corruption of a single row dilutes so far that it is not visible in the served number, but the corruption of a hundred rows in the same direction does not dilute that far.** At which threshold which corruption becomes visible is the real quantity this pipeline is to measure, and this lesson does not measure it yet; it only lays the ground. Compression has one more side that does not show in the row count: the pipeline drops not only rows but also a **distinguishing dimension**. The output of the first three stages is keyed by the `subscriber` and `period` pair, and two hundred separate subscribers stand in the table. The output of the `summary` stage, however, is keyed by the `region` and `period` pair; the subscriber column does **not remain** there. From none of the fifty-five cells can it be read which subscribers were added up. The `serve` stage does the same thing to the period axis: from none of the five values can it be read how much each period added. This is not a loss but the definition of a summary, yet its consequence in production is heavy. When a served number looks suspicious, walking backward cannot be done by **reading** the intermediate outputs; the first three stages of the pipeline have to be **rerun**. Writing at which stage the pipeline leaves which dimension is the same thing as writing in advance the cost of investigating a corruption. #### The Count of the Uninstrumented Pipeline The pipeline above runs, finishes and writes five numbers. The only thing known about this pipeline is **how many rows it carries**. Nothing is known about whether the rows it carries are **correct**, because the pipeline has no step that asks this. From here on, in every lesson of the curriculum, **three numbers will stand side by side**: 1. **Uninstrumented pipeline.** No check is enabled. Corruption caught **0**, detection lag **undefined**. The pipeline in this lesson is exactly this. 2. **Instrumented pipeline.** The same workload, with checks added. The corruption caught and the **detection lag** are written as a stage number. 3. **Cost.** The detection lag, the good rows wrongly blocked, the rows reprocessed and the stages added. The items in the third point are the most often skipped side of this curriculum. Adding a check is not a free improvement: a check can also block a good row, can stop the pipeline needlessly, can duplicate rows during reprocessing and **can itself be flawed**. That is why, when the instrumented pipeline is written, not only the corruption caught but also the **cost of the check itself** goes into the same table. When the number of corruptions caught is written alone, the improvement is overstated. The word "undefined" in the first point should be read carefully. The detection lag of the uninstrumented pipeline is not **large**, it **does not exist**; since nothing measures, a quantity called lag does not arise. This is the difference between zero and undefined, and it is often confused in reporting: the sentence "no corruption was caught" is, when there is no check, not an announcement of success but an announcement **that no measurement was made**. From here comes the rule of the curriculum, and it holds through six courses: **the number of a data pipeline is not the rows it carries, but after how many steps a wrong row is caught. A pipeline whose check that catches the corruption is not written counts as unmeasured.** The trace of the pipeline above looks flawless, because the only thing the trace shows is how many rows the stages hand over to one another. About the content of the rows handed over, the trace is silent, and the silence is written in green. #### Summary - The responsibility of data engineering is to make data produced in one place usable in another; this responsibility becomes measurable only when the stages of the pipeline and the rows each stage carries are counted. - A pipeline gives notice when it crashes and says nothing when it carries wrong data; the measure of this curriculum is the second case. - The five-stage pipeline carries 2400 source rows as 2400, 2400, 2200, 55 and 5; the `consumption` stage drops zero rows, yet its output is two hundred rows smaller, because the difference for the first period does not exist by definition. - Behind the five served values stand 2400 source rows: an average of 480 per value, 588 for the largest, 372 for the smallest. The corruption of a row enters the served number diluted by this weight. - In the uninstrumented pipeline the corruption caught is 0 and the detection lag is **undefined**; the sentence "no corruption was caught" is an announcement that no measurement was made. - The rule of the curriculum: the number of a data pipeline is after how many steps a wrong row is caught; a pipeline whose check that catches the corruption is not written counts as unmeasured. #### Next Step The pipeline of this lesson ran with an intact source, and the trace table gave the expected number at every stage. The source being intact was an assumption; in production the source does not arrive intact. The records of a period arrive late, the same record is written twice, the name of a field changes, a reading is sent in another unit, the partition of a region is never written, a number field turns into text. None of these accidents is an attack, and all of them happen on the producer side. The next lesson defines these six classes by name and passes them through the same pipeline; what is to be measured is **at which stage** each corruption shows its effect. The most uncomfortable column of the table will be the last one: three of the six classes leave the five values the pipeline serves **exactly the same**. ### Lifecycle Source: https://academia.sh/en/courses/introduction-to-data-engineering/lifecycle Passing six corruption classes through the same five-stage pipeline and counting each stage's own corruption exposure: screening sees only the field name change, deduplication sees only the double record, the consumption stage raises the cost of a missing row to exactly double (40 rows to 80, 46 rows to 92, 200 rows to 400), the summary cell count falls from 55 to 45 and 53, and the serving stage produces nothing other than five values in any of the six classes. Three classes leave the served number exactly the same: duplicated is swallowed in deduplication, type_drift is silently repaired by the pipeline's own type conversion, and unit_changed, although it destroys ten cells of the summary, does not change the served value because two periods cancel each other. late_arrival and unit_changed corrupt exactly the same number of summary cells, yet one changes the serving and the other does not. The previous lesson ran the pipeline with an intact source, and the trace table gave the expected number at every stage: 2400, 2400, 2200, 55, 5. The source arriving intact was an assumption, and in production this assumption does not hold. The records of a period arrive late, the same record is written twice, the name of a field changes, a reading is sent in another unit, the partition of a region is never written, a number field turns into text. The question of this lesson is this: when these accidents pass through the same pipeline, **at which stage** do they leave a trace. Each of the five stages is structurally sensitive to a particular error and structurally blind to the rest, and this blindness comes from the design of the pipeline, not from a deficiency. What is to be measured is each stage's own **corruption exposure**. - **DL9.** The pipeline, the source and the seed are the same as in the previous lesson. The only thing that changes is **the source itself**: the corruption is applied not to the pipeline but before the data enters the pipeline. - **DL10. None of the six corruption classes is an attack.** All of them are everyday accidents on the producer side: a collection job finished late, a resend wrote twice, a field name was renamed, a converter in the field changed, the write of a partition was skipped, an export gave every field as text. - **DL11.** Each class corrupts **a single period** (`2026-07`); the remaining eleven periods stay intact. This is to isolate the effect of the corruption: every difference in the trace table is the difference of a single period. - **DL12.** The number of rows the classes corrupt is by definition: `late_arrival` brings no record of the first forty subscribers in this period (**40**), `duplicated` writes the record of the first sixty subscribers twice (**60** extra rows), `schema_changed` changes the name of the `index` field (**200**), `unit_changed` multiplies the reading by a thousand but **leaves the unit label as `m3`** (**200**), `missing_partition` never writes the partition of one region (**46**), `type_drift` turns the number field into text (**200**). - **DL13. There is no check in this lesson.** The baseline is the uninstrumented pipeline: the corruption caught is **0**, the detection lag is **undefined**. The only thing measured is the trace of the pipeline itself. - **DL14.** The unit of measure is the **row** and the **summary cell**. Time is not written. Because the pipeline is deterministic, a second run gives the same table; every difference below comes from the corruption, not from variability. - **DL15.** The resolution is as in the previous lesson: **0.0004** on the row ratio, **0.0182** on the summary cell, **0.2000** on the served value. **No ratio is printed on the serving layer.** #### The Trace of the Six Classes The block below builds the pipeline of the previous lesson as it is, adds the six corruption classes on top of it, and passes each class separately through the same pipeline. Two tables are printed: **how many rows came out** of the stages and **how many rows each stage dropped**. ```python # FICTIONAL pipeline and FICTIONAL corruption classes -- no real store 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 # None of them is an attack; all are producer-side accidents. 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 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") # 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" # 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"]) # number turned into text 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): """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 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)] def pipeline(records): trace, data = [], records for name, f in STAGES: data, dropped = f(data) n = len(data) if hasattr(data, "__len__") else 0 trace.append({"stage": name, "remaining": n, "dropped": dropped}) return data, trace source = generate_source() _, base_trace = pipeline(source) STAGE_NAMES = ("screen", "dedupe", "consumption", "summary", "serve") print(f"{'label':<19}{'corrupted':>10}" + "".join(f"{a:>12}" for a in STAGE_NAMES)) print(f"{'intact':<19}{0:>10}" + "".join(f"{a['remaining']:>12}" for a in base_trace)) for cls in CORRUPTIONS: bad, n = corrupt(source, cls) print(f"{cls:<19}{n:>10}" + "".join(f"{a['remaining']:>12}" for a in pipeline(bad)[1])) print() print(f"{'label':<19}" + "".join(f"{a:>12}" for a in STAGE_NAMES[:3]) + f"{'secondary':>11}") for cls in CORRUPTIONS: bad, n = corrupt(source, cls) trace = pipeline(bad)[1] print(f"{cls:<19}" + "".join(f"{a['dropped']:>12}" for a in trace[:3]) + f"{base_trace[2]['remaining'] - trace[2]['remaining']:>11}") ``` ``` label corrupted screen dedupe consumption summary serve intact 0 2400 2400 2200 55 5 late_arrival 40 2360 2360 2120 55 5 duplicated 60 2460 2400 2200 55 5 schema_changed 200 2200 2200 1800 45 5 unit_changed 200 2400 2400 2200 55 5 missing_partition 46 2354 2354 2108 53 5 type_drift 200 2400 2400 2200 55 5 label screen dedupe consumption secondary late_arrival 0 0 40 80 duplicated 0 60 0 0 schema_changed 200 0 200 400 unit_changed 0 0 0 0 missing_partition 0 0 46 92 type_drift 0 0 0 0 ``` The second table writes exactly **what each stage is exposed to**. The `screen` stage sees only a single class: `schema_changed`, because when the field name changes the row no longer meets the required field list. None of the remaining five classes drops a row at this stage. The `dedupe` stage also sees only one class: `duplicated`, sixty double records. The exposure of these two stages is **narrow and exact**; they do not miss what they see, and they are completely blind to what they do not. The `consumption` stage behaves differently. On its own it recognizes no corruption, but it **carries** the effect of three classes: 40 rows for `late_arrival`, 200 for `schema_changed` and 46 for `missing_partition` drop at this stage. These rows are not corrupted; they are **intact** rows whose difference cannot be computed because their previous periods are lost. The row a stage drops may not be that stage's own error. The last two columns are the result of the reading. At the `summary` stage the cell count falls in only two classes: `schema_changed` costs ten cells (45), `missing_partition` two cells (53). And the `serve` column is **5 in all six rows**. The output count of the serving stage depends on the number of regions, not on the state of the source; so the most looked-at layer of the pipeline can **under no condition** give notice of corruption by the row count. #### A Row Costing Two Rows The `secondary` column of the first table measures a separate quantity: the rows that **fall away from the consumption output** as a result of the corruption. `late_arrival` removes 40 rows from the source and **80** rows fall away from the consumption output. `missing_partition` removes 46, **92** at the output. `schema_changed` removes 200, **400** at the output. The factor is exactly **two** in all three. This factor is not a measurement result but a result of the definition, and it is derived without a run. Consumption is the index difference of two neighboring periods; when the `2026-07` record is lost, for that subscriber **both the July difference and the August difference** can no longer be computed, because the previous period of August is July. The loss of one row takes two output rows with it. The same reasoning also says **why the factor is not greater than two**: the previous period of September is August, and the August record is in place, so the chain closes there. This corrects a common misconception in data pipelines. When a source shortfall is reported, its effect is usually written with the number of source rows that fell away. If the chain of stages includes a transformation that builds a **link** between rows, the real effect is greater than this number, and the growth multiplier can be read from the definition of the pipeline. Here the multiplier is two; in a transformation joining three neighboring periods it would be three. #### The Summary Is Corrupted While the Serving Is Not The two tables so far looked only at the **row and cell counts**. It is still possible for the **value** to have changed while the count stays the same. The block below compares the summary cells and the served values with the intact pipeline. ```python def last_two(records): v1, _ = s1_screen(records) v2, _ = s2_dedupe(v1) v3, _ = s3_consumption(v2) summ, _ = s4_summary(v3) return summ, s5_serve(summ)[0] base_summary, base_serving = last_two(source) print(f"{'label':<19}{'cells':>7}{'changed':>9}{'missing':>9}{'serving same':>14}") for cls in CORRUPTIONS: summ, srv = last_two(corrupt(source, cls)[0]) changed = sum(1 for a in summ if a in base_summary and summ[a] != base_summary[a]) missing = sum(1 for a in base_summary if a not in summ) print(f"{cls:<19}{len(summ):>7}{changed:>9}{missing:>9}" + f"{('yes' if srv == base_serving else 'no'):>14}") b_summary, b_serving = last_two(corrupt(source, "unit_changed")[0]) print("\nunit_changed , north summary cells") for d in ("2026-06", "2026-07", "2026-08", "2026-09"): print(f" {d} baseline {base_summary[('north', d)]:>5} corrupted {b_summary[('north', d)]:>12}") print(" total of 11 cells baseline", sum(v for (b, d), v in base_summary.items() if b == "north"), " corrupted", sum(v for (b, d), v in b_summary.items() if b == "north")) print(" served north value baseline", base_serving["north"], " corrupted", b_serving["north"]) ``` ``` label cells changed missing serving same late_arrival 55 10 0 no duplicated 55 0 0 yes schema_changed 45 0 10 no unit_changed 55 10 0 yes missing_partition 53 0 2 no type_drift 55 0 0 yes unit_changed , north summary cells 2026-06 baseline 1041 corrupted 1041 2026-07 baseline 1251 corrupted 1812946491 2026-08 baseline 1119 corrupted -1812944121 2026-09 baseline 1300 corrupted 1300 total of 11 cells baseline 13338 corrupted 13338 served north value baseline 1212.55 corrupted 1212.55 ``` The last column carries the most important claim of this curriculum: **three of the six classes leave the five served values exactly the same.** The mechanisms of the three are different too. `duplicated` is swallowed at the deduplication stage; sixty double records drop and exactly the intact set remains. `type_drift` is silently repaired by the pipeline's own type conversion; the `consumption` stage turns the text into a number, the conversion succeeds, and neither the rows nor the values change. This second case is especially misleading, because the repair works only for text made of digits only; when an empty or malformed value came from the field, the same row would be dropped silently and again it would be written nowhere. The third, `unit_changed`, is a separate class. In the table **ten summary cells change**, that is, 0.1818 of the fifty-five cells; yet none of the five served values moves. The mechanism is plainly visible in the sub-table: the July total of region `north` goes from 1251 to **1812946491**, the August total from 1119 to **negative 1812944121**. Because the July reading is multiplied by a thousand, the July difference is hugely positive and the August difference is negative of the same magnitude. The total of the eleven cells is **13338** in both cases; the served value is **1212.55** in both cases. The two corrupted periods cancel each other **exactly**, because the total is made of successive differences and the ends of the chain are not corrupted. The sharpest comparison of the table is the first and the fourth row. `late_arrival` and `unit_changed` corrupt **exactly the same number** of summary cells: ten cells each. One changes the served value, the other does not. The number of corrupted intermediate cells does not say whether the corruption will reach the serving; what determines it is the **sign and distribution** of the corruption. From here the third claim of the curriculum follows directly: **looking at the output is not enough.** A served number looking correct is proof not that the pipeline works correctly, only that this number is insensitive to that corruption. #### Summary - None of the six corruption classes is an attack; all of them are accidents on the producer side, and each corrupts a single period. - `screen` sees only `schema_changed` and `dedupe` sees only `duplicated`; the exposure of the two stages is narrow and exact, and they are completely blind to the remaining classes. - The `consumption` stage recognizes no corruption on its own but carries the effect of three classes, and raises the cost of a missing row to **exactly double**: 40 rows to 80, 46 rows to 92, 200 rows to 400. The factor is derived from the definition, not from the run. - The `serve` stage produces nothing other than 5 in any of the six classes; the most looked-at layer of the pipeline can under no condition give notice of corruption by the row count. - Three classes leave the served value exactly the same: `duplicated` is swallowed in deduplication, `type_drift` is repaired by the pipeline's own type conversion, and `unit_changed` never reaches the serving because two periods cancel each other. - `late_arrival` and `unit_changed` corrupt the same number of summary cells (ten cells each); one changes the serving, the other does not. The number of corrupted intermediate cells is not the measure of reaching the serving. #### Next Step The tables of this lesson read the pipeline from a single eye: how many rows went in, how many came out, which cell changed. In production there is no single eye around this pipeline. There is a role that writes the stages of the pipeline and looks at the trace table; there is another role that takes the table at the end of the pipeline as a feature table, builds a model on it and looks at the performance number. The two roles stand at different places of the same pipeline and see the same corruption with **different measures**. The next lesson puts the two measures side by side: for the same six classes, the row trace and the model performance are printed separately. The result is a table that neither role can be comfortable with, because one class leaves no trace in **either of them**. ### Data Engineering and Data Science Distinction Source: https://academia.sh/en/courses/introduction-to-data-engineering/data-engineering-and-data-science-distinction Printing the same six corruption classes side by side on the measure of two roles: the data engineer sees the row trace of five stages, while the model builder sees the mean absolute deviation (baseline 10.832 m3) of a forecast rule that takes the per-region mean served by the pipeline as its feature. The two measures rank the same six classes differently: schema_changed is the noisiest class in the engineer's trace (2200 rows fall to 1800) yet moves the model's deviation by only 0.003; late_arrival removes only 40 rows from the source yet moves the deviation by 0.105, shifting it the most; missing_partition lowers the deviation by 0.003, that is, the corruption improves the measure. unit_changed and type_drift leave no trace in either. The invisibility of the unit change comes from a telescoping identity: the sum of the eleven cells of a region equals the difference of the end indexes, and the corruption does not touch the ends; when the same table is read on the period axis, the July mean comes out as 40592074.02 instead of 26.7. The previous two lessons read the pipeline from a single eye. In production there is no single eye around this pipeline. There is a role that writes the stages, looks at the trace table and is alerted when a row drops; there is another role that takes the table at the end of the pipeline as a feature table, builds a rule or a model on it and looks at the performance number. The two stand at **different places** of the same pipeline. The role distinction itself is not the subject of this curriculum; the boundaries of responsibility were drawn separately in the Introduction to Data Analytics, Introduction to Machine Learning and Introduction to AI Engineering courses. The only thing this lesson adds is **seating the distinction on the stages of the pipeline**: from which stage the number of which role is read, and who measures the stage that remains between the two numbers. When this question is asked, an unpleasant result appears: the same corruption reflects differently on the two measures, the two rank the six classes differently, and one class leaves no trace in **either of them**. - **DL16.** The pipeline, the source, the six corruption classes and the seed are the same as in the previous two lessons. - **DL17. The procedure for role definitions is not repeated**; it was measured in the three introductory courses above. Here the distinction is written only as **the stage from which the measure is read**. - **DL18.** The measure of the data engineer is the **row trace**: the remaining counts of the five stages. This is the same as the table in the first lesson. - **DL19.** The measure of the model builder is the **performance**: the **mean absolute deviation**, in `m3`, of a forecast rule that takes the per-region mean served by the pipeline as its feature. The rule divides the served value of a region by the subscriber count of that region and gives this as the forecast to every subscriber in the region. - **DL20.** The test set is the periods `2026-11` and `2026-12` (**400 rows**). These two periods are not corrupted in any of the six classes; so every difference in the deviation comes from the model's **feature side**, not from the test side. - **DL21.** The model builder takes the feature table **from the pipeline** and does not reach the source. This is the boundary in production: the served layer is the contract between the two roles. - **DL22. The procedure for feature engineering and model evaluation is not repeated**; it was measured in the Data Preparation and Feature Engineering and Model Evaluation, Interpretability and Ethics courses. The model is deliberately at its simplest here. The aim is not for the model to be good, but **what its measure is sensitive to**. - **DL23.** No check; the baseline is the uninstrumented pipeline, the corruption caught is **0**, the lag is **undefined**. The pipeline is deterministic, a second run gives the same table. Time is not written. #### Two Measures, the Same Pipeline The block below builds the pipeline and the six corruption classes as they are, adds the model builder's measure on top, and prints two numbers side by side for each class: is the row trace the same as the intact pipeline, is the mean absolute deviation the same as the intact pipeline. ```python # FICTIONAL pipeline, FICTIONAL corruption classes and a FICTIONAL forecast rule. 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 values stay Turkish: the summary table below is printed in sorted order # dogu/bati/kuzey/guney/merkez = east/west/north/south/central REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] 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 # liters instead of m3 y["unit"] = "m3" # LABEL DID NOT CHANGE counter += 1 out.append(y) elif cls == "missing_partition": if k["region"] == "kuzey": 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 STAGES = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption), ("summary", s4_summary), ("serve", s5_serve)] def pipeline(records): trace, data = [], records for name, f in STAGES: data, dropped = f(data) n = len(data) if hasattr(data, "__len__") else 0 trace.append({"stage": name, "remaining": n, "dropped": dropped}) return data, trace TEST_PERIODS = ["2026-11", "2026-12"] def model_metric(records): """Model builder , takes the per-region mean served by the pipeline as a feature.""" v1, _ = s1_screen(records) v2, _ = s2_dedupe(v1) t, _ = s3_consumption(v2) summ, _ = s4_summary(t) features, _ = s5_serve(summ) n = {} for k in t: n.setdefault(k["region"], set()).add(k["subscriber"]) forecast = {b: features[b] / len(n[b]) for b in features} test = [k for k in t if k["period"] in TEST_PERIODS] return round(sum(abs(k["m3"] - forecast[k["region"]]) for k in test) / len(test), 3), len(test) source = generate_source() base_trace = [a["remaining"] for a in pipeline(source)[1]] base_dev, base_n = model_metric(source) print("test rows", base_n, " baseline deviation", base_dev, " m3") print(f"{'label':<19}{'row trace':>26}{'rows':>7}{'deviation':>11}{'diff':>8}{'model':>7}") print(f"{'intact':<19}{str(base_trace):>26}{'-':>7}{base_dev:>11}{'-':>8}{'-':>7}") for cls in CORRUPTIONS: trace = [a["remaining"] for a in pipeline(corrupt(source, cls)[0])[1]] dev, _ = model_metric(corrupt(source, cls)[0]) print(f"{cls:<19}{str(trace):>26}{('yes' if trace == base_trace else 'no'):>7}" + f"{dev:>11}{round(dev - base_dev, 3):>8}" + f"{('yes' if dev == base_dev else 'no'):>7}") ``` ``` test rows 400 baseline deviation 10.832 m3 label row trace rows deviation diff model intact [2400, 2400, 2200, 55, 5] - 10.832 - - late_arrival [2360, 2360, 2120, 55, 5] no 10.937 0.105 no duplicated [2460, 2400, 2200, 55, 5] no 10.832 0.0 yes schema_changed [2200, 2200, 1800, 45, 5] no 10.835 0.003 no unit_changed [2400, 2400, 2200, 55, 5] yes 10.832 0.0 yes missing_partition [2354, 2354, 2108, 53, 5] no 10.829 -0.003 no type_drift [2400, 2400, 2200, 55, 5] yes 10.832 0.0 yes ``` #### One Who Sees Rows and One Who Sees Performance The two measures rank the same six classes differently, and this difference in ranking is the real number of the role distinction. In the data engineer's trace the noisiest class is `schema_changed`: the consumption output falls from 2200 to **1800**, the summary from 55 to **45**. The change of a field name takes away a sixth of the source, and this is a drop that cannot be overlooked on a dashboard table. The same class moves the model builder's measure by **0.003**, that is, a relative difference of 0.0003 over 10.832. When the model builder looks at this number, they think nothing happened, and they are right: even if a sixth of the training data is lost, the per-region mean comes out almost the same from the remaining nine periods. The opposite direction is more interesting. `late_arrival` removes only **40 rows** from the source, that is, 0.0167 of 2400 (the smallest step of the row ratio is 0.0004, so this ratio can be read to three digits); this is a loss that falls below the band of most row count checks. In return it moves the model builder's deviation by **0.105**, and among the six classes it is the one that shifts it the most. The reason is not in the table but in the definition of the pipeline: the missing rows are not randomly distributed, they are concentrated in the first forty subscribers and pulled the region means down together. A loss that is **small in number but one-directional** is more harmful than a loss that is large in number but balanced, and the row count does not write this distinction. The third row is the cleanest distinction between the two roles. `duplicated` is plainly visible in the engineer's trace: the `screen` output is **2460** instead of 2400. In the model builder's measure, however, the difference is **exactly zero**, because the deduplication stage drops the sixty double records and the intact set reaches the feature table. This is a case where the pipeline does its job right; still, the engineer needs to see it, because the next double record from the same producer may not be caught at the same place. The `missing_partition` row shows the most dangerous side of reading a measure. The deviation **falls** from 10.832 to **10.829**, that is, the corruption **improves** the model's measure. The improvement is not real; since two periods of region `kuzey` (north) are completely lost, the served mean of that region is computed from the remaining nine cells and, by chance, falls slightly closer to the test periods. The result is this: **a measure not getting worse is not proof of health, because the corruption can improve the measure too.** The direction of the sign says nothing about the presence of corruption. #### What Is Visible in Neither The last reading of the table is two rows: `unit_changed` and `type_drift`. In both, the row trace is **exactly the same** as the intact pipeline, and the deviation is **exactly the same**. The routine measure of both roles is completely blind to these two classes. The reason is superficial for `type_drift`: the pipeline turns the text into a number, the conversion succeeds, and neither the rows nor the values change. The only thing that sees it is a type check that **never looks at the value** of the row, and such a check can run at the zeroth step of the pipeline. The reason is structural for `unit_changed`, and the block below shows it. ```python def region_total(records): v1, _ = s1_screen(records) v2, _ = s2_dedupe(v1) t, _ = s3_consumption(v2) summ, _ = s4_summary(t) total = {} for (b, d), v in summ.items(): total[b] = total.get(b, 0) + v return total def period_mean(records): v1, _ = s1_screen(records) v2, _ = s2_dedupe(v1) t, _ = s3_consumption(v2) d = {} for k in t: d.setdefault(k["period"], []).append(k["m3"]) return {a: round(sum(v) / len(v), 2) for a, v in d.items()} first = {k["subscriber"]: k["index"] for k in source if k["period"] == PERIODS[0]} last = {k["subscriber"]: k["index"] for k in source if k["period"] == PERIODS[-1]} s_top, b_top = region_total(source), region_total(corrupt(source, "unit_changed")[0]) print("region ends diff intact summary corrupted summary") for b in sorted(s_top): ends = sum(last[a] - first[a] for a, v in SUBSCRIBERS.items() if v["region"] == b) print(f"{b:<9}{ends:>11}{s_top[b]:>16}{b_top[b]:>19}") s_d, b_d = period_mean(source), period_mean(corrupt(source, "unit_changed")[0]) print("period intact corrupted") for d in PERIODS[5:9]: print(f"{d:<10}{s_d[d]:>10}{b_d[d]:>15}") ``` ``` region ends diff intact summary corrupted summary bati 9164 9164 9164 dogu 9695 9695 9695 guney 11721 11721 11721 kuzey 13338 13338 13338 merkez 14536 14536 14536 period intact corrupted 2026-06 24.08 24.08 2026-07 26.7 40592074.02 2026-08 25.55 -40592021.77 2026-09 28.96 28.96 ``` The three columns of the first table give **the same number** for every region. The left column was computed without going through the pipeline at all, directly from the end points of the source: the last period index of the region's subscribers minus the first period index. The middle and right columns are the sum of the eleven summary cells of the pipeline, with the intact and the corrupted source. The equality of the three is not a measurement result but an **identity**: the sum of successive differences equals the difference of the ends. Because `unit_changed` does not touch the ends, the total cannot change, and the served value, which divides the total by eleven, cannot change either. This is a result known even without a run; the run has only verified it. The second table reads the same feature table **on the period axis**, and there the corruption shouts: the July mean is **40592074.02** instead of 26.7, the August mean is **negative 40592021.77** instead of 25.55. The data is not hiding it; what hides it is **both roles summing over the period axis**. The engineer sums the rows per stage, the model builder sums the values per region; the corruption lives exactly on the axis under these two sums, with two signs that cancel each other. The rule that follows from here is general and applicable: **a measure that sums the axis on which the corruption lives cannot see that corruption.** The count of the role distinction can also be read from here. The engineer's measure is read from the first, second and third stages; the model builder's measure from the output of the fifth stage. Between them stands the **fourth stage**, and the routine measure of neither role reads those fifty-five cells. The place where the corruption sits is exactly there. The boundary is not a question of job definition; it is a question of **which number is read from which stage of the pipeline**, and as long as a stage remains unmeasured between the measures of the two roles, the boundary produces a gap. #### Summary - The boundary between data engineering and data science is not a job definition but **the stage from which the measure is read**: the engineer reads the row trace of the first to third stages, the model builder reads the output of the fifth stage. - The two measures rank the same six classes differently: `schema_changed` is the noisiest class in the trace (2200 rows fall to 1800) yet moves the deviation by only 0.003; `late_arrival` removes only 40 rows yet moves the deviation by 0.105, shifting it the most. - A loss that is small in number but one-directional is more harmful than a loss that is large in number but balanced, and the row count does not write this distinction. - `missing_partition` **lowers** the deviation from 10.832 to 10.829: a corruption can improve the measure, so a measure not getting worse is not proof of health. - `unit_changed` and `type_drift` give exactly the same result in both measures. The invisibility of the unit change comes from a telescoping identity: the sum of the eleven cells of a region equals the difference of the end indexes, and the corruption does not touch the ends. - When the same table is read on the period axis, the July mean comes out as 40592074.02 instead of 26.7; a measure that sums the axis on which the corruption lives cannot see that corruption. #### Next Step For three lessons the pipeline was treated as if it started **where the data arrives**. The source came about with the `generate_source` call, and the corruption was applied to it from outside. **Where the data comes from** was never asked. Yet the `unit_changed` in the last table of this lesson happens through the change of a converter in the field, and which source class can carry such an accident and which cannot is knowable in advance. A pipeline reading from an application database and a pipeline reading from a log stream or an external service are not exposed to the same corruption classes, and do not catch the same corruption at the same stage. The next lesson separates the sources by their **kind** and prints two numbers for each source class: which corruption classes it carries and the **detection lag** of each. ### Source Systems Source: https://academia.sh/en/courses/introduction-to-data-engineering/source-systems Extracting the corruption profile of three source classes and counting after how many stages each corruption class is caught: with an intact delivery all three give 2400 rows and produce the same five served values, and no check fires. They part ways once corruption arrives: the application database carries three of the six classes with a total lag of 4 stages, the log carries five with a total lag of 8, and the external service carries all six with a total lag of again 8. Permissive parsing makes type drift invisible in both sources: 200 rows are corrupted, no check catches it, and the five served values stay exactly the same. The row count check at tolerance 0.05 catches none of the three classes because the largest row diff is 60, that is, 0.0250; lowering the tolerance to 0.0100 catches all three, but when its reference is the producer's own declaration, it catches none at any tolerance. The previous three lessons built the pipeline not from where the data comes but from **where it stands**: two thousand four hundred rows were sitting there, passing through five stages, and the rows each stage dropped were counted. **Where those rows came from** was not asked. Yet corruption is not born inside the pipeline; it is born **on the producer side** and delivered to the pipeline. In terms of measurement, a **source system** (source system) is not one product but three things: the form in which it delivers the data, which corruption classes it **can produce**, and which check can find a reference to lean on. The three source classes are defined by this triple; for each, the number of corruption classes carried and **after how many stages** each class is caught are counted. - **DL41.** The code is **fictional**; no real store, engine or orchestrator is called. The source is the fictional meter readings of a measurement network: 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DL42.** A source class is not a product but a **guarantee set**; no product name is written, the classes are referred to by their kinds. - **DL43.** A corruption class a source class **does not carry** is not run; the carrying decision follows from the guarantee. A source with declared column types cannot produce type drift, and a source that enforces a primary key cannot produce duplicated rows. - **DL44.** The corruption classes are not attacks but producer-side accidents; each corrupts only the period 2026-07. - **DL45.** All seven checks are enabled on every source; the only thing left out is the check that **has no reference**. - **DL46.** Detection lag is written in **stages**; if nothing catches the corruption it is **undefined** and does not enter the total. The step of the row ratio is 1/2400, that is, 0.0004; no ratio is printed on serving. #### Three Source Classes and the Intact Delivery The block below sets up the shared definition, the fictional source, the five-stage pipeline, the six corruption classes and the seven checks, and adds three adapters on top. The application database carries a declared schema and column types and enforces the primary key. The log is append-only text lines: the parser supplies the field names and the type is read permissively. The external service delivers the numeric field as text. ```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 # ---- Six corruption classes. None of them is an attack; all are producer-side accidents. 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 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") # 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" # 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"]) # number turned into text counter += 1 out.append(y) return out, counter # ---- 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 STAGES = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption), ("summary", s4_summary), ("serve", s5_serve)] # ---- Seven checks, each runs AFTER a specific stage. 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_type(data): return any(not isinstance(k.get("index"), int) 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(data): """FLAWED CHECK: looks at the region set , does not see the period.""" return set(k["region"] for k in data) != set(REGION) 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 CHECKS = [("schema", 0, lambda v, t: c_schema(v)), ("type", 0, lambda v, t: c_type(v)), ("uniqueness", 0, lambda v, t: c_uniqueness(v)), ("row_count", 0, lambda v, t: c_row_count(v, t["expected_rows"])), ("partition", 0, lambda v, t: c_partition_completeness(v)), ("partition_period", 0, lambda v, t: c_partition_completeness_period(v)), ("distribution", 4, lambda v, t: c_distribution(v, t["baseline_summary"]))] def run_checks(records, baseline, enabled): """Returns the first detecting check and the detection lag (stage number).""" data, detector, lag = records, None, None for step in range(len(STAGES) + 1): if step > 0: data, _ = STAGES[step - 1][1](data) for cname, cstage, cf in CHECKS: if cname in enabled and cstage == step and cf(data, baseline) and detector is None: detector, lag = cname, step return data, detector, lag def build_baseline(delivery): v, _ = s1_screen(delivery) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) summ, _ = s4_summary(v) serve, _ = s5_serve(summ) return {"expected_rows": len(delivery), "baseline_summary": summ, "baseline_serving": serve} # ---- Three source classes: adapter , set of classes carried , set of checks that have a reference. FIELDS = ("subscriber", "period", "region", "type", "index", "unit") def u_database(delivery): """Declared schema and column type ; the primary key is enforced.""" unique = {} for k in delivery: unique[(k.get("subscriber"), k.get("period"))] = k return list(unique.values()) def u_log(delivery): """Append-only text line ; the parser supplies the field name , the type is read permissively.""" lines = ["|".join(str(k.get(a, "")) for a in FIELDS) for k in delivery] parsed = [] for ln in lines: p = dict(zip(FIELDS, ln.split("|"))) p["index"] = int(p["index"]) if p["index"].isdigit() else p["index"] parsed.append(p) return parsed def u_external_service(delivery): """The numeric field arrives as text ; the parser converts it permissively.""" parsed = [] for k in delivery: y = dict(k) if isinstance(y.get("index"), int): y["index"] = str(y["index"]) # the service's serialization format if isinstance(y.get("index"), str) and y["index"].isdigit(): y["index"] = int(y["index"]) parsed.append(y) return parsed SEVEN = ["schema", "type", "uniqueness", "row_count", "partition", "partition_period", "distribution"] SOURCES = {"app database": (u_database, ["schema_changed", "unit_changed", "missing_partition"], SEVEN), "log": (u_log, ["late_arrival", "duplicated", "unit_changed", "missing_partition", "type_drift"], [a for a in SEVEN if a != "row_count"]), "external service": (u_external_service, CORRUPTIONS, SEVEN)} raw = generate_source() print("source", len(raw), "rows =", len(SUBSCRIBERS), "subscribers x", len(PERIODS), "periods") for name, (u, carries, enabled) in SOURCES.items(): delivery = u(raw) base = build_baseline(delivery) served, detector, lag = run_checks(delivery, base, enabled) print(f"{name:<16} delivered {len(delivery)} rows serving {served == base['baseline_serving']} " f"enabled checks {len(enabled)} classes carried {len(carries)} detector {detector}") print("intact serving", build_baseline(raw)["baseline_serving"]) print("external service raw delivery type check", c_type([dict(k, index=str(k["index"])) for k in raw])) ``` ``` source 2400 rows = 200 subscribers x 12 periods app database delivered 2400 rows serving True enabled checks 7 classes carried 3 detector None log delivered 2400 rows serving True enabled checks 6 classes carried 5 detector None external service delivered 2400 rows serving True enabled checks 7 classes carried 6 detector None intact serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} external service raw delivery type check True ``` With an intact delivery the three source classes are **indistinguishable**: all three give 2400 rows, produce the same five served values, and no check fires in any of them. The only visible difference is that the number of enabled checks in the log drops from seven to six, because there is no reference called an expected row count there. The last line shows how the source class gets mixed into the check. Because the external service delivers the numeric field as text, the type check is red in all twelve periods **while there is no corruption at all**. A check that is permanently red is a check that is turned off. The only way to turn it green is permissive conversion in the parser, and the price of that is paid below. #### Corruption Profile per Source Class The block below runs, for each source class, only the corruptions **that class can carry**. ```python profile, summary = {}, {} for name, (u, carries, enabled) in SOURCES.items(): base = build_baseline(u(raw)) total, missed, late, silent = 0, 0, 0, 0 for cls in CORRUPTIONS: if cls not in carries: continue corrupted, n = corrupt(raw, cls) delivery = u(corrupted) served, detector, lag = run_checks(delivery, base, enabled) profile[(name, cls)] = (n, abs(len(delivery) - 2400), detector, lag, served == base["baseline_serving"]) silent += served == base["baseline_serving"] if detector is None: missed += 1 else: total, late = total + lag, late + (lag >= 4) summary[name] = (len(carries), total, missed, late, silent) print("corruption app database log external service") for cls in CORRUPTIONS: h = [] for name in SOURCES: v = profile.get((name, cls)) h.append("." if v is None else ("not caught" if v[2] is None else f"{v[2]}/{v[3]}")) print(f"{cls:<18}{h[0]:>20}{h[1]:>20}{h[2]:>20}") print("\nsource class classes corrupted rows total lag not caught fell to stage 4 serving unchanged") for name, v in summary.items(): rows = sum(profile[(name, c)][0] for c in CORRUPTIONS if (name, c) in profile) print(f"{name:<16}{v[0]:>9}{rows:>16}{v[1]:>11}{v[2]:>12}{v[3]:>17}{v[4]:>19}") print("\nrow count check tolerance 0.0500 -> 120 rows") print("class row diff ratio tolerance 0.0500 tolerance 0.0100 reference producer") for cls in ("late_arrival", "duplicated", "missing_partition"): t = corrupt(raw, cls)[0] f = abs(len(t) - 2400) d = ["misses", "catches"] print(f"{cls:<18}{f:>10}{f / 2400:>8.4f}" f"{d[c_row_count(t, 2400)]:>18}{d[c_row_count(t, 2400, 0.01)]:>18}" f"{d[c_row_count(t, len(t), 0.01)]:>20}") ``` ``` corruption app database log external service late_arrival . distribution/4 distribution/4 duplicated . uniqueness/0 uniqueness/0 schema_changed schema/0 . schema/0 unit_changed distribution/4 distribution/4 distribution/4 missing_partition partition_period/0 partition_period/0 partition_period/0 type_drift . not caught not caught source class classes corrupted rows total lag not caught fell to stage 4 serving unchanged app database 3 446 4 0 1 1 log 5 546 8 1 2 3 external service 6 746 8 1 2 3 row count check tolerance 0.0500 -> 120 rows class row diff ratio tolerance 0.0500 tolerance 0.0100 reference producer late_arrival 40 0.0167 misses catches misses duplicated 60 0.0250 misses catches misses missing_partition 46 0.0192 misses catches misses ``` #### Reading the Profile The first table lets us read **what a source class cannot produce**. The three dots in the application database row are three guarantees: a consistent read makes the late-arriving row, the primary key makes the duplicated row, and the declared column type makes type drift impossible on the producer side. These guarantees are not free, but they **halve the number of classes the pipeline has to measure**: from six to three, and the total lag from 8 stages to 4. The log and the external service make up the same total lag, **8 stages**, from different classes. The log does not carry schema change, because the field names are given by the parser, not by the producer; the external service carries all six classes and has the schema change caught at stage 0. A single total number does not justify the choice of a source; the rows of the table do. The last row is the quietest result of the lesson. Type drift is **not caught** in either permissive source: 200 rows are corrupted, the parser turns the text into a number, the type check stays green, and the five served values come out exactly the same. It leaves no trace, because the parser has erased the reference of the check that would have seen it. In these two sources the number of classes whose serving does not change is **three**: duplicated, unit change and type drift. #### A Check Whose Reference Comes from the Producer The third table tests the row count check on its own. The default tolerance is **0.05**, that is, 120 rows. The largest of the three classes that change the row count moves only 60 rows, that is, **0.0250**: the tolerance is twice the largest signal it wants to measure. This check catches none of the three and stays green through all twelve periods; a check being green does not mean that anything was measured. Lowering the tolerance to **0.0100** has all three caught at stage 0 and pulls the lag of the late row from four to zero; there is no false-alarm cost either, because the intact delivery is exactly 2400 rows. Still, this fix can be applied to only one of the three sources: the log has no reference, and the external service has one but it is **the producer's own declaration**; the service says how many rows it sent, that is what is received, and the ratio comes out 0.0000. The value of a check comes not from its threshold but from **where its reference comes from**. #### Summary - A source class is defined not by a product name but by three things: its adapter, the corruption classes it can carry and the checks that have a reference. With an intact delivery all three give the same number; the price shows only when corruption arrives. - A guarantee on the producer side lowers the number of classes carried: a declared type, a primary key and a consistent read bring the application database from six classes to three, and its total lag is 4 instead of 8. - The log and the external service make up the same total lag from different classes; the choice of source is justified not by the total number but by the rows of the profile. - Permissive parsing erases the reference of the type check: 200 rows are corrupted, nothing catches it, and the five served values stay the same; the number of classes that do not change the serving rises to three. - The row count check at tolerance 0.05 catches no class, because the largest row diff moves 0.0250; if the reference is the producer's declaration, no tolerance catches anything. #### Next Step The profile of this lesson was extracted **with all twelve periods at hand**: the corrupted period was sitting inside the pipeline and the checks could compare it with the remaining eleven periods. A pipeline does not work like that; it is tied to a schedule, and the schedule determines when the corrupted period will become visible. The next lesson runs the same pipeline in two modes: batch mode, which processes the twelve periods in one go, and stream mode, which processes a period when it arrives. The trade-off to be measured is this: in batch mode the corruption becomes visible **one period later**, in stream mode **in the same period**; in return, stream mode grows the pipeline's number of stages and the carried state. ### Batch and Stream Processing Source: https://academia.sh/en/courses/introduction-to-data-engineering/batch-and-stream-processing Running the same pipeline in two modes and counting the trade-off between freshness and detection lag: in batch mode the corruption becomes visible one period later, in the 2026-08 run, in stream mode in the same period, in the 2026-07 window. In return, stream mode raises the number of stages from 5 to 8 and the carried state from 0 to 2600 entries; on the other hand, the rows read fall from 13200 to 2400. The period window fixes the flawed partition check: the missing partition that batch mode never catches is caught at stage 0 in stream mode. For the late-arriving row the trade-off reverses: batch mode takes the row back in the next run and leaves all 55 summary cells correct, while stream mode drops the 40 rows at the watermark and permanently dirties 10 cells, yet the five served values stay exactly the same in both modes. The previous lesson extracted the profile of the three source classes **with all twelve periods at hand**: the corrupted period was sitting inside the pipeline and could be compared with the remaining eleven periods. A pipeline does not work like that; it is tied to a schedule, and the schedule determines **when the corrupted period will be looked at**. This lesson runs the same pipeline in two modes. **Batch processing** (batch processing) runs after a period closes and processes every row that has arrived up to that moment from the beginning; **stream processing** (stream processing) processes a row in the period in which it arrives and carries state to the next period. Both use the same five stages, the same six corruption classes and the same checks; their only difference is **the window**. - **DL47.** The code is **fictional**; no real stream system, queue product or orchestrator is called. The source is the fictional meter readings of the previous lesson: 200 subscribers, 12 periods, 2400 rows, seed 20260218. - **DL48. The queue, retry and latency procedures are not repeated**; they were measured in the Caching, Queues and Asynchronous Processing course of the Backend Development curriculum and in the Application Layer and Service Interaction course of the System Design and Distributed Systems curriculum. The only thing added here is **in which period the corruption is seen**. - **DL49.** In batch mode a run processes every row that has **arrived** up to that moment from the beginning; in stream mode a row is processed **in the period in which it arrives**, and the watermark (watermark) drops a row whose window has closed. - **DL50.** In this lesson a late-arriving row counts not as "never arrived" but as **arrived one period later**. The corruption class is the same; the only thing added is the row's **arrival period**. - **DL51.** The same five checks are enabled in both modes: schema, type, uniqueness, partition and distribution. The partition check is the flawed one: it looks at the region set and does not see the period. - **DL52.** Lag is written in two units: **periods** and **stages**. The step of the summary cell ratio is 1/55, that is, 0.0182. #### Two Modes, One Pipeline The block below sets up the shared definition and adds two runners. `batch` processes, at the start of each period, the rows that have arrived up to that moment **from the beginning** and carries nothing. `stream` processes each period in its own window and carries two things: the set of keys seen and the last index per subscriber: information the deduplication and consumption stages find for free in batch mode. ```python # FICTIONAL. No real stream system, queue or orchestrator is called; the counter # 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 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 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") # 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" # 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"]) # number turned into text 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: totals[(k["region"], k["period"])] = totals.get((k["region"], k["period"]), 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 STAGES = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption), ("summary", s4_summary), ("serve", s5_serve)] 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_type(data): return any(not isinstance(k.get("index"), int) 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_partition_completeness(data): """Looks at the region set , does not see the period: flawed over the whole history , not in the window.""" return set(k["region"] for k in data) != set(REGION) 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 CHECKS = [("schema", 0, lambda v, t: c_schema(v)), ("type", 0, lambda v, t: c_type(v)), ("uniqueness", 0, lambda v, t: c_uniqueness(v)), ("partition", 0, lambda v, t: c_partition_completeness(v)), ("distribution", 4, lambda v, t: c_distribution(v, t))] def full_pipeline(records): v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) summ, _ = s4_summary(v) return summ def deliver(records, cls): """Gives each row an arrival period ; the late-arriving row arrives one period later.""" corrupted, n = corrupt(records, cls) data = [dict(k, arrival=k["period"]) for k in corrupted] if cls == "late_arrival": data += [dict(k, arrival="2026-08") for k in records if k["period"] == "2026-07" and k["subscriber"] <= "A040"] return data, n STAGES_BATCH = [a for a, _ in STAGES] STAGES_STREAM = ["watermark"] + STAGES_BATCH + ["state_keys", "state_index"] def batch(data, baseline): """Run , processes ALL rows that have ARRIVED up to the end of the previous period from the beginning.""" rows_read, detector, stage, period = 0, None, None, None for i, p in enumerate(PERIODS): arrived = [k for k in data if PERIODS.index(k["arrival"]) < i] rows_read += len(arrived) v = arrived for step in range(6): if step > 0 and v: v, _ = STAGES[step - 1][1](v) for cname, cstage, cf in CHECKS: if cstage == step and detector is None and v and cf(v, baseline): detector, stage, period = cname, step, p return {"rows_read": rows_read, "detector": detector, "stage": stage, "period": period, "state": 0, "late_dropped": 0, "cells": full_pipeline(data)} def stream(data, baseline): """Period window: a row is processed in the period it arrives , state is carried to the next period.""" seen, last_index, cells = set(), {}, {} rows_read, late_dropped, state = 0, 0, 0 detector, stage, period = None, None, None for p in PERIODS: window = [k for k in data if k["arrival"] == p] rows_read += len(window) v = [k for k in window if k["period"] == p] # watermark: the window closed late_dropped += len(window) - len(v) for step in range(5): if step == 1: v, _ = s1_screen(v) elif step == 2: fresh, new_keys = [], set() for k in v: key = (k["subscriber"], k["period"]) if key not in seen and key not in new_keys: new_keys.add(key) fresh.append(k) seen |= new_keys # state: seen key v = fresh elif step == 3: c = [] for k in v: try: c.append(dict(k, m3=int(k["index"]) - int(last_index[k["subscriber"]]))) except (KeyError, TypeError, ValueError): pass for k in v: try: last_index[k["subscriber"]] = int(k["index"]) # state: last index except (TypeError, ValueError): pass v = c elif step == 4: totals = {} for k in v: totals[(k["region"], p)] = totals.get((k["region"], p), 0) + k["m3"] cells.update(totals) v = totals for cname, cstage, cf in CHECKS: if cstage == step and detector is None and v and cf(v, baseline): detector, stage, period = cname, step, p state = max(state, len(seen) + len(last_index)) return {"rows_read": rows_read, "detector": detector, "stage": stage, "period": period, "state": state, "late_dropped": late_dropped, "cells": cells} raw = generate_source() BASELINE = full_pipeline(raw) intact = [dict(k, arrival=k["period"]) for k in raw] sb, ss = batch(intact, BASELINE), stream(intact, BASELINE) print("baseline summary", len(BASELINE), "cells , serving", s5_serve(BASELINE)[0]) print("mode stages rows read carried state detector") for name, r, n in (("batch", sb, len(STAGES_BATCH)), ("stream", ss, len(STAGES_STREAM))): print(f"{name:<6} {n:>6} {r['rows_read']:>13} {r['state']:>14} {r['detector']}") ``` ``` baseline summary 55 cells , serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} mode stages rows read carried state detector batch 5 13200 0 None stream 8 2400 2600 None ``` With the intact run both modes produce the same 55 summary cells and the same five served values, and no check fires in either. The cost row is read here. Batch mode reads **13200 rows** in twelve runs, because every run processes the accumulated history from the beginning; stream reads each row once, **2400 rows**. In return, stream raises the pipeline's stages **from 5 to 8** (the watermark, and the storing of the seen key and of the last index) and carries up to **2600 entries**: 2400 keys and 200 subscriber indexes. The state carried by batch mode is zero. #### Detection Period The block below runs the six corruption classes in both modes and writes in which period and by which check each one is caught. ```python outcome = {} for cls in CORRUPTIONS: data, n = deliver(raw, cls) outcome[cls] = (n, batch(data, BASELINE), stream(data, BASELINE)) def label(r): if r["detector"] is None: return "not caught" return f"{r['detector']}/{r['period']}/{PERIODS.index(r['period']) - PERIODS.index('2026-07')}" def dirty(h): return sum(1 for key in set(h) | set(BASELINE) if h.get(key) != BASELINE.get(key)) print("corruption corrupted batch: check/period/lag stream: check/period/lag") for cls, (n, b, s) in outcome.items(): print(f"{cls:<18} {n:>9} {label(b):<28} {label(s)}") print("\ncorruption dirty cells/55 serving changed in stream") print(" batch stream batch stream dropped rows") for cls, (n, b, s) in outcome.items(): e = ["no", "yes"] print(f"{cls:<18} {dirty(b['cells']):>7} {dirty(s['cells']):>7} " f"{e[s5_serve(b['cells'])[0] != s5_serve(BASELINE)[0]]:>8}" f"{e[s5_serve(s['cells'])[0] != s5_serve(BASELINE)[0]]:>7} {s['late_dropped']:>12}") ``` ``` corruption corrupted batch: check/period/lag stream: check/period/lag late_arrival 40 distribution/2026-08/1 distribution/2026-07/0 duplicated 60 uniqueness/2026-08/1 uniqueness/2026-07/0 schema_changed 200 schema/2026-08/1 schema/2026-07/0 unit_changed 200 distribution/2026-08/1 distribution/2026-07/0 missing_partition 46 not caught partition/2026-07/0 type_drift 200 type/2026-08/1 type/2026-07/0 corruption dirty cells/55 serving changed in stream batch stream batch stream dropped rows late_arrival 0 10 no no 40 duplicated 0 0 no no 0 schema_changed 10 10 yes yes 0 unit_changed 10 10 no no 0 missing_partition 2 2 yes yes 0 type_drift 0 0 no no 0 ``` #### Looking One Period Earlier The first table gives the trade-off in a single row: in all five classes that are caught, batch mode says **2026-08** and stream mode says **2026-07**. The period lag is **1** in batch mode and **0** in stream. The stage lags are exactly the same: schema, type and uniqueness are caught at stage 0, and distribution at stage 4. The mode does not change at which **stage** the corruption is caught; it changes only in which **period** one looks. Freshness here is not a serving property but a check property. What a period means shows in the output: in batch mode the corrupted period has been served once by the time it is caught: 2026-07 has closed, the summary has been written, the served value has been read. This difference cannot be closed with a tolerance setting, because the lag of batch mode comes not from the check but from **the schedule**. The second table shows what is independent of the mode. The duplicated row, type drift and unit change **do not change the five served values in either mode**; the first two do not even leave a dirty cell, the unit change dirties ten cells but the serving looks flawless. The only thing that sees these three is a check. #### The Window's Fix of the Check The fifth row is the most unexpected result of the table. The missing partition is **never caught** in batch mode and is caught **at stage 0** in stream mode, and moreover **the same check code** runs in both modes. The partition check looks at the region set and does not see the period; in the batch run all twelve periods are in front of it, the set is complete because north shows up in the other eleven periods, and the check stays green. In the stream run the set put in front of the same check is **the window of a single period**; there north is absent and the check fires. The correctness of a check comes not only from its code but also from **the set put in front of it**. Although this looks like an advantage of stream mode, it is a side effect; the way to get the same result in batch mode is not to change the mode but to write the check period-aware. #### The Cost of State The first row of the second table reverses the trade-off. When the forty late-arriving rows arrive one period later, batch mode processes the accumulated history from the beginning and **backfills** the rows: all 55 summary cells stay correct. Stream drops the same rows at the watermark and, since the window has closed, never looks at them again. The result is **10 dirty summary cells**, that is, 0.1818 of the cells. The mechanism is the carried state. Because the 2026-07 readings of forty subscribers were never processed, the last-index state stays at 2026-06; when 2026-08 arrives, the consumption is computed from a two-period difference. In batch mode such a row **drops**, because the previous period is not in the table; in stream it is **computed wrongly**, because the state holds an old value. A missing row is a countable loss, not a wrong value. The last column is the real warning of this section. Despite ten dirty cells, **the five served values are exactly the same in both modes**: what falls short in July is in excess in August, and the average of twelve periods does not move. Stream mode had caught this corruption in 2026-07; had it not, it could not have been found by looking at any output. #### Summary - Batch mode processes the accumulated history from the beginning after a period closes, stream processes a row in the period in which it arrives; both use the same five stages and the same five checks, and their only difference is the window. - In all five classes that are caught, the period lag of batch mode is 1 and that of stream is 0; the stage lags are exactly the same. The mode changes not at which stage but in which period the corruption is seen. - The cost of stream is the number of stages rising from 5 to 8 and the carrying of 2600 entries; in return, the rows read fall from 13200 to 2400. - The period window fixes the flawed partition check: the same code never catches the missing partition in batch mode and catches it at stage 0 in stream mode. - For the late-arriving row the trade-off reverses: batch mode leaves all 55 cells correct, stream drops 40 rows and dirties 10 cells; even so, the five served values are the same in both modes. #### Next Step The pipeline was built and counted over five lessons: its stages were determined, the rows each stage drops were written, where the roles stand was separated, the data was broken down into source classes and the schedule it would run on was measured in two modes. One thing was never asked: **whether the row the pipeline carries is correct**. Everything counted is where a row came from, where it goes and when it is looked at; not a single measure that looks at the row itself was written. The checks were found ready-made, but **where they come from** was not defined. The next topic does this: it turns the quality dimensions into measurable checks, counts the detection lag of each check and measures the check's **own flaw**. ## Quality and Principles Turning quality dimensions into running checks and counting both what each check catches and its own flaw: the detection lag of six corruption classes and the three classes a flawed check set never catches; a partition check that stays green even though one period of one region was never written, because it looks at the region set and not at the period; the rows dropped by a schema change, the stage at which a data contract catches it, and the cost of the contract itself; a pipeline that is not idempotent duplicating rows on reprocessing, and the served number staying unchanged because the deduplication stage absorbs the duplicates; the stage and row count one must visit to go from a corrupted served value back to the source, compared with and without a lineage record; how many of the five stages carry an owner, and who receives a corruption caught at an ownerless stage; and the number of periods a retention policy makes impossible to audit retroactively. ### Data Quality Dimensions Source: https://academia.sh/en/courses/introduction-to-data-engineering/data-quality-dimensions Turning quality dimensions into running checks and counting both what each check catches and its own flaw: the detection lag of the six corruption classes is 4, 0, 0, 4, 0 and 0 stages in order; with a flawed check set three classes are never caught; because the partition check looks at the region set, it stays green when a single period of one region was never written; a unit change moves ten cells of the summary and leaves the served values exactly the same. Over five lessons the pipeline was built and counted: how many rows each stage drops, how batch and stream processing change freshness. All of those numbers answered a single question, how many rows. At no stage was the question asked whether a row is **correct**. These are two separate questions, and the gap between them is a flaw peculiar to a data pipeline: when the pipeline crashes it tells you, **when it carries wrong data it says nothing.** This lesson turns the quality dimensions into running checks, counts how many stages later each of the six corruption classes is caught, and shows in the same set that a check can itself be flawed. - **QP1.** The source, the five-stage pipeline, the corruption classes and the checks are **fictional**; no real database, file format or orchestrator is called. The seed is **20260218**, and the pipeline does not depend on the run. - **QP2.** **The corruption classes are not attacks**, they are producer-side accidents: a file that arrives late, a payload sent twice, a field whose name changed, a meter whose unit changed. - **QP3.** The baseline is the **uninstrumented pipeline**: corruption caught **0**, lag **undefined**. - **QP4.** Each corruption damages **a single period** (`2026-07`); the remaining eleven periods stay intact. - **QP5.** The procedure for cleaning, missing values and outliers was measured in the Data Collection and Preparation course, and leakage and label noise in the Data Preparation and Feature Engineering course, and **they are not repeated.** - **QP6.** Lag is written in **stages**, not in raw time: stage 0 is the source itself, and 1 through 5 are the stages of the pipeline. - **QP7.** Resolution is the lower bound of the measurement band: the step of the row share is **0.0004**, of a summary cell **0.0182**, of a served value **0.2000**. **No share is printed on the serving output.** #### A Dimension Is a Check, Not a Sentence A **data quality dimension** is usually written as a sentence: "the data must be complete", "the data must be consistent". These sentences cannot be measured, because they do not say on which set, at which stage and against what to look. A dimension is measured only once it is turned into a **predicate**: a function that runs after a specific stage and returns true or false. | Dimension | What the predicate asks | Check | Stage | |---|---|---|---| | completeness | did every expected row arrive | `row_count`, `partition`, `partition_period` | 0 | | uniqueness | does the same key appear twice | `uniqueness` | 0 | | validity | do the fields carry the expected name and type | `schema`, `type` | 0 | | consistency | are the values in the pattern of earlier periods | `distribution` | 4 | | accuracy | does the value reflect reality | **none** | — | The last row is in the table on purpose. **Accuracy** cannot be measured from inside the pipeline: to know how many cubic meters the meter really passed, a measurement outside the pipeline is needed, and this pipeline has no such source. What can be measured is **consistency**; a row in a quality report that says "accuracy ninety-nine percent" has most often measured consistency and called it accuracy. The stage column says a second thing: `distribution` looks not at the source but at the summary that the fourth stage produces. #### Common Definition The first block sets up the source, the five-stage pipeline, the six corruption classes and the seven checks. `instrumented_pipeline` runs the enabled checks after every stage; it returns the **first detecting check** and the **detection lag**. ```python # FICTIONAL. Meter readings of the measurement network, the five-stage pipeline, corruption # classes and checks are defined in the lesson; no real store 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 def pick(u, w): t = 0.0 for i, x in enumerate(w): t += x if u < t: return i return len(w) - 1 # ---- FICTIONAL source: meter readings of the measurement network, twelve periods # region values stay Turkish: the region set is printed in sorted order # dogu/bati/kuzey/guney/merkez = east/west/north/south/central REGION = ["dogu", "bati", "kuzey", "guney", "merkez"] 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)} # One reading per subscriber per period. Unit m3 , the meter index only increases. 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 # ---- FICTIONAL corruption classes. None of them is an attack; all are # producer-side accidents that happen in everyday operation. 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 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") # 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" # LABEL DID NOT CHANGE , silent corruption counter += 1 out.append(y) elif cls == "missing_partition": if k["region"] == "kuzey": # 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"]) # number turned into text counter += 1 out.append(y) return out, counter # ---- PIPELINE: five stages. Each stage counts the rows it drops. def s1_screen(records): """A row that lacks a required field 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. 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 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: 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 STAGES = [("screen", s1_screen), ("dedupe", s2_dedupe), ("consumption", s3_consumption), ("summary", s4_summary), ("serve", s5_serve)] def pipeline(records): trace, data = [], records for name, f in STAGES: data, dropped = f(data) n = len(data) if hasattr(data, "__len__") else 0 trace.append({"stage": name, "remaining": n, "dropped": dropped}) return data, trace # ---- CHECKS: each one runs AFTER a specific stage 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_type(data): return any(not isinstance(k.get("index"), int) 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(data): """FLAWED CHECK: looks at the region set , does not see the period.""" return set(k["region"] for k in data) != set(REGION) def c_partition_completeness_period(data): """CORRECT CHECK: is every region present 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) 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 # Check registry: (name , stage , function). Stage 0 = source , 1..5 = STAGES order. CHECKS = [ ("schema", 0, lambda v, t: c_schema(v)), ("type", 0, lambda v, t: c_type(v)), ("uniqueness", 0, lambda v, t: c_uniqueness(v)), ("row_count", 0, lambda v, t: c_row_count(v, t["expected_rows"])), ("partition", 0, lambda v, t: c_partition_completeness(v)), ("partition_period", 0, lambda v, t: c_partition_completeness_period(v)), ("distribution", 4, lambda v, t: c_distribution(v, t["baseline_summary"])), ] def instrumented_pipeline(records, baseline, enabled=None): """Runs the pipeline and applies the enabled checks after every stage. Returns: serving , trace , first detecting check and detection lag (stage).""" enabled = [a for a, _, _ in CHECKS] if enabled is None else enabled data, trace, detector, lag = records, [], None, None for step in range(len(STAGES) + 1): if step > 0: name, f = STAGES[step - 1] data, dropped = f(data) trace.append({"stage": name, "remaining": len(data), "dropped": dropped}) for cname, cstage, cf in CHECKS: if cname not in enabled or cstage != step: continue if cf(data, baseline): if detector is None: detector, lag = cname, step serving = data if isinstance(data, dict) and data and isinstance( list(data.values())[0], float) else None return serving, trace, detector, lag def build_baseline(source): v1, _ = s1_screen(source) v2, _ = s2_dedupe(v1) v3, _ = s3_consumption(v2) summ, _ = s4_summary(v3) serve, _ = s5_serve(summ) return {"expected_rows": len(source), "baseline_summary": summ, "baseline_serving": serve} ``` #### Detection Lag The second block turns the rule into a number. For each corruption class the pipeline runs twice: once with all seven checks enabled, once with the set `["schema", "type", "uniqueness", "partition"]`. The second set is flawed, because the period-aware completeness check and the distribution check are off. The last column says whether the five served values stayed exactly the same as the baseline. ```python SOURCE_RECORDS = generate_source() BASELINE = build_baseline(SOURCE_RECORDS) ALL_CHECKS = [a for a, _, _ in CHECKS] FLAWED_SET = ["schema", "type", "uniqueness", "partition"] _, trace = pipeline(SOURCE_RECORDS) print("healthy pipeline:", " ".join(f"{i['stage']}={i['remaining']}" for i in trace)) print("serving:", BASELINE["baseline_serving"]) print("check fired on healthy pipeline:", instrumented_pipeline(SOURCE_RECORDS, BASELINE, ALL_CHECKS)[2]) print() print(f"{'corruption':<18}{'rows':>6}{'detector':>17}{'lag':>9}" f"{'flawed set':>14}{'serving':>9}") for cls in CORRUPTIONS: corrupted, n = corrupt(SOURCE_RECORDS, cls) serving, _, det, lag = instrumented_pipeline(corrupted, BASELINE, ALL_CHECKS) _, _, fdet, _ = instrumented_pipeline(corrupted, BASELINE, FLAWED_SET) status = "same" if serving == BASELINE["baseline_serving"] else "changed" print(f"{cls:<18}{n:>6}{det:>17}{lag:>9}{str(fdet):>14}{status:>9}") ``` ``` healthy pipeline: screen=2400 dedupe=2400 consumption=2200 summary=55 serve=5 serving: {'merkez': 1321.45, 'kuzey': 1212.55, 'dogu': 881.36, 'bati': 833.09, 'guney': 1065.55} check fired on healthy pipeline: None corruption rows detector lag flawed set serving late_arrival 40 distribution 4 None changed duplicated 60 uniqueness 0 uniqueness same schema_changed 200 schema 0 schema changed unit_changed 200 distribution 4 None same missing_partition 46 partition_period 0 None changed type_drift 200 type 0 type same ``` The first three lines give the baseline: on the healthy source the five stages go down from 2400 to 5 values and **no check fires.** This is the proof that the check set does not falsely block a valid row. The lag column is the rule itself. Four classes are caught at **stage 0**, in the source itself; the pipeline is never contaminated. The other two are caught at **stage 4**: `late_arrival` and `unit_changed` become visible only through the distribution check that looks at the summary of the fourth stage. The four stages in between are a unit of cost, because the corrupted row was screened, deduplicated, turned into consumption and aggregated into the summary through those four stages; what will have to be backfilled after the fix is the output of those same four stages. The last column carries the third claim of the course: **three classes leave the served number exactly the same.** After `duplicated`, `unit_changed` and `type_drift` the five served values are identical to the baseline. No observation that looks at the output can see these three; the only evidence is the check itself. `type_drift` is the most extreme example: when the index turns into text, the consumption stage converts it back to a number, and **not a single row is dropped.** The fourth column measures the flawed set: **three classes are never caught.** `late_arrival` and `unit_changed` because the distribution check is off, and `missing_partition` because the `partition` check, which is on, is flawed. The team that built this set wrote four checks, all four stayed green, and three of the six classes went unmeasured. #### A Check That Stays Green The third block opens up the flaw. In the `missing_partition` corruption the `2026-07` period of the `kuzey` (north) region is never written. `c_partition_completeness` compares the region set with the expected set; because `kuzey` appears in the other eleven periods, the set is complete and the check **passes.** `c_partition_completeness_period` looks at the pair of region and period, and fires on the same source. ```python def interim_summary(records): """Output of the fourth stage: total per region and period.""" v, _ = s1_screen(records) v, _ = s2_dedupe(v) v, _ = s3_consumption(v) return s4_summary(v)[0] corrupted, n = corrupt(SOURCE_RECORDS, "missing_partition") print("rows not written:", n, " kuzey rows of 2026-07:", sum(1 for k in corrupted if k["region"] == "kuzey" and k["period"] == "2026-07")) print("region set in source:", sorted(set(k["region"] for k in corrupted))) print("flawed partition check:", c_partition_completeness(corrupted), " period-aware check:", c_partition_completeness_period(corrupted)) summary_1 = interim_summary(corrupted) print("summary cells:", len(summary_1), " periods of kuzey in summary:", sum(1 for b, d in summary_1 if b == "kuzey")) print("serving kuzey:", s5_serve(summary_1)[0]["kuzey"], " baseline:", BASELINE["baseline_serving"]["kuzey"]) print() summary_2 = interim_summary(corrupt(SOURCE_RECORDS, "unit_changed")[0]) for key in (("kuzey", "2026-06"), ("kuzey", "2026-07"), ("kuzey", "2026-08")): print("summary", key, "baseline:", BASELINE["baseline_summary"][key], " corrupted:", summary_2[key]) print("changed summary cells:", sum(1 for a in summary_2 if summary_2[a] != BASELINE["baseline_summary"][a]), "/", len(summary_2)) print("serving kuzey:", s5_serve(summary_2)[0]["kuzey"], " baseline:", BASELINE["baseline_serving"]["kuzey"]) ``` ``` rows not written: 46 kuzey rows of 2026-07: 0 region set in source: ['bati', 'dogu', 'guney', 'kuzey', 'merkez'] flawed partition check: False period-aware check: True summary cells: 53 periods of kuzey in summary: 9 serving kuzey: 1218.67 baseline: 1212.55 summary ('kuzey', '2026-06') baseline: 1041 corrupted: 1041 summary ('kuzey', '2026-07') baseline: 1251 corrupted: 1812946491 summary ('kuzey', '2026-08') baseline: 1119 corrupted: -1812944121 changed summary cells: 10 / 55 serving kuzey: 1212.55 baseline: 1212.55 ``` The first group states the sentence: **a check being green does not mean it measured.** The region set has five elements, and the flawed check returns `False`. In the same source 46 rows never arrived, the summary has dropped from 55 cells to 53, and the served value of `kuzey` has shifted from 1212.55 to 1218.67 — given that the serving resolution is 0.2000, a measurable difference of more than thirty steps, but nobody is looking. The number of periods of `kuzey` in the summary has fallen to **9**, because a period never being written drops **the period that follows it as well:** consumption works on the period difference, and the `2026-08` rows no longer have a previous period. The rows a stage drops are not the same number as the rows the corruption directly deleted. The second group opens the quietest corruption. In `unit_changed` the meter reads liters instead of cubic meters but leaves the unit label as `m3`. In the summary, the `kuzey` total of `2026-07` rises from 1251 to **1812946491**, and the total of `2026-08` falls from 1119 to **−1812944121**. The mechanism is cancellation: because the index grew a thousandfold in the corrupted period, the difference of that period is inflated, and because the index returns to its old scale in the following period, a negative difference of the same size appears. **Ten cells** of the summary move. When the mean over the region's eleven periods in the summary is taken, the two cells cancel each other and the served value stays exactly at **1212.55**; while the serving looks flawless, two periods are garbage. #### Summary - A quality dimension counts as measured only once it is turned into a running check that runs after a specific stage; **accuracy** cannot be measured in this pipeline, and what is measured instead is **consistency**. - On the healthy source the five stages bring 2400 rows down to 5 served values and none of the seven checks fires; a valid row is not falsely blocked. - The detection lag of the six corruption classes is 4, 0, 0, 4, 0 and 0 stages in order; four stages mean a pipeline whose intermediate tables are contaminated and have to be backfilled. - With the set `["schema", "type", "uniqueness", "partition"]` three classes are never caught: `late_arrival`, `unit_changed` and `missing_partition`. Because `partition` looks at the region set and does not see the period, it stays green on a source in which 46 rows were never written. - Three classes leave the five served values exactly the same; although `unit_changed` moves ten cells of the summary between 1812946491 and −1812944121, the mean stays at 1212.55. #### Next Step The fastest-caught corruption of this lesson was `schema_changed`: when the field name changed, the `schema` check fired at stage 0. But the list of fields the check looks at was buried inside the code, and who wrote that list was never asked. When a producer changes the name of a field it receives no error on its own side; what breaks is the consumer. The next lesson turns this list into a **data contract**, counts how many rows are dropped when a field name changes and at which stage the contract catches the corruption, and then measures the cost of the contract itself. ### Schema Contracts Source: https://academia.sh/en/courses/introduction-to-data-engineering/schema-contracts Counting whom the schema contract between producer and consumer breaks and what the contract itself costs: when a field is renamed, the producer side has 0 faulty rows while on the consumer side 200 rows are dropped at screening, 200 more at consumption, and the summary goes from 55 cells to 45; once the contract is written the same corruption is caught at stage 0 but the output is not repaired; a strict contract that pins the field set exactly blocks 1200 good rows with no corruption present and drops the summary to 25 cells; batch-level enforcement blocks 2200 good rows and yields no output at all, while an adapter stage returns the serving to the baseline and grows the pipeline to six stages. The fastest-caught corruption of the previous lesson was `schema_changed`: when the field name changed, the `schema` check fired at stage 0. The field list that check looked at was buried inside the code, and who wrote the list was not asked. The list is the consumer's assumption; the producer knows nothing about it. This lesson turns that list into a **data contract**. Three things are measured: on whose side how many rows are dropped when a field name changes, at which stage the corruption is caught once the contract is written, and what the contract itself costs. The third number is the most important, because it is an example of the second claim of the course: **adding a check also breaks things.** - **QP8.** The source, the pipeline and the contract are **fictional**; no real database, file format or orchestrator is called. The seed is **20260218**, and the measurement does not depend on the run. - **QP9.** The relational schema, normalization, constraint and type procedures were measured in the Databases curriculum, in the Data Modeling and Relational Theory course, and **are not repeated.** The question here is not how a schema is designed but **who breaks when it changes**. - **QP10.** **A schema change is not an attack.** A producer renames a field to make it clearer or adds a new field; both are ordinary maintenance work. - **QP11.** The contract runs **at stage 0**, before the source enters the pipeline, and is enforced at row level: a row that does not conform is blocked, a row that conforms passes. - **QP12.** The baseline is the **pipeline without a contract**: blocked rows **0**, corruption caught **0**. - **QP13.** Resolution: the step of the row share is **0.0004**, of a summary cell **0.0182**. **No share is printed on the serving output**, the value itself is written. #### A Contract Is Not a List of Fields A list of fields only says "let these names exist". A contract has to declare three things at once: which fields **exist**, what the **type** of each is, and what will happen when an undeclared field arrives. The third is mostly not written, and the most expensive number of this lesson comes from there. The difference between the two implementations is a single line. A **permissive** contract requires the declared fields to exist and their types to match; it leaves an extra field free. A **strict** contract requires the field set to be **exactly equal** to the declared one; an extra field is also a violation. Both read the same contract text and behave the same on a corrupted source; where they part ways is a **harmless** change. ```python # FICTIONAL. Meter readings, the five-stage pipeline and the contract are defined in the lesson; # no real store, file format 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 # ---- PIPELINE: five stages, each 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)] def pipeline(records): trace, data = [], records for name, f in STAGES: data, dropped = f(data) trace.append({"stage": name, "remaining": len(data), "dropped": dropped}) return data, trace # ---- Of the six corruption classes of the common definition this lesson runs only one. def corrupt_schema(records, period="2026-07"): """The producer renamed the field ; the value itself stayed the same.""" out, counter = [], 0 for k in records: if k["period"] != period: out.append(k) continue y = dict(k) y["index_value"] = y.pop("index") counter += 1 out.append(y) return out, counter def add_field(records, first="2026-07"): """The producer added a harmless field ; the existing fields did not change.""" return [dict(k, meter_serial="S" + k["subscriber"][1:]) if k["period"] >= first else k for k in records] # ---- CONTRACT: the exact schema of the row the producer declares it writes CONTRACT = {"subscriber": str, "period": str, "region": str, "type": str, "index": int, "unit": str} def permissive(k): """Declared fields exist and their types are right ; an extra field is free.""" return all(a in k and isinstance(k[a], t) for a, t in CONTRACT.items()) def strict(k): """The field set must be exactly equal to the declared one ; an added field is also a violation.""" return set(k) == set(CONTRACT) and permissive(k) def ingest_with_contract(records, check=None): """The contract runs at stage 0: a conforming row passes , a non-conforming one is blocked.""" if check is None: return records, 0 passed = [k for k in records if check(k)] return passed, len(records) - len(passed) ``` #### Whom a Schema Change Breaks The second block applies six setups to the same source. The `entered` column is the rows that entered the pipeline, `blocked` the rows the contract blocked at stage 0, `screen` and `consumption` the rows those two stages dropped, and `summary` the cell count of the fourth stage. The last column says whether the five served values stayed the same as the baseline. ```python SOURCE_RECORDS = generate_source() BASELINE, _ = pipeline(SOURCE_RECORDS) def measure(name, records, check): entered, blocked = ingest_with_contract(records, check) serving, trace = pipeline(entered) print(f"{name:<28}{len(entered):>9}{blocked:>9}{trace[0]['dropped']:>8}{trace[2]['dropped']:>13}" f"{trace[3]['remaining']:>9}{'same' if serving == BASELINE else 'changed':>9}") corrupted, n = corrupt_schema(SOURCE_RECORDS) extended = add_field(SOURCE_RECORDS) print("baseline serving:", BASELINE) print("renamed rows:", n, " rows carrying the added field:", sum(1 for k in extended if "meter_serial" in k)) print() print(f"{'setup':<28}{'entered':>9}{'blocked':>9}{'screen':>8}{'consumption':>13}" f"{'summary':>9}{'serving':>9}") measure("healthy , no contract", SOURCE_RECORDS, None) measure("renamed , no contract", corrupted, None) measure("renamed , permissive", corrupted, permissive) measure("field added , no contract", extended, None) measure("field added , permissive", extended, permissive) measure("field added , strict", extended, strict) print() PRODUCER = ("subscriber", "period", "region", "type", "index_value", "unit") print("rows violating the producer's own schema:", sum(1 for k in corrupted if k["period"] == "2026-07" and not all(a in k for a in PRODUCER))) print("fields declared by the contract:", len(CONTRACT), " fields read by the pipeline:", 5) ``` ``` baseline serving: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} renamed rows: 200 rows carrying the added field: 1200 setup entered blocked screen consumption summary serving healthy , no contract 2400 0 0 0 55 same renamed , no contract 2400 0 200 200 45 changed renamed , permissive 2200 200 0 200 45 changed field added , no contract 2400 0 0 0 55 same field added , permissive 2400 0 0 0 55 same field added , strict 1200 1200 0 0 25 changed rows violating the producer's own schema: 0 fields declared by the contract: 6 fields read by the pipeline: 5 ``` The second to last line is the answer to the question: **the number of rows violating the producer's own schema is 0.** The producer renamed the field, wrote all the fields completely, and the write finished without error. There is no measurable flaw on its side. The side that breaks is the consumer, and it can count this only in its own pipeline. The second row gives the situation without a contract. The corrupted payload enters the pipeline as it is: the screening stage drops **200 rows** because it cannot find the required field, and the consumption stage drops **200 more rows** because it cannot find the previous value of the period that follows. That is 400 rows in total, a sixth of the source. The summary goes from 55 cells down to **45** and all five served values change. There is no error record anywhere; the pipeline completed successfully. The third row gives the situation with a contract and must be read carefully. The contract blocks 200 rows at stage 0, **detection lag 0**. But the summary is again **45** cells, and the serving has again changed. The contract did not repair the output; it only gave notice. The dropped rows stopped being scattered across stages and were gathered in one place: 200 of the 400 rows now stand at the gate rather than at the screening stage, and the remaining 200 still drop at consumption, because the previous value of the next period really does not exist. **What the contract produces is not the right number but a signal that arrives at the right time.** #### The Cost of the Contract The last three rows carry the second claim of the course. Starting from the `2026-07` period, the producer adds a harmless field to the rows: the meter serial number. None of the existing fields changes, no type drifts, and the five fields the pipeline reads stay in place. The pipeline without a contract is not affected by this change at all: 2400 rows enter, 55 cells, the serving is the same. The **permissive** contract is not affected either, because it does not count an extra field as a violation: blocked rows **0**, serving the same. The **strict** contract, however, blocks **1200 rows**. The number of rows carrying the added field is exactly 1200 as well, which means the strict contract rejects every row that sees the new field. The summary falls from 55 cells to **25**, and all five served values change. This row is not an accident; it is a consequence of the definition of the contract. Corruption: **none**. Corruption caught: **0**. Falsely blocked valid rows: **1200**. Half of the source did not enter the pipeline even though it carried no quality flaw. The cost of a check is not only the effort of writing and maintaining it; **a wrongly defined contract can stop a pipeline without any corruption at all.** The last line shows the second face of the cost. The contract declares **six** fields, and the pipeline reads **five** of them; the `type` field is not used at any stage. As the contract widens, the producer becomes tied to it, and the pipeline stops even when the type of a field the consumer never reads changes. The permissive contract's advantage over the strict one is therefore not absolute: while the permissive contract leaves the extra field free, it blocks a change in the declared fields with the same severity. The decision in writing a contract is which fields to declare, and every declared field produces both a guarantee and a lock. #### Stopping and Adapting What to do when the contract is violated is a separate decision, and its cost is measured separately. All the rows up to here are **row-level** enforcement: a row that does not conform is blocked, a row that conforms passes. The second option is **batch-level** enforcement: if there is a single non-conforming row in the payload, the whole of it is rejected. The third option is **adapting** instead of blocking: the consumer adds a stage that translates the known old name into the name in the contract. ```python def batch_level(records, check): """If even a single row does not conform , the whole payload is rejected.""" return ([], len(records)) if any(not check(k) for k in records) else (records, 0) MAPPING = {"index_value": "index"} def adapt(records): """Extra stage: translates the known old name into the name in the contract.""" out = [] for k in records: y = dict(k) for old, name in MAPPING.items(): if old in y: y[name] = y.pop(old) out.append(y) return out def report(name, entered, blocked): serving, trace = pipeline(entered) status = "same" if serving == BASELINE else ("none" if not serving else "changed") print(f"{name:<26}{len(entered):>9}{blocked:>9}{trace[3]['remaining']:>9}{status:>9}") print(f"{'enforcement':<26}{'entered':>9}{'blocked':>9}{'summary':>9}{'serving':>9}") report("row-level , permissive", *ingest_with_contract(corrupted, permissive)) report("batch-level , permissive", *batch_level(corrupted, permissive)) report("adapter + permissive", *ingest_with_contract(adapt(corrupted), permissive)) print() print("good rows falsely blocked at batch level:", sum(1 for k in corrupted if permissive(k))) print("stages added by the adapter:", 1, " names in the mapping table:", len(MAPPING)) ``` ``` enforcement entered blocked summary serving row-level , permissive 2200 200 45 changed batch-level , permissive 0 2400 0 none adapter + permissive 2400 0 55 same good rows falsely blocked at batch level: 2200 stages added by the adapter: 1 names in the mapping table: 1 ``` The three rows show three separate costs. Row-level enforcement filters out the wrong rows and keeps the pipeline running; in return it produces **an incomplete output**, and that output does not carry on itself the fact that it is incomplete. Batch-level enforcement produces no output at all; there is no serving. This is sometimes the right decision, because publishing a wrong number is more expensive than publishing no number — but its cost is plain: **2200 good rows** were falsely blocked, and all eleven healthy periods were put on hold because of the flaw of a single period. The third row is the only repair path: when the adapter stage translates the old name into the new one, all 2400 rows pass, the summary returns to **55** cells and the serving becomes **exactly the same** as the baseline. Its cost is written in two items. The pipeline is now **six-staged**, not five; and the mapping table is a burden the consumer will carry forever. As this table grows, the contract text and the schema the pipeline actually accepts drift apart: the contract says `index`, the pipeline accepts both names. If the producer one day brings the old name back with a different meaning, the mapping silently reads the wrong field, and by the definition of this course that is precisely a silent corruption. #### Summary - A data contract declares three things at once: which fields exist, what their types are, and what will happen when an undeclared field arrives. When the third is not written, the cost comes from the strict reading. - When a field name changes, **the number of rows violating the producer's own schema is 0**; the side that breaks is the consumer, and it can count the loss of 400 rows only in its own pipeline. - Without a contract the corruption is silent: 200 rows drop at screening, 200 at consumption, the summary goes from 55 cells to 45, the five served values change and no error record is produced. - Once the contract is written, the same corruption is caught **at stage 0**, but the summary is again 45 cells; the contract does not repair the output, it produces a signal at the right time. - A strict contract blocks **1200 good rows** on a harmless field addition, drops the summary to 25 cells, and does so while **corruption caught is 0**; adding a check also breaks things. - What to do on a violation is a separate cost: row-level enforcement gives an incomplete output, batch-level enforcement blocks 2200 good rows and gives no output at all, and the adapter returns the serving to the baseline but grows the pipeline to six stages and leaves the consumer a permanent mapping table. #### Next Step The contract caught the corruption at stage 0, but catching is not repairing: the 200 blocked rows have to wait somewhere, the producer has to fix them, and that period has to be given to the pipeline **again**. In no line of this lesson did the pipeline run twice. The second run is the thing the pipeline does most often, and it carries its own number: what happens to the row count when the same period is reprocessed, which stage absorbs the surplus, and does the served value show it. The next lesson runs the pipeline in two forms, **idempotent** and not, counts the duplicated rows and measures the difference in the serving. ### Resilience to Repetition and Reprocessing Source: https://academia.sh/en/courses/introduction-to-data-engineering/resilience-to-repetition-and-reprocessing Measuring that processing the same load a second time duplicates rows and that the served number does not show it: in an appending store, backfilling one period takes the raw rows from 2400 to 2600 and a full rerun takes them to 4800, reaching 7200 in three runs, yet the served value stays at 1212.55 in all three runs because the deduplication stage absorbs the surplus; when the resent load carries a correction, the same stage throws the correction away and leaves the value at 794.36; and when deduplication looks at the whole row instead of the key, the same reruns carry the value to 2425.09 and 3637.64. The previous lesson showed that the contract catches the corruption at stage 0, but catching is not repairing: the 200 blocked rows wait somewhere, the producer fixes them, and that period is **resent** to the pipeline. In none of the measurements so far has the pipeline seen the same data twice. The second run is not an exception; it is the most frequent thing a pipeline does. A producer corrects a load, a stage fails and is retried, a past period is **backfilled**. This lesson counts what happens to the row count of a pipeline that processes the same load twice, which stage absorbs the surplus, and whether the served value shows it. The answer is the cleanest example of the third claim of the course. - **QP14.** The source, the pipeline and the store are **fictional**; no real store, file format or orchestrator is called. The seed is **20260218**, and the measurement does not depend on the run. - **QP15.** The equivalent of `idempotent` in this curriculum is **idempotent**. The concept itself was measured in the Shell Programming course, and the retry and queue procedure in the Caching, Queues and Asynchronous Processing course, and **is not repeated**. The only question here is **whether reprocessing duplicates the row**. - **QP16.** Reprocessing is not an accident: a corrected load, a retried stage and a backfilled period are ordinary jobs. - **QP17.** The **value of a resent row stays the same**, only the load stamp changes; the one exception is the load that carries a correction. This detail determines the whole of the last two measurements. - **QP18.** The baseline is the **single run**: raw rows 2400, duplicated rows **0**. - **QP19.** The units counted: reprocessed rows, duplicated rows, rows dropped by deduplication and the difference at serving. **No raw time is written**, the unit is written as **run**. - **QP20.** Resolution: the step of the row share is **0.0004**, of a summary cell **0.0182**, of the serving value **0.2000**. **No share is printed on the serving output.** #### Two Write Modes Whether a pipeline is idempotent is decided by a single choice: **how** the incoming row is written to the store. An **appending** write puts the row at the end of the store; when the same row arrives twice, it sits in the store twice. A **keyed** write overwrites the row at a key; when the same key arrives twice, it sits in the store once. The key here is the pair of subscriber and period, which is already what the pipeline uses as its uniqueness criterion. The second part is how reprocessing looks. When a corrected load arrives, the **value of the row is the same**; the only thing that changes is the load stamp. This detail is often overlooked, and the whole of the last measurement rests on it. ```python # FICTIONAL. Meter readings, the five-stage pipeline and the store are defined in the lesson; # no real store, file format 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 # ---- PIPELINE: five stages, each 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)] def pipeline(records, stages=None): trace, data = [], records for name, f in (stages or STAGES): data, dropped = f(data) trace.append({"stage": name, "remaining": len(data), "dropped": dropped}) return data, trace # ---- INGESTION and REPROCESSING def ingest(records, load): """Ingestion stage: each load is given a stamp.""" return [dict(k, load=load) for k in records] def resend(records, periods, load): """The corrected load is resent ; the stamp changes, the value does not.""" return [dict(k, load=load) for k in records if k["period"] in periods] def store_append(store, new): """NON-idempotent write: the incoming row is appended to the end of the store.""" return store + new def store_keyed(store, new): """IDEMPOTENT write: the (subscriber , period) key is overwritten.""" table = {(k["subscriber"], k["period"]): k for k in store} for k in new: table[(k["subscriber"], k["period"])] = k return list(table.values()) def s2_dedupe_full(records): """FLAWED DEDUPLICATION: looks at the whole row, not at the key.""" seen, keep = set(), [] for k in records: stamp = tuple(sorted((a, str(v)) for a, v in k.items())) if stamp in seen: continue seen.add(stamp) keep.append(k) return keep, len(records) - len(keep) FLAWED = [("screen", s1_screen), ("dedupe", s2_dedupe_full), ("consumption", s3_consumption), ("summary", s4_summary), ("serve", s5_serve)] def c_uniqueness(data): return len(data) != len(set((k["subscriber"], k["period"]) for k in data)) def c_row_count(data, expected=2400, tolerance=0.05): return abs(len(data) - expected) / expected > tolerance ``` #### Duplicated Rows and an Unchanged Serving The second block applies five setups to the same source. The `raw` column gives the row count in the store, `dedupe` the rows dropped by that stage, `summary` the cell count of the fourth stage, and `north` one of the five served values. ```python SOURCE_RECORDS = generate_source() FIRST = ingest(SOURCE_RECORDS, 1) BASELINE, _ = pipeline(FIRST) def report(name, store, stages=None): serving, trace = pipeline(store, stages) print(f"{name:<30}{len(store):>6}{trace[1]['dropped']:>9}{trace[3]['remaining']:>9}" f"{serving['north']:>11.2f}{'same' if serving == BASELINE else 'changed':>9}") one_period = resend(SOURCE_RECORDS, {"2026-07"}, 2) full = resend(SOURCE_RECORDS, set(PERIODS), 2) print("baseline serving:", BASELINE) print("resent rows: one period", len(one_period), " full run", len(full)) print() print(f"{'setup':<30}{'raw':>6}{'dedupe':>9}{'summary':>9}{'north':>11}{'serving':>9}") report("single run", FIRST) report("appending store , one period", store_append(FIRST, one_period)) report("appending store , full run", store_append(FIRST, full)) report("keyed store , full run", store_keyed(FIRST, full)) report("flawed deduplication , full", store_append(FIRST, full), FLAWED) print() d = store_append(FIRST, full) print("uniqueness check:", c_uniqueness(d), " row count check:", c_row_count(d)) ``` ``` baseline serving: {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55} resent rows: one period 200 full run 2400 setup raw dedupe summary north serving single run 2400 0 55 1212.55 same appending store , one period 2600 200 55 1212.55 same appending store , full run 4800 2400 55 1212.55 same keyed store , full run 2400 0 55 1212.55 same flawed deduplication , full 4800 0 55 2425.09 changed uniqueness check: True row count check: True ``` The second row is the headline of the lesson: when one period is backfilled, the appending store carries **2600** rows, which means **200 rows were duplicated**. The deduplication stage drops exactly those 200 rows, and the served value stays at **1212.55**. The third row shows the same thing at a larger scale: a full rerun takes the store to **4800** rows, deduplication drops **2400** rows, and the serving is again the same. This is the cleanest example of the third claim of the course. The size of the store has doubled, the pipeline has done the same work twice, yet **none of the five served values has moved**. An observation that looks at the output concludes that the pipeline is idempotent. But what is idempotent is not the pipeline, it is **a single stage inside the pipeline**: deduplication absorbs the surplus and tells nobody while doing so. The fourth row shows what real idempotence is. After the second run the keyed store still carries **2400** rows: duplicated rows **0**, rows dropped by deduplication **0**. The difference is not in the store but beyond the pipeline; while producing the right number, the appending store takes up **twice the space** and cannot be used without a deduplication stage. The last two rows are what the checks say. The uniqueness check fires on the appending store, and the row count check fires too — 4800 is far outside the five percent band around the expected 2400. So the detection lag of the corruption is **0**, and the check that catches it is **not a check that looks at the serving.** #### The Absorbed Correction So far the value of the resent row was the same as the first one's, so which one deduplication kept did not matter. In real backfills the situation is the reverse: the load is resent **because it was corrected**. The deduplication stage keeps the row it sees first, that is, **the old one**. In the setup below, the producer read the last period short, then corrected it and resent it. The corruption is deliberately placed in the last period: a shift in the middle of the series returns with the opposite sign in the following period and the two cancel out in the average. ```python def short_reading(records, period="2026-12", diff=100): """The producer read the last period short ; it will then correct it and resend.""" return [dict(k, index=k["index"] - diff) if k["period"] == period else k for k in records] SHORT = ingest(short_reading(SOURCE_RECORDS), 1) CORRECTION = resend(SOURCE_RECORDS, {"2026-12"}, 2) print("corrected rows:", len(CORRECTION)) print(f"{'setup':<32}{'raw':>6}{'dedupe':>9}{'north':>10}{'baseline':>10}") for name, store in (("short load , no correction", SHORT), ("appending store , correction", store_append(SHORT, CORRECTION)), ("keyed store , correction", store_keyed(SHORT, CORRECTION))): serving, trace = pipeline(store) print(f"{name:<32}{len(store):>6}{trace[1]['dropped']:>9}{serving['north']:>10.2f}" f"{BASELINE['north']:>10.2f}") ``` ``` corrected rows: 200 setup raw dedupe north baseline short load , no correction 2400 0 794.36 1212.55 appending store , correction 2600 200 794.36 1212.55 keyed store , correction 2400 0 1212.55 1212.55 ``` The second row is the most expensive finding of the lesson. The correction arrived at the pipeline, 200 rows were reprocessed, the store went up to 2600 rows, deduplication dropped 200 rows — and the served value stayed at **794.36**, that is, **exactly the same** as before the correction. The 200 rows that were dropped were the corrected ones. The deduplication stage absorbed the second run; here, absorbing is **losing**. This is the exact opposite reading of the previous section, and both come from the same behavior. When the resent row is the same as the first, deduplication saves the pipeline; when it is corrected, the same deduplication throws the correction away. The stage cannot tell the two apart, because all it has is the key and the key of the two rows is the same. The third row shows the solution: because the keyed store overwrites the old row with the new one, the serving returns to **1212.55**, that is, to the baseline. For a pipeline to count as idempotent, it is not enough that processing the same load twice leaves the result unchanged; **processing a corrected load a second time must also correct the result.** The appending store meets the first condition and does not meet the second. #### When the Key Is Chosen Wrongly The last row of the table opens up the most fragile assumption of the pipeline. The deduplication stage looks at the `(subscriber, period)` key. Had the same stage been written to look at **the whole row**, the resent row would look different because of the load stamp and would not be eliminated. This is not a made-up flaw: asking whether a load is entirely the same seems intuitively safer, and that is exactly why it gets written. ```python appending, keyed = FIRST, FIRST print(f"{'run':<6}{'appending raw':>15}{'north':>9}{'flawed north':>15}" f"{'keyed raw':>18}{'north':>9}") for run in range(1, 4): if run > 1: new = resend(SOURCE_RECORDS, set(PERIODS), run) appending = store_append(appending, new) keyed = store_keyed(keyed, new) serve_append, _ = pipeline(appending) serve_flawed, _ = pipeline(appending, FLAWED) serve_keyed, _ = pipeline(keyed) print(f"{run:<6}{len(appending):>15}{serve_append['north']:>9.2f}" f"{serve_flawed['north']:>15.2f}{len(keyed):>18}" f"{serve_keyed['north']:>9.2f}") print() summ, _ = s4_summary(s3_consumption(s2_dedupe_full(s1_screen(appending)[0])[0])[0]) summ_base, _ = s4_summary(s3_consumption(s2_dedupe(s1_screen(FIRST)[0])[0])[0]) print("summary ('north', '2026-07') baseline:", summ_base[("north", "2026-07")], " flawed deduplication:", summ[("north", "2026-07")]) print("keys kept by deduplication:", len(set((k["subscriber"], k["period"]) for k in FIRST))) ``` ``` run appending raw north flawed north keyed raw north 1 2400 1212.55 1212.55 2400 1212.55 2 4800 1212.55 2425.09 2400 1212.55 3 7200 1212.55 3637.64 2400 1212.55 summary ('north', '2026-07') baseline: 1251 flawed deduplication: 3753 keys kept by deduplication: 2400 ``` The three-run table sets two sentences side by side. In the pipeline that works with the right key, the raw store goes from 2400 to 4800 to 7200 and the served value is **1212.55 in all three runs**. In the pipeline that works with the flawed key, the same raw store and the same reruns make the served value grow as **1212.55 → 2425.09 → 3637.64**: in direct proportion to the number of runs. The same shows in the summary layer: the `2026-07` cell of the north is 1251 at the baseline and **3753** after the third run. The only difference between the two pipelines is the key choice of one stage. No corruption class, no field name changed, no type drifted; the only thing that happened on the producer side is that the same correct load was sent a second time. This is the form that the second claim of the course takes in this lesson: **a pipeline's own stage, with a flawed key, can triple the served number without any source corruption.** The column of the keyed store says where the cost lies. Idempotence is not free: at write time a key table of **2400 entries** is kept and every row is looked up before it is written. The deduplication stage carries a set of the same size. In return, the store stays at 2400 rows regardless of the number of runs, and the correctness of the pipeline no longer depends on the key of a single stage. #### Summary - Reprocessing is an ordinary job; in an appending store, backfilling one period takes the raw rows from 2400 to 2600, a full rerun to 4800, and the third run to 7200. - None of the five served values shows this growth: the north is 1212.55 in all three runs, because the deduplication stage absorbs the surplus and does not say so while absorbing. - What is idempotent is not the pipeline but a single stage inside it; when keyed writing is used, duplicated rows fall to 0 and the store stays independent of the number of runs. - When the resent load carries a **correction**, the same deduplication throws it away: the served value stays at 794.36, whereas the keyed store returns to 1212.55 with the same correction. Idempotence is not that the second run leaves the result unchanged, but **that it corrects when it should correct**. - When deduplication looks at the whole row instead of `(subscriber, period)`, the same reruns carry the served value from 1212.55 to 2425.09 and then to 3637.64; the summary cell goes from 1251 to 3753. - The checks that catch duplication are uniqueness and row count, and both fire at stage 0; no check that looks at the serving can see it. #### Next Step In three lessons the check always looked at the same place: **the row itself**. Does it have its fields, is its type right, is its key unique, is its value in line with the previous periods. These questions give no answer when a serving value is corrupted, because none of them says **where the row came from**. Someone looking at the figure 3753 produced by the flawed deduplication has to read the whole pipeline from the beginning to find which loads contributed to that cell and at which link of the chain the error is. The next lesson counts this search: how many stages and how many rows one has to look at to go from a corrupted serving value back to the source, with a record kept and without one. ### Data Lineage Source: https://academia.sh/en/courses/introduction-to-data-engineering/data-lineage Counting the rows one has to look at to go from a corrupted served value back to the source, with and without a lineage record: a five-stage pipeline carries 9,460 rows across six layers while the lineage record holds 11,455 links, because every consumption row has two ancestors; without the record, explaining a served value means reading 9,260 rows and that number does not depend on which value one starts from, while the same search with the record ends between 94 and 155 rows. Two runs give two different loss stages: for late-arriving records the loss is at the source, for a schema change it is at the screening stage. The size of the serving deviation does not give the size of the corruption: when five times as many rows are corrupted, the serving moves less and three values go up. The previous three lessons made the check look at the row itself. The quality dimensions measured the value the row carries, the schema contract the fields of the row, reprocessing the same row passing a second time. In all three the object the check looked at was the row in hand. One question was never asked: where **did this row come from**. Five values stand at the serving; when one of them moves away from what was expected, which rows produce that value, and how many rows must be read to find them. What this lesson measures is the cost of a search. Going from a corrupted served value back to the source can be done in two ways: by reading every layer of the pipeline from the beginning, or by tracing backwards the **lineage** record that the pipeline wrote while it ran. Both find the same answer; the number of rows looked at differs by two orders of magnitude. The record itself also has a cost, and that is counted as well. - **QP41.** The source, the five-stage pipeline and the corruption classes are **fictional** and are defined in the lesson. No store, file format, catalog product or orchestrator is called. The seed is 20260218. - **QP42.** The **lineage record** is the set of **links** that each stage writes from its own output key to the input keys that produced that output. The row key is the subscriber and period, the summary key is the region and period, the serving key is the region. - **QP43.** The search is measured in two ways. **Without a record**, all six layers are read from the beginning to explain a served value. **With a record**, only the recorded ancestors are read at each layer. The unit counted is **the row looked at**; no time is written. - **QP44.** The deviation band is taken as 0.02 in the summary cell. The summary layer is 55 cells, so the smallest step of the cell shares is 0.0182; the band sits just above this step. **No share is printed on the serving output**, the value itself and its difference are written. - **QP45.** Observability, trace and logging procedure was measured in the Observability and Reliability course of the Backend Development curriculum and in the Observability and Operations course of the DevOps and Platform Engineering curriculum; **it is not repeated here**. The object of a trace is a request that moves through time; the object of lineage is **the rows a row depends on**. - **QP46.** The lineage record comes **from the run that happened**. It does not know what should have happened; the expected ancestor set comes from the record of the healthy run. #### The Lineage Record and How Many Links It Holds The five stages of the pipeline already count the rows they drop. Lineage asks for more than that: for each output row to write, by key, which input rows it was born from. The record is placed not inside the stage functions but between their inputs and outputs; the pipeline is the same pipeline. ```python # FICTIONAL. The meter readings of the measurement network, the five-stage pipeline and the corruption # classes are defined in the lesson; no store, file format 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 corrupt(records, cls, period="2026-07"): """Corruption classes. None of them is an attack, they are producer-side accidents.""" 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 this period counter += 1 continue out.append(k) elif cls == "schema_changed": y = dict(k) y["index_value"] = y.pop("index") # field name changed counter += 1 out.append(y) else: out.append(k) 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 LAYERS = ["source", "screen", "dedupe", "consumption", "summary", "serve"] def key_of(k): return (k["subscriber"], k["period"]) def lineage(records): """Runs the pipeline and writes a link from the output key to the input keys at each stage.""" v1, _ = s1_screen(records) v2, _ = s2_dedupe(v1) v3, _ = s3_consumption(v2) summ, _ = s4_summary(v3) serv, _ = s5_serve(summ) links = {"screen": {key_of(k): [key_of(k)] for k in v1}, "dedupe": {}, "consumption": {}, "summary": {}, "serve": {}} for k in v1: links["dedupe"].setdefault(key_of(k), []).append(key_of(k)) for k in v3: i = PERIODS.index(k["period"]) links["consumption"][key_of(k)] = [key_of(k), (k["subscriber"], PERIODS[i - 1])] links["summary"].setdefault((k["region"], k["period"]), []).append(key_of(k)) for b, d in summ: links["serve"].setdefault(b, []).append((b, d)) rows = {"source": len(records), "screen": len(v1), "dedupe": len(v2), "consumption": len(v3), "summary": len(summ), "serve": len(serv)} return serv, summ, links, rows source = generate_source() serving, summary, links, rows = lineage(source) print("layer ", " ".join(f"{a:>11}" for a in LAYERS)) print("rows ", " ".join(f"{rows[a]:>11}" for a in LAYERS)) print("links ", " ".join(f"{sum(len(v) for v in links[a].values()) if a in links else 0:>11}" for a in LAYERS)) total_links = sum(sum(len(v) for v in links[a].values()) for a in links) print(f"\nrows total {sum(rows.values())} , links total {total_links}") print("\nserving value value ancestor source rows") for b, v in serving.items(): ancestors = {x for c in links["serve"][b] for k in links["summary"][c] for t in links["consumption"][k] for a2 in links["dedupe"][t] for x in links["screen"][a2]} print(f"{b:<13} {v:>7} {len(ancestors):>20}") ``` ``` layer source screen dedupe consumption summary serve rows 2400 2400 2400 2200 55 5 links 0 2400 2400 4400 2200 55 rows total 9460 , links total 11455 serving value value ancestor source rows central 1321.45 588 north 1212.55 552 east 881.36 408 west 833.09 372 south 1065.55 480 ``` The record is **larger** than the data the pipeline carries. The six layers hold 9,460 rows in total, and the lineage holds 11,455 links; 1.21 links per row. The whole of the surplus comes from a single stage: `consumption` produces 2,200 rows but writes 4,400 links, because a consumption row is the **difference** of two readings and has two ancestors. How many links a stage writes is decided not by the row count but by how many inputs the output depends on. The last table is the most direct reading of the lineage: how many source rows each of the five served values stands on. Central 588, north 552, south 480, east 408, west 372; they total 2,400, that is, the whole of the source. Five numbers are visible at the serving layer; beneath them are two thousand four hundred rows, and which row went to which number is known only if it was written down. #### From the Corrupted Value to the Source: Two Runs The search works as follows. The recorded ancestors of the suspect served value are read; the summary cells that deviate from the corresponding cells of the healthy run, or that are not found at all, are selected; only their ancestors are read in a lower layer. The number of rows read at each layer is counted separately. Without a record there is no such narrowing; to explain a served value, all six layers are read. ```python # The definitions of the previous block continue: lineage, corrupt, serving, summary, links, rows. def trace_back(target, links_c, summary_c, source_c, tolerance=0.02): """From the served value to the source: at each layer only the recorded ancestors are read.""" looked, loss = {"serve": 1}, None candidates = links["serve"][target] looked["summary"] = len(candidates) suspect = [c for c in candidates if c not in links_c["summary"] or abs(summary_c[c] - summary[c]) / max(abs(summary[c]), 1) > tolerance] for upper, lower in (("summary", "consumption"), ("consumption", "dedupe"), ("dedupe", "screen")): candidates = sorted({x for c in suspect for x in links[upper][c]}) looked[lower] = len(candidates) missing = [x for x in candidates if x not in links_c[lower]] suspect, loss = (missing, lower) if missing else (candidates, loss) candidates = sorted({x for c in suspect for x in links["screen"][c]}) looked["source"] = len(candidates) absent = [x for x in candidates if x not in source_c] return looked, suspect, ("source" if absent else loss) for cls in ("late_arrival", "schema_changed"): corrupted, n = corrupt(source, cls) serving_c, summary_c, links_c, rows_c = lineage(corrupted) source_c = {key_of(k) for k in corrupted} no_record = sum(rows_c.values()) print(f"== {cls} , corrupted source rows {n} , rows looked at without a record {no_record}") print("region value diff serve summary consumption dedupe screen source total ratio origin" " loss stage") for b in REGION: look, susp, stage = trace_back(b, links_c, summary_c, source_c) total = sum(look.values()) print(f"{b:<9} {serving_c[b]:>8.2f} {serving_c[b] - serving[b]:>8.2f}" f" {look['serve']:>6} {look['summary']:>7} {look['consumption']:>11} {look['dedupe']:>7}" f" {look['screen']:>6} {look['source']:>6} {total:>7} {no_record / total:>5.0f}" f" {len(susp):>6} {stage}") print() ``` ``` == late_arrival , corrupted source rows 40 , rows looked at without a record 9260 region value diff serve summary consumption dedupe screen source total ratio origin loss stage east 837.18 -44.18 1 11 68 27 9 9 125 74 9 source west 811.18 -21.91 1 11 62 12 4 4 94 99 4 source north 1185.09 -27.46 1 11 92 21 7 7 139 67 7 source south 1011.27 -54.28 1 11 80 33 11 11 147 63 11 source central 1284.00 -37.45 1 11 98 27 9 9 155 60 9 source == schema_changed , corrupted source rows 200 , rows looked at without a record 8650 region value diff serve summary consumption dedupe screen source total ratio origin loss stage east 878.11 -3.25 1 11 68 102 34 34 250 35 34 screen west 845.89 12.80 1 11 62 93 31 31 229 38 31 screen north 1218.67 6.12 1 11 92 138 46 46 334 26 46 screen south 1072.56 7.01 1 11 80 120 40 40 292 30 40 screen central 1318.56 -2.89 1 11 98 147 49 49 355 24 49 screen ``` Without a record, the number of rows looked at is **9,260**, and this number does not depend on which of the five served values one starts from. The same holds in the second run: there too there is a single number, 8,650. The cost of the search without a record does not change with the target, because there is nothing to narrow; every row the pipeline holds in that run is read. The same search with lineage ends between 94 and 155 rows in the first run and between 229 and 355 in the second. For the south value the ratio is 63-fold, for the west 99-fold. How the number of links narrows can be seen in the columns. In the summary layer **11** cells are always read; there are twelve periods, but because consumption drops the first period, eleven summary cells per region remain. The rows read in the consumption layer are twice the number of subscribers of the region: central 98, east 68. These are the ancestors of the two deviating summary cells, that is, of `2026-07` and `2026-08`. The corruption corrupts a single period, but consumption is a difference, and the corrupted period enters the difference of **two** periods. Where the two runs really part ways is the last column. In the `late_arrival` run the loss stage is **source**: the suspect keys are absent from every layer of the pipeline and from the source record as well, which means those rows never arrived. In the `schema_changed` run the loss stage is **screen**: the same keys stand in the source, but have no link after the screening stage. The record tells these two situations apart. The search without a record reads the same 9,260 or 8,650 rows in both and in the end says only "missing". The size of the serving deviation does not give the size of the corruption. `schema_changed` corrupts five times as many rows (200 rows instead of 40) but moves the serving less: the largest difference is 12.80, where in the first run it was 54.28. Moreover, three values go **up**. The reason is that in the second run the corrupted rows drop out of the period average entirely, whereas in the first run they enter the average with an incomplete total. Reading the scale of the corruption from the serving gives a wrong result in both runs. #### What Lineage Does Not Say The record narrows a search field; it does not give an answer. The clearest proof is the source column of the `late_arrival` run. Nine, eleven and four keys are read there and none of them is found; but the list of those keys did not come out of the record of the corrupted run, because there **is no link at all** belonging to them there. The list came from the record of the healthy run. Lineage writes what happened; a row that did not happen becomes visible only by comparison with the expected ancestor set. Without the expected set the record shows only what exists, and the missing row is silent in the record as well. The second limit of the record is its cost. The 11,455 links are rewritten on every run; when the pipeline runs a second time for a reprocessed period, the record is written a second time too. The previous lesson measured that reprocessing duplicates the row; lineage carries this duplication as well, and unless it is written down which run a link belongs to, the links of two runs get mixed. The third limit goes beyond this lesson. When the loss stage comes out as `source`, the pipeline is clean within itself: no stage dropped a row, the row never arrived. Lineage says this correctly and stops there. It does not know why forty rows did not arrive, who should have sent them, or to whom the finding should be passed. The source is a place the pipeline **does not own**. #### Summary - The lineage record is the set of links that each stage writes from its output key to the input keys; while the pipeline carries 9,460 rows across six layers, the record holds 11,455 links, 1.21 per row. - The number of links is decided not by the row count but by how many inputs the output depends on: `consumption` produces 2,200 rows and writes 4,400 links, because a consumption row is the difference of two readings. - Without a record, explaining a served value means reading 9,260 rows, and this number does not depend on which value one starts from; the same search with the record ends between 94 and 155 rows. - Two runs give two different loss stages: source for `late_arrival`, screening for `schema_changed`. The search without a record cannot tell the two apart. - The size of the serving deviation does not give the size of the corruption: when five times as many rows are corrupted, the largest difference falls from 54.28 to 12.80 and three values go up. - Lineage cannot write a link for a row that never arrived; a missing row becomes visible only by comparison with the expected ancestor set of the healthy run. #### Next Step In this lesson the search went down to the source and stopped there. When the loss stage came out as `source`, the question answered was "which rows"; the question left unanswered was "who". The pipeline has five stages, six if the source is counted; how many of them carry an owner, to whom does the finding go for a corruption caught at a stage with no owner, and how many steps away from the stage it can fix does that person stand. The same question is asked backwards too: even if lineage has written links to the rows of a period, where does a link point if those rows were deleted because their retention period expired. The next lesson turns ownership, access and retention into three numbers and closes the course. ### Data Governance Source: https://academia.sh/en/courses/introduction-to-data-engineering/data-governance Counting the ownership gap and the retention policy: in a pipeline with six positions counting the source, three of the five stages carry an owner, yet all six corruptions are caught at an ownerless position, because the checks run only at the source and at the summary layer. A finding walks downstream until it reaches an owner, and three of the six findings go to an owner who can repair them; in the other three the recipient is one to five stages away from the repair point and has no access to the evidence. The three classes whose repair point is ownerless are exactly the three classes the flawed check set never catches. A six-period retention window deletes six periods but raises the periods that can no longer be audited retroactively to seven, the summary falls from 55 cells to 25, 2200 of the lineage's 4400 consumption links point to a deleted row, and the correct partition check raises a false alarm on intact data. The lesson is the course wrap-up. In the previous lesson the search went down to the source and stopped there. Lineage said which rows were missing, wrote that the loss stage was `source` and fell silent. Why forty rows never arrived and who should be told the record could not say. The pipeline has five stages, six counting the source; how many carry an owner. This lesson measures a **gap**: the distance between where a corruption is caught and where it can be repaired. The gap is read with three numbers: who the finding goes to, whether that role can make the repair, whether it can look at the evidence. The second measure reads the same gap on the time axis: how many periods can no longer be audited retroactively when the retention window narrows. All the numbers are counts; no ratio is printed. - **QP47.** The source, the five-stage pipeline, the corruption classes and the ownership map are **fictional** and are defined within the lesson. No store, catalog product, orchestrator or policy text is called. Seed 20260218. - **QP48.** An **owner** is the **role class** with the authority to change the output of a position; it is not a person or an institution. **Access** is which positions' rows a role can read. - **QP49.** A finding is **born at the position where it is caught**. If that position has no owner, the finding walks downstream to the first owned position; if none remains, it goes to no one. - **QP50.** The **repair point** is the **earliest position** at which a corruption can be fixed; it need not be the position where the finding was received. The **distance** is the number of stages between the two. - **QP51.** The detecting check and the detection lag were measured in the first lesson of this course; here they are taken as input and **not measured again**. Access control, the retention policy and the personal data procedure were measured in the Identity, Access and Cryptography and the Governance, Risk and Compliance courses of the Cybersecurity curriculum; they are **not repeated**. - **QP52.** The **retention window** keeps the last N periods and deletes what came before. A period is **retroactively auditable** if its consumption row can be recomputed from the remaining source. #### Ownership Map and the Path a Finding Walks Ownership is not a document but a mapping: who can change the output of which position. The map below recognizes two roles; the remaining positions are empty. The gap is not contrived but a result of the pipeline; intermediate layers are no one's product. ```python # FICTIONAL. The ownership, access and repair map is defined within the lesson; no # regulation text, institution, catalog product or orchestrator is called. POSITION = ["source", "screen", "dedupe", "consumption", "summary", "serve"] # An owner is not a person but a role class; the value of an ownerless position is None. OWNER = {"source": None, "screen": "pipeline team", "dedupe": "pipeline team", "consumption": None, "summary": None, "serve": "reporting team"} ACCESS = {"pipeline team": {"screen", "dedupe", "consumption", "summary"}, "reporting team": {"summary", "serve"}} # The detecting check and the detection lag (stage) were measured in the first lesson of this course. DETECTION = {"late_arrival": ("distribution", 4), "duplicated": ("uniqueness", 0), "schema_changed": ("schema", 0), "unit_changed": ("distribution", 4), "missing_partition": ("partition_period", 0), "type_drift": ("type", 0)} # The earliest position at which the corruption can really be repaired. REPAIR = {"late_arrival": "source", "duplicated": "dedupe", "schema_changed": "screen", "unit_changed": "source", "missing_partition": "source", "type_drift": "screen"} # The classes the flawed check set never catches, again from the first lesson. MISSED = {"late_arrival", "unit_changed", "missing_partition"} def route(i): """A finding is born at the position where it is caught; if the position is ownerless it walks to the first owned one.""" for j in range(i, len(POSITION)): if OWNER[POSITION[j]]: return j, OWNER[POSITION[j]] return None, None def yn(x): return "yes" if x else "no" print(f"{sum(1 for a in POSITION[1:] if OWNER[a])} of the five stages are owned ;" " the source position is outside the pipeline and ownerless") print("\ncorruption detector lag position owner finding goes to repair" " distance authority access in flawed set") authorized = accessed = 0 for cls, (det, lag) in DETECTION.items(): j, recipient = route(lag) o = REPAIR[cls] can_fix, can_see = OWNER[o] == recipient, o in ACCESS[recipient] authorized, accessed = authorized + can_fix, accessed + can_see print(f"{cls:<17} {det:>17} {lag:>8} {OWNER[POSITION[lag]] or 'none':>14} {recipient:<17}" f" {o:<7} {abs(POSITION.index(o) - j):>8} {yn(can_fix):>9} {yn(can_see):>6}" f" {'escapes' if cls in MISSED else 'caught':>14}") ownerless = {c for c in REPAIR if OWNER[REPAIR[c]] is None} print(f"\nthe owner receiving the finding can repair in {authorized} corruptions ," f" {accessed} corruptions have access to the evidence") print(f"{len(ownerless)} classes with an ownerless repair point , {len(MISSED)} classes the flawed set never catches" f" , same set: {yn(ownerless == MISSED)}") ``` ``` 3 of the five stages are owned ; the source position is outside the pipeline and ownerless corruption detector lag position owner finding goes to repair distance authority access in flawed set late_arrival distribution 4 none reporting team source 5 no no escapes duplicated uniqueness 0 none pipeline team dedupe 1 yes yes caught schema_changed schema 0 none pipeline team screen 0 yes yes caught unit_changed distribution 4 none reporting team source 5 no no escapes missing_partition partition_period 0 none pipeline team source 1 no no escapes type_drift type 0 none pipeline team screen 0 yes yes caught the owner receiving the finding can repair in 3 corruptions , 3 corruptions have access to the evidence 3 classes with an ownerless repair point , 3 classes the flawed set never catches , same set: yes ``` The harshest column is the fourth, and it holds a single value: **none**. Although three of the five stages are owned, not one of the six corruptions is caught at an owned position. The reason is where the checks sit: of the seven checks, six run at the source and one at the summary layer, and both positions are ownerless. Ownership was written where the pipeline **produces**, not where the check **looks**. Owning a position and running a check there are two separate decisions; that they do not overlap is a measurable flaw. Because the finding walks, it still finds a recipient, but the recipient is most often the wrong role. In three corruptions the pipeline team is both authorized and has access: `duplicated` is repaired in deduplication, `schema_changed` and `type_drift` in screening. In the other three the repair point is `source`, which has no owner. The `missing_partition` finding lands one stage away; `late_arrival` and `unit_changed` go **five stages** away, to the reporting team, which holds only a deviated served value. The last line is the narrowest finding of this lesson. The three classes whose repair point is ownerless are **exactly the same** as the three the flawed check set never catches. The two lists came out independently, one from the coverage of the checks, the other from the role map. That they match shows the least audited position is the least owned position. #### Retention Window and Retroactive Audit The ownership gap has a second face in time. Deleted rows free space and also take away the basis of the retroactive audit. The block below runs the same fictional source with five retention windows. ```python # FICTIONAL. The meter readings of the measurement network are generated within the lesson. In an # intact source the first two stages drop no rows, so consumption is computed directly from the source. 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 key_of(k): return (k["subscriber"], k["period"]) def s3_consumption(v): """Consumption from the period difference; a row cannot be computed if there is no previous period.""" table, out = {key_of(k): k for k in v}, [] for k in v: i = PERIODS.index(k["period"]) prev = table.get((k["subscriber"], PERIODS[i - 1])) if i else None if prev is not None: out.append(dict(k, m3=k["index"] - prev["index"])) return out def c_region_period(v): """Correct partition check: has every region been written in every period.""" return {(k["region"], k["period"]) for k in v} != {(b, d) for b in REGION for d in PERIODS} def retain(records, window): """Retention policy: only the last `window` periods are kept, the rest is deleted.""" kept = set(PERIODS[len(PERIODS) - window:]) return [k for k in records if k["period"] in kept], kept source = generate_source() # Every consumption row has two ancestors: its own period and the previous period. LINKS = [x for k in s3_consumption(source) for x in (key_of(k), (k["subscriber"], PERIODS[PERIODS.index(k["period"]) - 1]))] print(f"source {len(source)} rows , consumption rows {len(s3_consumption(source))} ," f" consumption links {len(LINKS)}") print("\n window deleted kept recomputable unauditable 2026-07 alarm dangling") for p in (12, 9, 6, 3, 1): remaining, kept = retain(source, p) recomputable = sum(1 for i, d in enumerate(PERIODS) if i and d in kept and PERIODS[i - 1] in kept) keys = {key_of(k) for k in remaining} status = ("deleted" if "2026-07" not in kept else "ancestor deleted" if "2026-06" not in kept else "auditable") print(f"{p:>7} {12 - p:>7} {len(remaining):>5} {recomputable:>20}" f" {12 - recomputable:>12} {status:>16}" f" {yn(c_region_period(remaining)):>5} {sum(1 for b in LINKS if b not in keys):>8}") ``` ``` source 2400 rows , consumption rows 2200 , consumption links 4400 window deleted kept recomputable unauditable 2026-07 alarm dangling 12 0 2400 11 1 auditable no 0 9 3 1800 8 4 auditable yes 1000 6 6 1200 5 7 ancestor deleted yes 2200 3 9 600 2 10 deleted yes 3400 1 11 200 0 12 deleted yes 4200 ``` The retention policy makes **one more** period unauditable than it deletes: with six periods deleted the unauditable periods are seven, with nine deleted ten. The extra period is always **the oldest of the window**; its rows remain, but consumption is a difference and the other end of that difference is deleted. In the six-period window the `2026-07` rows remain, yet they cannot be recomputed and the summary layer falls to 25 cells. Reading a retention period as "the last six periods are auditable" is a misreading; five periods are auditable. The last column answers the question the previous lesson left open. Lineage had written 4,400 consumption links; in the six-period window **2,200 of them** now point to a row that no longer exists, and 200 of those are precisely the ancestors of the boundary period. Keeping the record does not keep traceability; if a link's row is deleted, the record only reports where the loss is. The alarm column is the second cost. The period-aware partition check was built in this course as the **correct** check. When retention is applied, the set of periods it expects stays larger than the set that exists, and the check fires on intact data: no corruption, yet a finding. A correct check becomes a false alarm generator when the field it must look at is narrowed, because no one updated the expected period set with the policy. Whose job that is is the first section's question: the position where the policy is applied is `source`, and it has no owner. #### Summary - Ownership is the authority to change the output of a position; although three of the five stages are owned, all six corruptions are caught at an **ownerless** position, because the checks run at the source and at the summary layer while ownership was written to the produced layers. - A finding finds a recipient by walking downstream, but in three corruptions the recipient cannot make the repair; `late_arrival` and `unit_changed` go to a role **five stages** from the repair point, with no access to the evidence. - The three classes whose repair point is ownerless are exactly the three classes the flawed check set never catches: the least audited position is the least owned position. - The retention policy makes one more period unauditable than it deletes; with six periods deleted the unauditable periods become 7, the summary falls to 25 cells and the window's oldest period cannot be recomputed though its rows remain. - The record is not independent of retention: 2,200 of the 4,400 consumption links point to a deleted row, and the correct partition check turns into a false alarm. #### Course Wrap-Up | Lesson | Uninstrumented pipeline | Instrumented pipeline | Cost | |---|---|---|---| | `domain-and-lifecycle/01` — the corruption caught | 0 detections, lag **undefined** | **not measured** — the lesson only builds the measurement axis, it runs no check | trace 2400/2400/2200/55/5; 480 rows per value (588 / 372) | | `domain-and-lifecycle/02` — the stage that sees the corruption (6 classes) | `serve` holds 5 values in every class: **0** | `screen` 1, `dedupe` 1 (**+2**) | `consumption` doubles the missing rows: 40→80, 46→92, 200→400 | | `domain-and-lifecycle/03` — two measures: rows and deviation (m3) | rows: `schema_changed` 2200→1800, the noisiest | deviation: the same class 0.003, `late_arrival` **0.105** — the ranking reverses | `missing_partition` lowers the deviation 10.832 → 10.829 | | `domain-and-lifecycle/04` — the corruption class carried by the source class | tolerant external service 6 classes, lag 8 stages | declared type, primary key, consistent read 3 classes (**−3**), 4 stages (**−4**) | tolerant parsing corrupts 200 rows, no one catches it; the row count check with a 0.05 share catches none | | `domain-and-lifecycle/05` — the period in which the corruption is seen (mode) | batch: period lag 1, 5 stages, 13200 rows | stream: 0 (**−1**), 8 stages (**+3**), 2400 rows (**−10800**) | the stream carries 2600 inputs; on a late-arriving row batch leaves 55 cells correct, stream pollutes 10 of them | | `quality-and-principles/01` — detection lag (stage) | 0 detections, lag **undefined** | six of six classes are caught (**+6**), lag 4/0/0/4/0/0 | the flawed set never catches three classes; `partition` stays green with 46 rows missing | | `quality-and-principles/02` — the stage where the loss is caught | no contract: 200 + 200 rows silently dropped, summary 55→45 | with contract: stage **0**, summary still 45 — the output does not get fixed | the strict contract blocks 1200 good rows (summary 25); load-level enforcement blocks 2200, no output | | `quality-and-principles/03` — raw rows and served value | appending store 2400 → 4800 → 7200 | keyed write 2400 constant, duplicated rows **0** (**−4800**) | north is 1212.55 in all three runs; 2425.09 and 3637.64 with the flawed deduplication | | `quality-and-principles/04` — rows looked at in a search back to the source | no record: 9260 rows, independent of the target | with lineage 94–155 rows (**60–99 times fewer**) | 11455 links, 1.21 per row; a link cannot be written to a row that never arrived | | `quality-and-principles/05` — ownership gap and unauditable period | no ownership written: the finding has no recipient, unauditable period 1 | 3 of the 5 stages owned, 3 of the 6 findings go to an owner who can repair (**+3**) | **0** of the six detection positions are owned; a six-period retention raises the unauditable period to 7 (**+6**), dangling links 2200 | The table is one rule written ten times. **A data pipeline's number is not the rows it carries but how many steps later a wrong row is caught; a pipeline whose check for catching a corruption is not written counts as unmeasured.** No number in the uninstrumented column comes from bad code; they are the numbers of an arrangement that counts the carried rows but never asks whether they are correct. **The second claim is also read from the same table: an added check also corrupts.** The rows that say so by name are these. In `quality-and-principles/02` the strict contract blocks 1200 good rows on a harmless field addition and drops the summary to 25 cells, though the corruption caught is zero; in the same lesson load-level enforcement blocks 2200 good rows and yields no output. In `quality-and-principles/01` the `partition` check stays green while 46 rows are unwritten: false assurance costs more than no check. In `domain-and-lifecycle/04` the row count check with a 0.05 share catches no class. In `domain-and-lifecycle/05` stream mode permanently pollutes 10 summary cells on a late-arriving row, while batch mode leaves 55 correct. In `quality-and-principles/03` deduplication absorbs the corrected load. In `quality-and-principles/05` the retention policy turns the correct partition check into a false alarm. **The third claim is why looking at the served number is not enough**, and it is counted with three corruptions by name. `duplicated` leaves all five values exactly the same, because deduplication absorbs the surplus; `domain-and-lifecycle/02`, `domain-and-lifecycle/04` and `quality-and-principles/01` each measured this separately. `type_drift` does the same in a type conversion. `unit_changed` is the quietest: it moves ten summary cells between 1812946491 and −1812944121, the two periods cancel each other, and the value of north stays at 1212.55. None of the three can be seen by observing the served output. Across ten lessons the rows the pipeline carries were counted and how many stages later a corruption was caught was measured. One thing was never asked: **where** those rows sit. The source was cited as 2400 rows, the summary as 55 cells, the links as 11455 records; but the layout they are written in, its partitioning, and how much it **multiplies** a query were never counted. The next course turns this layout into an object of measurement.