Pipeline Observability

The previous lesson dealt with the failing task: a retry duplicates rows in a pipeline that is not idempotent, a half-written partition makes serving wrong in every region, and an atomic write brought the deviation down to between −2.89 and +12.80 by not writing the partition at all. In three lessons the pipeline was run and what happens when it fails was written down.

A failing task announces itself. The subject of this lesson is the run that does not fail: all tasks finish green, the graph completes and the pipeline carries wrong data. There the only evidence is the indicator placed on top of the pipeline. The lesson builds three of them — freshness, volume and schema change — and asks a single question: how many periods later do they show the six corruption classes of the Introduction to Data Engineering course, and which one do they never show.

• OO41. The source, the five-stage pipeline, the six corruption classes and the three indicators are fictional and are generated inside the lesson. No orchestrator, observability product or data store is called. The seed is 20260218.
• OO42. The observability procedure — metric collection, tracing, logging, alert design — was measured in the Observability and Reliability course of the Backend Development curriculum, the Observability and Operations course of the DevOps and Platform Engineering curriculum and the AI Agents course of the AI Engineering curriculum; it is not repeated. The only measure here is the lag of three indicators specific to the data pipeline.
• OO43. The six corruption classes are taken as input from the Introduction to Data Engineering course. Each of them corrupts only the 2026-07 period; the remaining eleven periods are healthy.
• OO44. Freshness means that the newest period in serving is at most one period behind the period being worked on; volume means that the row count of the period does not deviate from the median of the previous three periods by more than the band; schema evolution means that the field name and type pairs stay the same as the registered signature. The tolerance, the band and the signature are each a choice; the last section counts them.
• OO45. Drift is the change of the distribution of the input over time and was measured in the Model Evaluation, Interpretability and Ethics course of the Machine Learning and Data Science curriculum. The three indicators do not see drift; the first section counts the reason.
• OO46. The baseline is the run without orchestration: no indicator, classes seen 0, lag undefined. The lag is written in periods.

The Surface the Indicators Look At

The block below sets up the shared definition and prints a single thing: the run-independent measure of the three indicators, that is, the number of attributes each looks at. This number is known without running the pipeline at all.

# FICTIONAL. The meter readings of the measurement network, the five-stage pipeline, the six corruption
# classes and the three indicators are generated inside the lesson; no orchestrator, observability
# product or data store is called. Seed 20260218.
SEED, M32 = 20260218, 0xFFFFFFFF


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


REGION = ["east", "west", "north", "south", "central"]
TYPE = ["residential", "commercial"]
PERIODS = [f"2026-{a:02d}" for a in range(1, 13)]
SUBSCRIBERS = {}
for i in range(200):
    r = rng(SEED + 37 * i)
    SUBSCRIBERS[f"A{i + 1:03d}"] = {"region": REGION[int(r() * 5)], "type": TYPE[int(r() * 2)],
                                    "start": 10000 + int(r() * 60000)}


def generate_source():
    records = []
    for i, (sub, a) in enumerate(SUBSCRIBERS.items()):
        e = a["start"]
        for p in PERIODS:
            r = rng(SEED + 101 * i + 7 * PERIODS.index(p))
            e += 4 + int(r() * 46)
            records.append({"subscriber": sub, "period": p, "region": a["region"], "type": a["type"],
                            "index": e, "unit": "m3"})
    return records


CORRUPTIONS = ["late_arrival", "duplicated", "schema_changed", "unit_changed", "missing_partition",
               "type_drift"]


def corrupt(records, cls, period="2026-07"):
    """Corrupts a single period , the remaining periods stay healthy."""
    new, count = [], 0
    for k in records:
        if k["period"] != period:
            new.append(k)
            continue
        if cls == "late_arrival":
            if k["subscriber"] <= "A040":
                count += 1
                continue
            new.append(k)
        elif cls == "duplicated":
            new.append(k)
            if k["subscriber"] <= "A060":
                new.append(dict(k))
                count += 1
        elif cls == "schema_changed":
            y = dict(k)
            y["index_value"] = y.pop("index")
            count += 1
            new.append(y)
        elif cls == "unit_changed":
            y = dict(k)
            y["index"] = k["index"] * 1000
            y["unit"] = "m3"                        # LABEL UNCHANGED
            count += 1
            new.append(y)
        elif cls == "missing_partition":
            if k["region"] == "north":
                count += 1
                continue
            new.append(k)
        elif cls == "type_drift":
            y = dict(k)
            y["index"] = str(k["index"])
            count += 1
            new.append(y)
    return new, count


def s1_screen(records):
    REQUIRED = ("subscriber", "period", "region", "index", "unit")
    keep = [k for k in records if all(a in k for a in REQUIRED)]
    return keep, len(records) - len(keep)


def s2_dedupe(records):
    seen, keep = set(), []
    for k in records:
        key = (k["subscriber"], k["period"])
        if key not in seen:
            seen.add(key)
            keep.append(k)
    return keep, len(records) - len(keep)


def s3_consumption(records):
    table = {(k["subscriber"], k["period"]): k for k in records}
    out, dropped = [], 0
    for k in records:
        i = PERIODS.index(k["period"])
        if i == 0:
            continue
        prev = table.get((k["subscriber"], PERIODS[i - 1]))
        if prev is None:
            dropped += 1
            continue
        try:
            f = int(k["index"]) - int(prev["index"])
        except (TypeError, ValueError):
            dropped += 1
            continue
        out.append(dict(k, m3=f))
    return out, dropped


def s4_summary(records):
    totals = {}
    for k in records:
        key = (k["region"], k["period"])
        totals[key] = totals.get(key, 0) + k["m3"]
    return totals, 0


def s5_serve(summary):
    region = {}
    for (b, d), v in summary.items():
        region.setdefault(b, []).append(v)
    return {b: round(sum(v) / len(v), 2) for b, v in region.items()}, 0


# ---- THREE INDICATORS. Each looks at one projection of the record.
REGISTERED = {("subscriber", "str"), ("period", "str"), ("region", "str"), ("type", "str"),
              ("index", "int"), ("unit", "str")}
ATTRIBUTES = {"freshness": 1, "volume": 1, "schema": 2 * len(REGISTERED)}
print(f"source {200 * len(PERIODS)} rows x {len(REGISTERED)} fields ="
      f" {200 * len(PERIODS) * len(REGISTERED)} cells")
print("indicator  what it looks at              attributes")
print(f"freshness  newest period in serving      {ATTRIBUTES['freshness']:>10}")
print(f"volume     rows per period               {ATTRIBUTES['volume']:>10}")
print(f"schema     field name and type pairs     {ATTRIBUTES['schema']:>10}")
print(f"total observation points {sum(ATTRIBUTES.values())} ,"
      f" cells whose value is read 0")

source 2400 rows x 6 fields = 14400 cells
indicator  what it looks at              attributes
freshness  newest period in serving               1
volume     rows per period                        1
schema     field name and type pairs             12
total observation points 14 , cells whose value is read 0

The source carries 14400 cells and the three indicators look at 14 points. What matters is the last line: cells whose value is read: zero. Freshness asks whether a period exists, volume asks how many rows there are, schema asks the name and type of the fields. None of them looks at how many cubic meters a meter wrote; this is also the reason drift is not seen.

This line gives a prediction before the run: if among the six classes there is one that changes only the value, that class will be invisible to all three indicators. There is one class in the list that fits the definition. unit_changed multiplies the index by a thousand, leaves the unit label as m3, keeps the field name and type, and does not change the row count. That is the prediction; the run tests it.

How Many Periods Later

The block below runs the pipeline period by period over twelve periods and evaluates the three indicators after each period. For each class it prints the three lags, the first indicator that shows, how many periods the corruption lasts and whether serving changed.

# This block continues the generate_source , corrupt , s1..s5 and REGISTERED definitions of the first block.
SOURCE = generate_source()
INDICATORS = ("freshness", "volume", "schema")
CORRUPT_PERIOD = "2026-07"


def signature(records):
    return set((a, type(v).__name__) for k in records for a, v in k.items())


def median(x):
    y = sorted(x)
    return y[len(y) // 2]


def summarize(records):
    v, _ = s1_screen(records)
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    summary, _ = s4_summary(v)
    return summary


def watch(records, band=0.05, tolerance=1):
    """The pipeline is run each period and three indicators are evaluated. Freshness carries a
    tolerance of tolerance periods; volume looks at the deviation from the median of the last three periods."""
    fired, history = {g: [] for g in INDICATORS}, []
    for i, d in enumerate(PERIODS):
        current = [k for k in records if k.get("period") == d]
        summary = summarize([k for k in records if PERIODS.index(k["period"]) <= i])
        if i - max([PERIODS.index(x) for _, x in summary], default=-1) > tolerance:
            fired["freshness"].append(d)
        if len(history) >= 3:
            t = median(history[-3:])
            if abs(len(current) - t) / t > band:
                fired["volume"].append(d)
        history.append(len(current))
        if signature(current) != REGISTERED:
            fired["schema"].append(d)
    return fired


def lag(periods):
    """How many periods later it showed. A dash if it never showed."""
    after = [x for x in periods if x >= CORRUPT_PERIOD]
    return PERIODS.index(after[0]) - PERIODS.index(CORRUPT_PERIOD) if after else "-"


BASELINE = summarize(SOURCE)
BASELINE_SERVING, _ = s5_serve(BASELINE)
print(f"healthy pipeline: no indicator fires , serving {BASELINE_SERVING}")
print("\ncorruption         freshness  volume  schema   first indicator   affected periods  serving")
for c in CORRUPTIONS:
    v, _ = corrupt(SOURCE, c)
    a, summary = watch(v), summarize(v)
    serving, _ = s5_serve(summary)
    lags = {x: lag(a[x]) for x in INDICATORS}
    seen = [(v2, x) for x, v2 in lags.items() if v2 != "-"]
    impact = [d for (b, d) in set(BASELINE) | set(summary) if BASELINE.get((b, d)) != summary.get((b, d))]
    print(f"{c:<18}{lags['freshness']:>10}{lags['volume']:>8}{lags['schema']:>8}"
          f"   {(min(seen)[1] if seen else 'none'):<17}{len(set(impact)):>17}"
          f"  {'changed' if serving != BASELINE_SERVING else 'same'}")

print("\nband  largest hidden loss  classes seen by volume")
for band in (0.05, 0.10, 0.20, 0.25, 0.30):
    seen = [c for c in CORRUPTIONS if watch(corrupt(SOURCE, c)[0], band=band)["volume"]]
    print(f"{band:<5} {int(band * 200):<21}{len(seen)}  {seen}")

print("\ntolerance  false alarms on healthy pipeline  lag of schema_changed")
for tolerance in (0, 1, 2):
    false_alarms = watch(SOURCE, tolerance=tolerance)["freshness"]
    corrupted = watch(corrupt(SOURCE, "schema_changed")[0], tolerance=tolerance)["freshness"]
    print(f"{tolerance:<11}{len(false_alarms):<34}{lag(corrupted)}")

healthy pipeline: no indicator fires , serving {'central': 1321.45, 'north': 1212.55, 'east': 881.36, 'west': 833.09, 'south': 1065.55}

corruption         freshness  volume  schema   first indicator   affected periods  serving
late_arrival               -       0       -   volume                           2  changed
duplicated                 -       0       -   volume                           0  same
schema_changed             1       -       0   schema                           2  changed
unit_changed               -       -       -   none                             2  same
missing_partition          -       0       -   volume                           2  changed
type_drift                 -       -       0   schema                           0  same

band  largest hidden loss  classes seen by volume
0.05  10                   3  ['late_arrival', 'duplicated', 'missing_partition']
0.1   20                   3  ['late_arrival', 'duplicated', 'missing_partition']
0.2   40                   2  ['duplicated', 'missing_partition']
0.25  50                   1  ['duplicated']
0.3   60                   0  []

tolerance  false alarms on healthy pipeline  lag of schema_changed
0          1                                 0
1          0                                 1
2          0                                 -

On the healthy pipeline no indicator fires; false alarms are zero and serving is the serving of the shared definition.

The volume column shows three classes in the same period: late_arrival removes 40 rows, missing_partition 46, duplicated adds 60, and all three exceed the band of 0.05. Volume does not say what happened, it only says that the number has shifted; it does not show from which region the 46 rows dropped, because it carries no region breakdown. The period-aware completeness check of the Introduction to Data Engineering course says that.

schema_changed is shown by two indicators in two separate periods. Schema fires in the same period: the signature has index_value instead of index. Freshness fires one period later: the 200 rows whose field name changed drop at screening, the serving cell of that period is never formed, and because freshness carries a tolerance of one period it stays silent at 2026-07 and speaks at 2026-08. A single event, two indicators, a difference of one period.

type_drift is visible only to schema, and serving does not change: the index turned into text can be converted back to an integer in the consumption computation. The indicator fires, there is no damage. duplicated is the same: volume fires, the deduplication stage swallows the 60 copies, and the affected periods are 0.

The unit_changed row has three dashes and none. The prediction held. Serving does not change either; the Introduction to Data Engineering course measured the mechanism — in the intermediate layer two periods move by about 1.8 billion and cancel each other, and the average of the twelve periods stays the same. The affected periods column still writes 2: two summary cells are garbage. The only thing that sees this is the distribution check of the Introduction to Data Engineering course, and that is a data quality check, not a pipeline indicator. The difference is said in a single word: it looks at the value.

The hardest reading is across the columns. The indicator fires for at most one period, the corruption lasts two periods. The reason is the edge of the graph between periods: because consumption depends on the reading of the previous period, the 2026-08 rows of the 40 subscribers lost at 2026-07 cannot be computed either. One period after the alert goes silent, the pipeline is still corrupted. The opposite direction is even worse: in duplicated and type_drift the indicator fires and the damage is zero; in unit_changed the indicator is silent and two periods are garbage. There is no link between firing and damage.

Band and Tolerance Are Choices

The last two tables count the cost of the indicator itself. The number of classes volume sees is three at the bands 0.05 and 0.10, two at 0.20, one at 0.25 and zero at 0.30. The second column is the real meaning of the band: because a period carries 200 rows, the band is the largest loss it hides — 0.05 stays silent up to ten rows, 0.30 up to sixty rows. The resolution of the set is one row, that is 0.005; everything under the band is invisible, and this is not a flaw but the definition itself.

In this fictional source a narrow band is free, because every period carries exactly 200 rows and false alarms on the healthy pipeline are zero. In a real source the row count of a period varies, and the band is widened precisely to reduce the false alarm (false positive). As it widens, the three classes disappear in order of size: first the 40-row late_arrival, then the 46-row missing_partition, last the 60-row duplicated. The band is where false alarms and blindness are traded off, and the currency is the row.

The tolerance table shows the same tradeoff in freshness, in a barer form. At tolerance 0 the lag of schema_changed falls from 1 period to 0; the cost is one false alarm on the healthy pipeline, because the first period has no period before it and produces no consumption, and freshness without tolerance counts this as an empty serving. At tolerance 2 the false alarm stays zero but freshness sees nothing, the column is a dash. Three settings, three separate indicators.

The schema indicator has no such row. It cannot be tuned because it has no threshold: the signature either holds or does not. In return it sees only what is written in the signature, and it is blind to unit_changed because that does not break the signature. A tunable indicator chooses what it will miss; an untunable indicator fixes in advance, by its definition, what it will see.

Summary

• The three indicators look at a total of 14 attributes and none of them reads the value of a cell; this run-independent measure says in advance that the unit_changed class, which changes only the value, will stay invisible, and the run confirms it.
• Volume shows late_arrival, missing_partition and duplicated in the same period; schema shows schema_changed and type_drift in the same period; freshness shows schema_changed one period later, because it carries a tolerance of one period.
• The indicator fires for at most one period, the corruption lasts two periods: consumption depends on the reading of the previous period. In duplicated and type_drift the indicator fires while the damage is 0 periods, in unit_changed the indicator is silent while the damage is 2 periods.
• As the band widens from 0.05 to 0.30, the number of classes volume sees falls from 3 to 0 and the largest hidden loss rises from 10 rows to 60 rows; the band is the tradeoff between false alarms and blindness, in rows.
• When the freshness tolerance is lowered to 0, the lag falls from 1 period to 0 but one false alarm arises on the healthy pipeline; at tolerance 2 there is no false alarm and the indicator sees nothing.

Next Step

Every number in this lesson leaned on a single silent assumption: the code that runs the pipeline is always the same code, runs with the same setting and over the same dataset. The indicator band was 0.05, the tolerance was 1, the number of loaded periods was twelve — none of them was written in the code of the pipeline, they stood around the run. The next lesson runs the same pipeline code in two environments and measures how far serving deviates when the environment setting changes; then, when no version stamp is written on top of the run, it counts why it cannot be known with which code the report in your hands was produced.