Staged Layers

The previous lesson kept eleven models on a single plane: all equally durable, all equally queryable. On that plane, someone looking at the output of a model cannot tell whether the number came from the source or from after four transformations. This is why, in a durable transformation layer, models do not sit on a single plane; they are split into staging layers stacked on one another, and each layer makes a different promise.

This lesson builds three layers and measures two things: the rows each layer stores and the layer in which a corruption can be stopped. The second question depends on the first, because a rule can look only at the table it can read. The six corruption classes defined in the previous course are distributed over these three layers.

• TF17. The layers, the rules and the corruption classes are fictional; no real store, table format or data test library is called. The seed is 20260218, and the measurement does not depend on the run.
• TF18. Three layers: raw stores the source as it is, refined applies screening, deduplication and the consumption calculation, serving carries the summary and the published value. Each layer leaves its own table, and the next layer reads the previous one.
• TF19. A rule can run only if the table of the layer it reads is stored. When a layer is removed, the rules at its boundary go with it.
• TF20. The six corruption classes were defined in the Introduction to Data Engineering course and are not repeated; here they are used as input. None of them is an attack, all of them are accidents on the producer side, and each corrupts a single period.
• TF21. The routine that measures detection delay in stages was also measured in that course and is not repeated. The question here is not after how many stages but in which layer the corruption is stopped.
• TF22. The baseline is the single-layer pipeline: only the serving layer is stored, no intermediate table, no raw row.
• TF23. The units counted are the stored row, the summary cell and the stopped corruption class.
• TF24. Resolution: the step of the row ratio is 0.0004, that of the summary cell 0.0182. No ratio is printed over the serving output, the value itself is written.

Three Layers, Three Separate Promises

The layers are not copies of one another; each becomes able to answer a different question and loses something in return. The raw layer stores what comes from the source without changing it: field names, types and duplicated rows stay as they are. In this layer one can look at the name of a field or the type of a value, but nothing can be said about consumption, because consumption has not been calculated yet.

The refined layer applies screening, deduplication and the consumption calculation. Here there is now one consumption value per row, and one can look at that value’s sign, size and distribution. In return, the dropped rows are gone: someone looking at the refined layer cannot see what was dropped and why.

The serving layer carries only what will be published. Five values and the fifty-five summary cells that produce them. The only thing that can be done in this layer is to compare today’s number with yesterday’s.

# FICTIONAL. Meter readings, layers and rules are defined in the lesson;
# no real store, table format or data test library 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


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":
                counter += 1
                continue
            out.append(k)
        elif cls == "duplicated":
            out.append(k)
            if k["subscriber"] <= "A060":
                out.append(dict(k))
                counter += 1
        elif cls == "schema_changed":
            y = dict(k)
            y["index_value"] = y.pop("index")
            counter += 1
            out.append(y)
        elif cls == "unit_changed":
            y = dict(k)
            y["index"] = k["index"] * 1000
            y["unit"] = "m3"
            counter += 1
            out.append(y)
        elif cls == "missing_partition":
            if k["region"] == "north":
                counter += 1
                continue
            out.append(k)
        elif cls == "type_drift":
            y = dict(k)
            y["index"] = str(k["index"])
            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):
    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 LAYERS: each leaves its own table
LAYERS = ["raw", "refined", "serving"]


def build_layers(records):
    v = records
    for f in (s1_screen, s2_dedupe, s3_consumption):
        v, _ = f(v)
    summary, _ = s4_summary(v)
    serving, _ = s5_serve(summary)
    return {"raw": records, "refined": v, "summary": summary, "serving": serving}

The rules sit on the boundaries of the layers, and each reads only the table of its own layer. The field, type and key rules read the raw layer. The row count and the sign of consumption read the refined layer; the second cannot be written in the raw layer, because there is no consumption there. The distribution rule reads the summary.

def d_schema(v, fields=("subscriber", "period", "region", "index", "unit")):
    return any(any(a not in k for a in fields) for k in v)


def d_type(v):
    return any(not isinstance(k.get("index"), int) for k in v)


def d_uniqueness(v):
    key = [(k["subscriber"], k["period"]) for k in v]
    return len(key) != len(set(key))


def d_partition_period(v):
    seen = set((k["region"], k["period"]) for k in v)
    return seen != set((b, p) for b in REGION for p in PERIODS)


def d_negative(v):
    """Since the index only increases , consumption is never negative on a sound pipeline."""
    return any(k["m3"] < 0 for k in v)


def d_row_count(v, expected, tolerance=0.05):
    return abs(len(v) - expected) / expected > tolerance


def d_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


RULES = [
    ("schema", "raw", lambda lay, base: d_schema(lay["raw"])),
    ("type", "raw", lambda lay, base: d_type(lay["raw"])),
    ("uniqueness", "raw", lambda lay, base: d_uniqueness(lay["raw"])),
    ("partition_period", "raw", lambda lay, base: d_partition_period(lay["raw"])),
    ("row_count", "refined", lambda lay, base: d_row_count(lay["refined"], base["rows"])),
    ("negative", "refined", lambda lay, base: d_negative(lay["refined"])),
    ("distribution", "serving", lambda lay, base: d_distribution(lay["summary"], base["summary"])),
]


def stopper(lay, base, open_layers=LAYERS):
    """The first layer to stop the corruption and the rule that fires in that layer."""
    for layer in LAYERS:
        if layer not in open_layers:
            continue
        for name, k, f in RULES:
            if k == layer and f(lay, base):
                return layer, name
    return "-", "-"


SOURCE_RECORDS = generate_source()
T = build_layers(SOURCE_RECORDS)
BASELINE = {"rows": len(T["refined"]), "summary": T["summary"], "serving": T["serving"]}
print("sound pipeline  raw:", len(T["raw"]), " refined:", len(T["refined"]),
      " summary cells:", len(T["summary"]), " serving:", len(T["serving"]))
print("rule that fires on the sound pipeline:", stopper(T, BASELINE)[1])
print()

print(f"{'corruption':<19}{'raw':>6}{'refined':>11}{'summary':>9}{'serving':>9}"
      f"{'stopping layer':>16}{'rule':>17}")
for s in CORRUPTIONS:
    corrupted, _ = corrupt(SOURCE_RECORDS, s)
    lay = build_layers(corrupted)
    layer, rule = stopper(lay, BASELINE)
    print(f"{s:<19}{len(lay['raw']):>6}{len(lay['refined']):>11}{len(lay['summary']):>9}"
          f"{('same' if lay['serving'] == BASELINE['serving'] else 'changed'):>9}{layer:>16}{rule:>17}")

sound pipeline  raw: 2400  refined: 2200  summary cells: 55  serving: 5
rule that fires on the sound pipeline: -

corruption            raw    refined  summary  serving  stopping layer             rule
late_arrival         2360       2120       55  changed         serving     distribution
duplicated           2460       2200       55     same             raw       uniqueness
schema_changed       2400       1800       45  changed             raw           schema
unit_changed         2400       2200       55     same         refined         negative
missing_partition    2354       2108       53  changed             raw partition_period
type_drift           2400       2200       55     same             raw             type

In Which Layer the Corruption Stops

The sound pipeline leaves 2400 rows, 2200 rows and 55 summary cells in the three layers respectively; no rule fires. The distribution of the corrupted runs, on the other hand, is four, one and one.

The raw layer stops four classes, and all four for the same reason: duplicated, schema_changed, missing_partition and type_drift are visible in the row itself. A field is missing, a type has drifted, a key is duplicated, or a region-period pair is absent altogether. Because the raw layer stores the row as it is, it can look at these four. Had a transformation been applied, most of them would have become invisible: the evidence of the type_drift row is not in the refined layer, because the consumption calculation silently turns the number that became text back into a number and the row is not dropped.

The refined layer stops a single class: unit_changed. The rule that does this says that consumption cannot be negative; because the index only increases, it never fires on the sound pipeline. The rule could not have been written in the raw layer, because in the raw layer there is no such thing as consumption, only the index. This is the clearest proof of why the layer exists: the refined layer produces a new field, and a new rule can be written over that field.

The serving layer also stops a single class: late_arrival. Forty rows never arrived and the refined layer fell to 2120, but the drop did not exceed the five percent tolerance of the row count rule. Only after the corruption has passed both layers and entered the summary does it become visible, and only through the distribution comparison. By that point the corruption has already contaminated two tables, and the repair has to cover not only the serving layer but the two layers beneath it.

The last columns carry the quiet finding of the lesson. Of the six classes caught, three do not change the five served values at all: duplicated, unit_changed and type_drift. All three were caught, but none of them could have been noticed by looking at the output. unit_changed is the clearest example: 2400 rows in the raw layer, 2200 rows in the refined layer, 55 cells in the summary, and in serving the same five values as the baseline. Only the sign of consumption in the middle layer says that the corruption exists.

When a Layer Is Removed

Three layers are not free. The same data is stored in three different forms, and each layer brings a separate write job, a separate freshness question and a separate ownership. The last block removes the layers one by one and puts the number of stopped corruption classes next to the stored rows.

SETUPS = [("three layers", LAYERS), ("two layers", ["raw", "serving"]),
          ("one layer", ["serving"])]
print(f"{'setup':<12}{'rows stored':>16}{'stopped':>12}"
      f"{'at serving layer':>18}  not caught")
for name, open_layers in SETUPS:
    stored = sum(len(T[k]) if k != "serving" else len(T["summary"]) for k in open_layers)
    stopped, late, missed = 0, 0, []
    for s in CORRUPTIONS:
        corrupted, _ = corrupt(SOURCE_RECORDS, s)
        layer, _ = stopper(build_layers(corrupted), BASELINE, open_layers)
        if layer == "-":
            missed.append(s)
        else:
            stopped += 1
            late += layer == "serving"
    print(f"{name:<12}{stored:>16}{stopped:>12}{late:>18}  {missed}")

setup            rows stored     stopped  at serving layer  not caught
three layers            4655           6                 1  []
two layers              2455           6                 2  []
one layer                 55           2                 2  ['duplicated', 'schema_changed', 'missing_partition', 'type_drift']

The second row gives the real contribution of the refined layer. While the raw and serving layers are stored, all six of the six classes are stopped; the only difference is that unit_changed is now left to the distribution rule, that is, it is caught one layer late. That is what the refined layer buys, and its cost is 2200 rows. The value of a layer is not always catching a new corruption class; most of the time it is catching the same class higher up.

The third row is the baseline and carries the harshest number of the lesson. When only the serving layer is stored, total storage falls to 55 cells, but the stopped corruption classes fall from six to two. The four classes that are not caught are written by name in the list. Two of them (schema_changed, missing_partition) change the served value, so there is a chance that an attentive reader might notice one day. The other two (duplicated, type_drift) do not change the served value at all: in a single-layer pipeline there is nowhere to look for these two corruption classes, because the table that carries the evidence is not stored.

This is the layer-level counterpart of the rule of the course. The single-layer pipeline runs, produces numbers, and says nothing in four of the six corruptions. A pipeline that does not store the table that could show the corruption counts as unmeasured for that corruption. The cost of the layers, 4655 rows against 55, is exactly the price of this measurability.

Summary

• Three layers make three separate promises: the raw layer stores the row as it is, the refined layer produces the consumption value, the serving layer carries only what will be published. On the sound pipeline they leave 2400 rows, 2200 rows and 55 summary cells respectively.
• A rule can run only if the table of the layer it reads is stored; the rule that looks at the sign of consumption cannot be written in the raw layer, because there is no consumption there.
• Four of the six corruption classes stop in the raw layer, one in the refined layer and one only in the serving layer; when the last one is stopped, the two tables below it have long been contaminated.
• Of the six classes caught, three do not change the five served values at all; the only evidence of unit_changed is the negative consumption in the refined layer.
• When the refined layer is removed, the number of stopped classes does not fall, but unit_changed is caught one layer late; the 2200-row cost of the layer buys that one layer.
• When only the serving layer is stored, storage falls to 55 cells and the stopped classes fall from six to two; because two of the four classes that are not caught do not change the served value either, there is nowhere left to look.

Next Step

The rules of this lesson looked at the rows the pipeline carries: is there a field, is the type right, is the key unique, is consumption negative. All of them were about the data itself, and none said anything about the transformation itself. A transformation can take correct data and calculate wrongly; an aggregation can be grouped by the wrong key, a factor can be applied to the wrong type, a join can duplicated rows. None of these cases is visible in the input, and none of them trips the rules at the layer boundaries. The next lesson carries the rule from above the data to above the output of the transformation: an expectation is written on the output of each transformation, the number of corruption classes it catches is counted, and the residue that no expectation catches is measured.