Data Quality Dimensions

Over five lessons the pipeline was built and counted: how many rows each stage drops, how batch and stream processing change freshness. All of those numbers answered a single question, how many rows. At no stage was the question asked whether a row is correct.

These are two separate questions, and the gap between them is a flaw peculiar to a data pipeline: when the pipeline crashes it tells you, when it carries wrong data it says nothing. This lesson turns the quality dimensions into running checks, counts how many stages later each of the six corruption classes is caught, and shows in the same set that a check can itself be flawed.

• QP1. The source, the five-stage pipeline, the corruption classes and the checks are fictional; no real database, file format or orchestrator is called. The seed is 20260218, and the pipeline does not depend on the run.
• QP2. The corruption classes are not attacks, they are producer-side accidents: a file that arrives late, a payload sent twice, a field whose name changed, a meter whose unit changed.
• QP3. The baseline is the uninstrumented pipeline: corruption caught 0, lag undefined.
• QP4. Each corruption damages a single period (2026-07); the remaining eleven periods stay intact.
• QP5. The procedure for cleaning, missing values and outliers was measured in the Data Collection and Preparation course, and leakage and label noise in the Data Preparation and Feature Engineering course, and they are not repeated.
• QP6. Lag is written in stages, not in raw time: stage 0 is the source itself, and 1 through 5 are the stages of the pipeline.
• QP7. Resolution is the lower bound of the measurement band: the step of the row share is 0.0004, of a summary cell 0.0182, of a served value 0.2000. No share is printed on the serving output.

A Dimension Is a Check, Not a Sentence

A data quality dimension is usually written as a sentence: “the data must be complete”, “the data must be consistent”. These sentences cannot be measured, because they do not say on which set, at which stage and against what to look. A dimension is measured only once it is turned into a predicate: a function that runs after a specific stage and returns true or false.

Dimension	What the predicate asks	Check	Stage
completeness	did every expected row arrive	row_count, partition, partition_period	0
uniqueness	does the same key appear twice	uniqueness	0
validity	do the fields carry the expected name and type	schema, type	0
consistency	are the values in the pattern of earlier periods	distribution	4
accuracy	does the value reflect reality	none	—

The last row is in the table on purpose. Accuracy cannot be measured from inside the pipeline: to know how many cubic meters the meter really passed, a measurement outside the pipeline is needed, and this pipeline has no such source. What can be measured is consistency; a row in a quality report that says “accuracy ninety-nine percent” has most often measured consistency and called it accuracy. The stage column says a second thing: distribution looks not at the source but at the summary that the fourth stage produces.

Common Definition

The first block sets up the source, the five-stage pipeline, the six corruption classes and the seven checks. instrumented_pipeline runs the enabled checks after every stage; it returns the first detecting check and the detection lag.

# FICTIONAL. Meter readings of the measurement network, the five-stage pipeline, corruption
# classes and checks are defined in the lesson; no real store or orchestrator 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


def pick(u, w):
    t = 0.0
    for i, x in enumerate(w):
        t += x
        if u < t:
            return i
    return len(w) - 1


# ---- FICTIONAL source: meter readings of the measurement network, twelve periods
# region values stay Turkish: the region set is printed in sorted order
# dogu/bati/kuzey/guney/merkez = east/west/north/south/central
REGION = ["dogu", "bati", "kuzey", "guney", "merkez"]
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)}


# One reading per subscriber per period. Unit m3 , the meter index only increases.
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


# ---- FICTIONAL corruption classes. None of them is an attack; all are
# producer-side accidents that happen in everyday operation.
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"] == "kuzey":              # 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):
    """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


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)]


def pipeline(records):
    trace, data = [], records
    for name, f in STAGES:
        data, dropped = f(data)
        n = len(data) if hasattr(data, "__len__") else 0
        trace.append({"stage": name, "remaining": n, "dropped": dropped})
    return data, trace


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


# Check registry: (name , stage , function). Stage 0 = source , 1..5 = STAGES order.
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}

Detection Lag

The second block turns the rule into a number. For each corruption class the pipeline runs twice: once with all seven checks enabled, once with the set ["schema", "type", "uniqueness", "partition"]. The second set is flawed, because the period-aware completeness check and the distribution check are off. The last column says whether the five served values stayed exactly the same as the baseline.

SOURCE_RECORDS = generate_source()
BASELINE = build_baseline(SOURCE_RECORDS)
ALL_CHECKS = [a for a, _, _ in CHECKS]
FLAWED_SET = ["schema", "type", "uniqueness", "partition"]

_, trace = pipeline(SOURCE_RECORDS)
print("healthy pipeline:", " ".join(f"{i['stage']}={i['remaining']}" for i in trace))
print("serving:", BASELINE["baseline_serving"])
print("check fired on healthy pipeline:", instrumented_pipeline(SOURCE_RECORDS, BASELINE, ALL_CHECKS)[2])
print()
print(f"{'corruption':<18}{'rows':>6}{'detector':>17}{'lag':>9}"
      f"{'flawed set':>14}{'serving':>9}")
for cls in CORRUPTIONS:
    corrupted, n = corrupt(SOURCE_RECORDS, cls)
    serving, _, det, lag = instrumented_pipeline(corrupted, BASELINE, ALL_CHECKS)
    _, _, fdet, _ = instrumented_pipeline(corrupted, BASELINE, FLAWED_SET)
    status = "same" if serving == BASELINE["baseline_serving"] else "changed"
    print(f"{cls:<18}{n:>6}{det:>17}{lag:>9}{str(fdet):>14}{status:>9}")

healthy pipeline: screen=2400 dedupe=2400 consumption=2200 summary=55 serve=5
serving: {'merkez': 1321.45, 'kuzey': 1212.55, 'dogu': 881.36, 'bati': 833.09, 'guney': 1065.55}
check fired on healthy pipeline: None

corruption          rows         detector      lag    flawed set  serving
late_arrival          40     distribution        4          None  changed
duplicated            60       uniqueness        0    uniqueness     same
schema_changed       200           schema        0        schema  changed
unit_changed         200     distribution        4          None     same
missing_partition     46 partition_period        0          None  changed
type_drift           200             type        0          type     same

The first three lines give the baseline: on the healthy source the five stages go down from 2400 to 5 values and no check fires. This is the proof that the check set does not falsely block a valid row.

The lag column is the rule itself. Four classes are caught at stage 0, in the source itself; the pipeline is never contaminated. The other two are caught at stage 4: late_arrival and unit_changed become visible only through the distribution check that looks at the summary of the fourth stage. The four stages in between are a unit of cost, because the corrupted row was screened, deduplicated, turned into consumption and aggregated into the summary through those four stages; what will have to be backfilled after the fix is the output of those same four stages.

The last column carries the third claim of the course: three classes leave the served number exactly the same. After duplicated, unit_changed and type_drift the five served values are identical to the baseline. No observation that looks at the output can see these three; the only evidence is the check itself. type_drift is the most extreme example: when the index turns into text, the consumption stage converts it back to a number, and not a single row is dropped.

The fourth column measures the flawed set: three classes are never caught. late_arrival and unit_changed because the distribution check is off, and missing_partition because the partition check, which is on, is flawed. The team that built this set wrote four checks, all four stayed green, and three of the six classes went unmeasured.

A Check That Stays Green

The third block opens up the flaw. In the missing_partition corruption the 2026-07 period of the kuzey (north) region is never written. c_partition_completeness compares the region set with the expected set; because kuzey appears in the other eleven periods, the set is complete and the check passes. c_partition_completeness_period looks at the pair of region and period, and fires on the same source.

def interim_summary(records):
    """Output of the fourth stage: total per region and period."""
    v, _ = s1_screen(records)
    v, _ = s2_dedupe(v)
    v, _ = s3_consumption(v)
    return s4_summary(v)[0]


corrupted, n = corrupt(SOURCE_RECORDS, "missing_partition")
print("rows not written:", n, " kuzey rows of 2026-07:",
      sum(1 for k in corrupted if k["region"] == "kuzey" and k["period"] == "2026-07"))
print("region set in source:", sorted(set(k["region"] for k in corrupted)))
print("flawed partition check:", c_partition_completeness(corrupted),
      " period-aware check:", c_partition_completeness_period(corrupted))
summary_1 = interim_summary(corrupted)
print("summary cells:", len(summary_1), " periods of kuzey in summary:",
      sum(1 for b, d in summary_1 if b == "kuzey"))
print("serving kuzey:", s5_serve(summary_1)[0]["kuzey"], " baseline:", BASELINE["baseline_serving"]["kuzey"])

print()
summary_2 = interim_summary(corrupt(SOURCE_RECORDS, "unit_changed")[0])
for key in (("kuzey", "2026-06"), ("kuzey", "2026-07"), ("kuzey", "2026-08")):
    print("summary", key, "baseline:", BASELINE["baseline_summary"][key], " corrupted:", summary_2[key])
print("changed summary cells:", sum(1 for a in summary_2 if summary_2[a] != BASELINE["baseline_summary"][a]),
      "/", len(summary_2))
print("serving kuzey:", s5_serve(summary_2)[0]["kuzey"], " baseline:", BASELINE["baseline_serving"]["kuzey"])

rows not written: 46  kuzey rows of 2026-07: 0
region set in source: ['bati', 'dogu', 'guney', 'kuzey', 'merkez']
flawed partition check: False  period-aware check: True
summary cells: 53  periods of kuzey in summary: 9
serving kuzey: 1218.67  baseline: 1212.55

summary ('kuzey', '2026-06') baseline: 1041  corrupted: 1041
summary ('kuzey', '2026-07') baseline: 1251  corrupted: 1812946491
summary ('kuzey', '2026-08') baseline: 1119  corrupted: -1812944121
changed summary cells: 10 / 55
serving kuzey: 1212.55  baseline: 1212.55

The first group states the sentence: a check being green does not mean it measured. The region set has five elements, and the flawed check returns False. In the same source 46 rows never arrived, the summary has dropped from 55 cells to 53, and the served value of kuzey has shifted from 1212.55 to 1218.67 — given that the serving resolution is 0.2000, a measurable difference of more than thirty steps, but nobody is looking. The number of periods of kuzey in the summary has fallen to 9, because a period never being written drops the period that follows it as well: consumption works on the period difference, and the 2026-08 rows no longer have a previous period. The rows a stage drops are not the same number as the rows the corruption directly deleted.

The second group opens the quietest corruption. In unit_changed the meter reads liters instead of cubic meters but leaves the unit label as m3. In the summary, the kuzey total of 2026-07 rises from 1251 to 1812946491, and the total of 2026-08 falls from 1119 to −1812944121. The mechanism is cancellation: because the index grew a thousandfold in the corrupted period, the difference of that period is inflated, and because the index returns to its old scale in the following period, a negative difference of the same size appears. Ten cells of the summary move. When the mean over the region’s eleven periods in the summary is taken, the two cells cancel each other and the served value stays exactly at 1212.55; while the serving looks flawless, two periods are garbage.

Summary

• A quality dimension counts as measured only once it is turned into a running check that runs after a specific stage; accuracy cannot be measured in this pipeline, and what is measured instead is consistency.
• On the healthy source the five stages bring 2400 rows down to 5 served values and none of the seven checks fires; a valid row is not falsely blocked.
• The detection lag of the six corruption classes is 4, 0, 0, 4, 0 and 0 stages in order; four stages mean a pipeline whose intermediate tables are contaminated and have to be backfilled.
• With the set ["schema", "type", "uniqueness", "partition"] three classes are never caught: late_arrival, unit_changed and missing_partition. Because partition looks at the region set and does not see the period, it stays green on a source in which 46 rows were never written.
• Three classes leave the five served values exactly the same; although unit_changed moves ten cells of the summary between 1812946491 and −1812944121, the mean stays at 1212.55.

Next Step

The fastest-caught corruption of this lesson was schema_changed: when the field name changed, the schema check fired at stage 0. But the list of fields the check looks at was buried inside the code, and who wrote that list was never asked. When a producer changes the name of a field it receives no error on its own side; what breaks is the consumer. The next lesson turns this list into a data contract, counts how many rows are dropped when a field name changes and at which stage the contract catches the corruption, and then measures the cost of the contract itself.