Lakehouse Approach

The previous two lessons measured two extreme decisions. The data warehouse enforces the schema on write, gives the query a read costing 520, and in return destroys 400 recoverable rows. The data lake destroys no row, makes the same query pay 14520 in return, and leaves the correct interpretation to the consumer; two consumers reading the same raw layer finding 48004 and 58454 m3 was the result of this.

The lakehouse separates these two decisions and combines them again. The raw layer is preserved as in the lake: incoming rows are written as they are, and nothing is rejected. Schema enforcement does not disappear, but it is taken from write time, moved to read time and gathered in a single place; the result is, as in the warehouse, a schema-conforming and partitioned curated layer. What this combination gains and loses is read on a single axis: detection lag. The measure is written in the unit of the Introduction to Data Engineering course, that is, in stages, so that the three setups can be compared with the same ruler.

• SA16. The code is a fiction; no lakehouse product, table format, catalog or query engine name is written. The source, the pipeline and the checks are taken over from the Introduction to Data Engineering course, seed 20260218.
• SA17. The lakehouse has two layers: the raw layer keeps the incoming row as it is, the curated layer keeps the output of the pipeline with enforcement applied. Enforcement runs while the curated layer is being built, that is, at read time.
• SA18. Detection lag is written in stages: 0 with no stage run, 1 to 5 the order of the five stages. Time is not written as a raw figure.
• SA19. The seven checks and the six corruption classes are taken as they are from the Introduction to Data Engineering course and their procedures are not repeated. The new question of this lesson is a single line: what happens to the lag when enforcement moves to read time.
• SA20. The warehouse’s write gate sees only the row scope, that is, schema and type; the checks in the set and aggregate scopes cannot run at write time. This distinction was measured in the previous lesson.
• SA21. Since the raw layer is preserved, repair is possible: a field-name mapping and a type conversion. There is no repair at the warehouse gate; a row that does not conform is rejected.
• SA22. Query cost is not measured again in this lesson, it is taken as input from the previous two lessons: 14520 on the raw layer, 520 on the curated layer partitioned by period. The per-file constant of 120 is a choice.

Moving Enforcement to Read Time

The three setups see the same corrupted source and differ only in which check runs where. In the warehouse the write gate runs only the two checks in the row scope. In the lake there is no check at all. In the lakehouse all seven checks run at read time, while the curated layer is being built, and each fires at its own stage.

# FICTIONAL pipeline and checks -- no real lakehouse product, table format or query
# engine is called. The source is FICTIONAL meter readings of a measurement network, 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 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):
    """Serving: 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)]


# ---- CHECKS: each one 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):
    """CORRECT CHECK: is every region present in every period."""
    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 instrumented_pipeline(records, baseline, enabled=None):
    """Runs the pipeline and applies the enabled checks after every stage.
    Returns: serving , trace , first detecting check and detection lag (stage)."""
    enabled = [a for a, _, _ in CHECKS] if enabled is None else enabled
    data, trace, detector, lag = records, [], None, None
    for step in range(len(STAGES) + 1):
        if step > 0:
            name, f = STAGES[step - 1]
            data, dropped = f(data)
            trace.append({"stage": name, "remaining": len(data), "dropped": dropped})
        for cname, cstage, cf in CHECKS:
            if cname not in enabled or cstage != step:
                continue
            if cf(data, baseline):
                if detector is None:
                    detector, lag = cname, step
    serving = data if isinstance(data, dict) and data and isinstance(
        list(data.values())[0], float) else None
    return serving, trace, detector, lag


def build_baseline(source):
    v1, _ = s1_screen(source)
    v2, _ = s2_dedupe(v1)
    v3, _ = s3_consumption(v2)
    summ, _ = s4_summary(v3)
    serve, _ = s5_serve(summ)
    return {"expected_rows": len(source), "baseline_summary": summ, "baseline_serving": serve}


source = generate_source()
baseline = build_baseline(source)
SETUPS = [("data warehouse", ["schema", "type"]),   # the gate looks at the row only
          ("data lake", []),                        # no enforcement at all
          ("lakehouse", None)]                      # all checks at read time


def cell_text(detector, lag):
    return f"{detector}/{lag}" if detector else "-"


lag_log = {name: [] for name, _ in SETUPS}
print(f"{'class':<19}{'data warehouse':>16}{'data lake':>11}{'lakehouse':>20}")
for s in CORRUPTIONS:
    incoming, _ = corrupt(source, s)
    cells = []
    for name, enabled in SETUPS:
        _, _, d, lag = instrumented_pipeline(incoming, baseline, enabled=enabled)
        if d:
            lag_log[name].append(lag)
        cells.append(cell_text(d, lag))
    print(f"{s:<19}{cells[0]:>16}{cells[1]:>11}{cells[2]:>20}")
for name, _ in SETUPS:
    v = lag_log[name]
    mean = f"{sum(v) / len(v):.2f}" if v else "undefined"
    print(f"{name:<19}caught {len(v)}/{len(CORRUPTIONS)}  mean lag {mean}")

class                data warehouse  data lake           lakehouse
late_arrival                      -          -      distribution/4
duplicated                        -          -        uniqueness/0
schema_changed             schema/0          -            schema/0
unit_changed                      -          -      distribution/4
missing_partition                 -          -  partition_period/0
type_drift                   type/0          -              type/0
data warehouse     caught 2/6  mean lag 0.00
data lake          caught 0/6  mean lag undefined
lakehouse          caught 6/6  mean lag 1.33

The last column is exactly the same as the first-detecting-check column of the Introduction to Data Engineering course; the lakehouse setup gives the same result as that course’s fully instrumented pipeline, because although where the enforcement stands changes, the set of checks does not. The comparison is therefore valid: the three columns are read on the same ruler, in the unit of stages.

The gain is on the coverage side. The write gate catches 2/6, the setup with no enforcement 0/6, the setup that enforces at read time 6/6. The four classes the gate fails to catch are not merely overlooked; the gate is not in a position to see them, because two need the set scope, two need the aggregate scope, and these scopes have not yet formed at write time.

The mean lag row must be read carefully. The warehouse’s 0.00 is smaller than the lakehouse’s 1.33, but this does not mean the warehouse catches faster. The two means are taken over different numbers of classes: the warehouse’s over only the two classes it catches, the lakehouse’s over all six. The two shared classes, schema_changed and type_drift, are caught at 0 stages in the lakehouse as well; moving enforcement to read time does not delay these two at all. The mean rising from 0.00 to 1.33 has one cause: the four values that are undefined in the warehouse becoming finite: 0, 0, 4 and 4. This is the storage-side counterpart of the distinction in the Introduction to Data Engineering course: an undefined lag is not better than zero, it is what was not measured. The mean lag of a setup cannot be read without writing how many classes it never catches.

What the Raw Layer Rescues and What It Costs

The lag table shows that the two setups give the same result against schema_changed: both catch it at zero stages. Where they part is what they do after catching. The warehouse rejects the row; the lakehouse keeps it on the raw layer and repairs it at read time.

SCHEMA = {"subscriber": str, "period": str, "region": str, "type": str, "index": int, "unit": str}
FIELD_MAP = {"index_value": "index"}


def violates(k):
    return any(a not in k or not isinstance(k[a], t) for a, t in SCHEMA.items())


def screen_at_gate(records):
    """Warehouse: a row that does not conform is never written."""
    accepted = [k for k in records if not violates(k)]
    return accepted, len(records) - len(accepted)


def repair_on_read(records):
    """Lakehouse: the raw layer is kept , enforcement is applied at read time with repair."""
    y = []
    for k in records:
        if violates(k):
            k = dict(k)
            for old, new in FIELD_MAP.items():
                if old in k and new not in k:
                    k[new] = k.pop(old)
            if isinstance(k.get("index"), str) and k["index"].isdigit():
                k["index"] = int(k["index"])
        y.append(k)
    return y


def run_pipeline(records):
    v = records
    for f in (s1_screen, s2_dedupe, s3_consumption):
        v, _ = f(v)
    return v


def count_cells(records):
    return sum(len(k) for k in records)


corrupted, _ = corrupt(source, "schema_changed")
accepted, rejected = screen_at_gate(corrupted)
repaired = repair_on_read(corrupted)
print("three setups under schema_changed")
print(f"{'setup':<14}{'rejected':>11}{'raw layer':>12}{'curated rows':>15}"
      f"{'stored cells':>16}")
for name, raw, curated_source in [("data lake", corrupted, corrupted),
                                  ("data warehouse", [], accepted),
                                  ("lakehouse", corrupted, repaired)]:
    d = run_pipeline(curated_source)
    h = count_cells(raw) + (count_cells(d) if name != "data lake" else 0)
    print(f"{name:<14}{rejected if name == 'data warehouse' else 0:>11}{len(raw):>12}{len(d):>15}{h:>16}")
clean = run_pipeline(source)
print("stored cells on clean source  lake", count_cells(source), " warehouse", count_cells(clean),
      " lakehouse", count_cells(source) + count_cells(clean))

three setups under schema_changed
setup            rejected   raw layer   curated rows    stored cells
data lake               0        2400           1800           14400
data warehouse        200           0           1800           12600
lakehouse               0        2400           2200           29800
stored cells on clean source  lake 14400  warehouse 15400  lakehouse 29800

The gain is in the third column: the lakehouse produces 2200 curated rows, the other two setups 1800. Half of the 400-row gap is the directly rejected records, the other half is their indirect consequence; when the period whose field name changed drops out, the consumption difference of the next period cannot be computed either. The warehouse’s loss is therefore larger than the number of rows it rejects, and the gate itself does not report this growth. In the lakehouse, since the raw layer stands, both periods come back the moment the mapping table is written.

The cost is in the last column: the lakehouse stores 29800 cells, the warehouse 12600, the lake 14400. On the clean source the picture is clearer: 14400, 15400 and 29800. The lakehouse stores the sum of the two, because it keeps the same data both raw and curated. On query cost it is at the warehouse’s level, that is 520; its raw layer can also be queried at 14520. So what the lakehouse buys is an option bought with stored cells: it can be regenerated when the curated layer is corrupted or the understanding of the schema changes.

The course’s second claim is written here too: a layout also corrupts. That the lakehouse rejects no row does not mean its curated layer is clean. The unit_changed row is the proof: it passes no gate, requires no repair, all 2200 rows are written to the curated layer, and the corruption is seen only at the fourth stage. Up to that point the curated layer is a table with a flawless-looking schema, carrying the full row count, with two periods grown a thousandfold. Conforming to the schema is not the same as being correct; enforcement moved to read time does not close this gap.

Summary

• The lakehouse preserves the raw layer and gathers schema enforcement in a single place by moving it from write time to read time; the result is a schema-conforming curated layer.
• Corruption classes caught: 2/6 at the write gate, 0/6 in the setup with no enforcement, 6/6 in the setup that enforces at read time.
• The mean detection lag is 0.00 stages in the warehouse and 1.33 in the lakehouse; since the two shared classes are caught at 0 stages in both, the rise of the mean is not a deterioration but four undefined values becoming finite.
• Under a field-name change the lakehouse produces 2200 curated rows, the other two setups 1800; half of the loss is the rejected rows, half is the next period they cause to drop.
• The cost is stored cells: on the clean source the lake is 14400, the warehouse 15400, the lakehouse 29800.
• Enforcement at read time does not prevent a bad row from being written to the curated layer; unit_changed passes no gate, is written as 2200 rows and is seen at the fourth stage.

Next Step

The three lessons separated what storage keeps. The warehouse keeps the schema-conforming row, the lake keeps the incoming row itself, the lakehouse keeps both, and what each choice costs the query, the detection and the stored cells was written one by one. There is a question that was never asked in the three lessons: for whom are these rows stored. It was assumed that all three setups served a single consumer, yet the parties reading from the same warehouse ask different questions and look at different columns. A party interested in one region has to filter a layout carrying all regions from scratch on every query. The next lesson asks this question and measures the subsets focused on a domain, that is, data marts; its measure is two numbers side by side: what the query cost becomes when the subset is written separately, and in how many places the same row stands.