Source Systems

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.

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

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.