--- title: 'Data Lineage' source: 'https://academia.sh/en/courses/introduction-to-data-engineering/data-lineage' course: 'Introduction to Data Engineering' language: en updated: '2026-10-21T05:03:01+00:00' license: 'CC BY-SA 4.0' objectives: - 'Defines the lineage record as the links written from the output key to the input keys and computes the number of links the pipeline holds.' - 'Measures the search from a corrupted served value back to the source in two ways: by the number of rows looked at with and without the lineage record.' - 'Reads from the record at which stage the loss is, in two separate corruption runs.' - 'Writes the cost of the lineage record itself and its limit against a row that never arrived, as numbers.' --- # 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.