Lifecycle

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.

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

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.