Data Lineage

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.

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

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