Data Lake

The previous lesson defined the data warehouse by two decisions: the schema is enforced on write and the layout is chosen in advance. What the gate adds was measured, and the query finished with 400 cells. Its cost was measured as well: four of the six corruption classes pass the gate unaffected, and the 400 rejected rows never enter the store even though their measurement is recoverable.

This lesson measures the opposite decision. A data lake is the storage class that keeps incoming data as it is, enforcing no schema and performing no transformation. To say there is no schema would be wrong; the schema is deferred to read time, and the reading side carries it. The deferral has two sides, and both are countable: writing gets cheaper, reading gets more expensive. The measure of the lesson is the contrast of these two numbers; its second half is more uncomfortable, because it counts which corruption class is never seen at write time.

• SA10. The code is a fiction; no lake product, file format, catalog or query engine name is written. The source and the pipeline are taken over from the Introduction to Data Engineering course, seed 20260218.
• SA11. On the raw layer the field set is unknown. For this reason columns cannot be selected and partitions cannot be pruned while reading: all fields of every row are read, and a single file is touched.
• SA12. The warehouse side is the setup of the previous lesson: the pipeline output partitioned by the period key, the query selecting two columns. The per-file constant of 120 is a choice.
• SA13. Rows processed is a separate item: the number of rows a stage sees. The warehouse pays this cost once at write time, the lake pays it again at every read. Duration is not written as a raw figure.
• SA14. The six corruption classes are taken from the Introduction to Data Engineering course as input and the detection lag table is not repeated; the new question here is which scope is needed to see a corruption.
• SA15. Resolution: on 2200 rows the step of the row share is 0.0005. Cost is not a share; the value itself is written.

Writing and Reading on the Raw Layer

The write side of the lake is defined in one sentence: the incoming row is written as it is. The write side of the warehouse does two jobs: it passes the rows through the pipeline and places the output rows into the chosen layout. The code below gives the same source to both sides and asks both a single question, the region total of one period.

# FICTIONAL storage -- no real lake product, file 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


def s1_screen(records):
    """A row that lacks a required field is dropped."""
    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


COLUMNS = ("subscriber", "period", "region", "type", "index", "unit", "m3")
FILE_CONSTANT = 120        # fixed cost of touching one file , in cells


def write(records, partitioning=()):
    """Distributes the records into files by the partition key."""
    layout = {}
    for k in records:
        key = tuple(k[a] for a in partitioning)
        layout.setdefault(key, []).append(k)
    return layout


def query(layout, partitioning, selected, filter_=None, columnar=True):
    """Filter: {field: value}. Fields in the partition key prune files ,
    other fields make the whole file be read."""
    filter_ = {} if filter_ is None else filter_
    pruning = {a: v for a, v in filter_.items() if a in partitioning}
    remaining = {a: v for a, v in filter_.items() if a not in partitioning}
    files, cells, matched = 0, 0, 0
    columns_read = len(selected) if columnar else len(COLUMNS)
    for key, record in layout.items():
        if pruning:
            match = dict(zip(partitioning, key))
            if any(match[a] != v for a, v in pruning.items()):
                continue
        files += 1
        cells += len(record) * columns_read
        matched += sum(1 for k in record if all(k[a] == v for a, v in remaining.items()))
    return {"files": files, "cells": cells, "matching_rows": matched,
            "cost": files * FILE_CONSTANT + cells}


# ---- RAW LAYER: no schema , no partitioning , columns cannot be selected
def read_raw(records):
    """The field set is unknown, so all fields of every row are read."""
    cells = sum(len(k) for k in records)
    return {"files": 1, "cells": cells, "cost": 1 * FILE_CONSTANT + cells}


def run_pipeline(records):
    """Three stages; the rows each stage sees are counted as processed."""
    processed, v = 0, records
    for f in (s1_screen, s2_dedupe, s3_consumption):
        processed += len(v)
        v, _ = f(v)
    return v, processed


source = generate_source()
lake = read_raw(source)
warehouse_rows, transform = run_pipeline(source)
partitioned = write(warehouse_rows, ("period",))
wh = query(partitioned, ("period",), ("region", "m3"), {"period": "2026-07"})
ROWS = [("written rows", len(source), len(warehouse_rows)),
        ("written cells", lake["cells"], sum(len(k) for k in warehouse_rows)),
        ("rows processed at write time", 0, transform),
        ("query: files touched", lake["files"], wh["files"]),
        ("query: cells read", lake["cells"], wh["cells"]),
        ("query: cost", lake["cost"], wh["cost"]),
        ("rows processed at read time", transform, 0)]
print(f"{'measure':<28}{'data lake':>11}{'data warehouse':>16}")
for name, a, b in ROWS:
    print(f"{name:<28}{a:>11}{b:>16}")
print("lake gain on write", ROWS[1][2] - ROWS[1][1],
      " warehouse gain on one query", lake["cost"] - wh["cost"])

measure                       data lake  data warehouse
written rows                       2400            2200
written cells                     14400           15400
rows processed at write time          0            7200
query: files touched                  1               1
query: cells read                 14400             400
query: cost                       14520             520
rows processed at read time        7200               0
lake gain on write 1000  warehouse gain on one query 14000

The upper half of the table confirms the claim of the lake. The lake writes 2400 rows into 14400 cells and processes no row while writing; the warehouse writes 2200 rows into 15400 cells and for this processes a total of 7200 rows across the three stages of the pipeline. On the write side the lake’s gain is two items: 1000 cells and 7200 rows of transformation.

The lower half takes this gain back. The same question reads 14400 cells on the lake and 400 on the warehouse; the cost is 14520 against 520. The difference has three separate causes, and all three come from the absence of a schema. Because the field set is unknown, columns cannot be selected on the raw layer; because the layout does not say which row belongs to which period, partitions cannot be pruned; and because consumption has not been computed, the pipeline reruns at read time, so 7200 rows are processed again in every query.

The last row compares the two gains in the same unit: the 1000 cells the lake gains on write are paid back as 14000 cells in a single query. In this model the warehouse amortizes itself before the first query, and the gap grows linearly with the number of queries. The conclusion is not that the lake is unnecessary, but that the lake’s case is not in the cost table. The lake is chosen because it can accept data whose schema is not yet known and rejects no row. The count for these two properties is in the next table.

What the Write Time Cannot See

How much data must be looked at to see a corruption decides whether that corruption can be caught at write time. A gate at write time sees a single row: it accepts or rejects the row in hand. A check that looks at the whole set or at the aggregate cannot run at write time, because the set is not yet complete. The code below separates the six classes by these three scopes.

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


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


# ---- THREE SCOPES: row , set , aggregate
SCHEMA = {"subscriber": str, "period": str, "region": str, "type": str, "index": int, "unit": str}


def row_scope(k):
    """What can be seen at write time by looking at a single row: field and type."""
    return int(any(a not in k or not isinstance(k[a], t) for a, t in SCHEMA.items()))


def set_scope(data):
    """What cannot be seen without the whole set: uniqueness , row count , partition completeness."""
    key = [(k["subscriber"], k["period"]) for k in data]
    uniqueness = len(key) != len(set(key))
    count = abs(len(data) - 2400) / 2400 > 0.05
    seen = set((k["region"], k["period"]) for k in data)
    partition = seen != set((b, d) for b in REGION for d in PERIODS)
    return int(uniqueness or count or partition)


def aggregate_scope(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 1
    return 0


baseline_summary, _ = s4_summary(warehouse_rows)
print(f"{'class':<19}{'arrived':>7}{'row':>7}{'set':>6}{'aggregate':>11}  "
      f"{'scope':<11}{'lake':>7}{'warehouse':>11}")
for s in ["clean"] + CORRUPTIONS:
    incoming, _ = (source, 0) if s == "clean" else corrupt(source, s)
    ro = sum(row_scope(k) for k in incoming)
    se = set_scope(incoming)
    ag = aggregate_scope(s4_summary(run_pipeline(incoming)[0])[0], baseline_summary)
    scope = "row" if ro else ("set" if se else ("aggregate" if ag else "-"))
    print(f"{s:<19}{len(incoming):>7}{ro:>7}{se:>6}{ag:>11}  "
          f"{scope:<11}{len(incoming):>7}{len(incoming) - ro:>11}")


# ---- Two readers interpret the same lake layer differently
def repairing_reader(records):
    """A reader that carries a field-name mapping: the old name is met with the new one."""
    y = []
    for k in records:
        if "index" not in k and "index_value" in k:
            k = dict(k)
            k["index"] = k.pop("index_value")
        y.append(k)
    return y


corrupted, _ = corrupt(source, "schema_changed")
for name, data in [("unrepaired reader", corrupted), ("repaired reader", repairing_reader(corrupted))]:
    v, _ = run_pipeline(data)
    summary, _ = s4_summary(v)
    print(f"{name:<19}rows {len(v):>5}  aggregate cells {len(summary):>3}  total m3 {sum(summary.values()):>6}")

class              arrived    row   set  aggregate  scope         lake  warehouse
clean                 2400      0     0          0  -             2400       2400
late_arrival          2360      0     0          1  aggregate     2360       2360
duplicated            2460      0     1          0  set           2460       2460
schema_changed        2400    200     0          0  row           2400       2200
unit_changed          2400      0     0          1  aggregate     2400       2400
missing_partition     2354      0     1          0  set           2354       2354
type_drift            2400    200     0          0  row           2400       2200
unrepaired reader  rows  1800  aggregate cells  45  total m3  48004
repaired reader    rows  2200  aggregate cells  55  total m3  58454

The scope column carries the main result of the lesson. Of the six classes only two can be seen by looking at a single row; two need the whole set and two need the aggregate. So no row gate at write time can see four of them, and this is not an implementation shortcoming but a limit that comes from the scope itself: a row that never arrives passes no gate, a row that arrives a second time is known to be a repeat only if the first is remembered, and a region’s missing period is seen only after all periods have been written.

The unit_changed row is the quietest one: 0 in the row scope and 0 in the set scope as well. The value has grown a thousandfold, the unit label has stayed m3, and the row fits the schema flawlessly; uniqueness is intact, the partitions are complete and the row count is right. The only scope that sees this corruption is the aggregate, that is, the time after the pipeline has run and the totals have come out. In storage terms the conclusion is this: against this class there is no difference at all between the warehouse’s write gate and the lake’s lack of one; both write 2400 rows.

The lake and the warehouse part ways in the last two columns, and the difference is only two rows: schema_changed and type_drift stand as 2400 rows in the lake and 2200 in the warehouse. The detection difference the lake pays consists of these two classes alone; on the other hand, the 200 rows left in the lake are not lost. The last two rows show the consequence: two consumers reading the same lake layer produce different answers. The unrepaired reader does not recognize the field name, so it drops 200 rows, and then loses 200 more because the difference of the next period cannot be computed; it is left with 1800 rows, 45 aggregate cells and 48004 m3. The reader carrying a field-name mapping finds 2200 rows, 55 cells and 58454 m3. The gap of 10450 m3 between them is the difference not of the bytes on disk but of the interpretation at read time.

The course’s second claim holds here too: a layout also corrupts. The raw layout destroys no row, but the storage layer does not say what the correct answer is. Deferring schema enforcement to read time does not remove it, it only multiplies it: every consumer carries its own schema, and as many answers appear as there are consumers.

Summary

• A data lake stores incoming data as it is; the schema does not disappear, it is deferred to read time, and the reading side carries it.
• The lake writes 2400 rows into 14400 cells without any transformation; the warehouse processes 7200 rows to write 2200 rows into 15400 cells.
• The same period query gives 14400 cells and 14520 cost on the lake and 400 cells and 520 cost on the warehouse; the 1000 cells the lake gains on write are paid back as 14000 in a single query.
• Of the six corruption classes two are seen in the row scope, two in the set scope and two in the aggregate scope; a row gate at write time cannot see four of them, and unit_changed is invisible in both the row and the set scope.
• The write difference between the lake and the warehouse lies in only two classes; on the other hand, two consumers reading the same lake layer find 1800 and 2200 rows and 48004 and 58454 m3.

Next Step

The two lessons measured two extreme decisions. The warehouse enforces the schema at write time, gives the query a cheap read, and in return destroys recoverable rows. The lake destroys no row, and in return reruns the pipeline for every query and leaves the correct interpretation to the consumer. The desired sides of the two decisions seem not to exclude each other: the raw layer can be kept and schema enforcement can still be gathered in a single place, only not at write time but at read time. The next lesson measures this combination, that is, the lakehouse approach. Its question is tied to a single number: when enforcement is moved to read time, after how many stages is a corruption caught. The measure will be read in the unit of the Introduction to Data Engineering course, that is, as stages; so that the loss can be written next to the gain, the stored cells will also be counted separately.