Lesson 1 - Validation Rules

Welcome to Validation Rules#

Module 6 built CityFlow’s quarterly ETL and made it good: an explicit schema, three tested functions, a partitioned write, an idempotent re-run. Every number landed on Course 1’s ground truth to the cent. And that job, exactly as written, will happily process a February file containing 300,000 rows instead of 3,000,000 — an export that died halfway through the night — and publish a revenue figure that is off by ninety percent without printing a single warning. It will do this because nothing in the job has an opinion about what the data should look like. It reads what it is given.

This lesson gives the job opinions, and it makes them executable. A validation rule is not a comment in the code, not a line in a runbook, and not something you remember to eyeball on Monday morning. It is a small function that runs on every batch and returns a structured verdict: this rule, on this data, observed this, expected that, and therefore passed or failed at this severity. You will write four families of them — volume, range and domain, referential, and nullability and uniqueness — and run all eighteen as a single suite over the real 9,554,778-row quarter. Nine of them will pass. Nine of them will fire, on data that Module 6’s pipeline processes without complaint. That ratio is the point of the lesson, and it leads to the uncomfortable rule underneath all of them: a check that never fires is either describing perfect data or was written wrong, and the second is far more common than the first.

By the end of this lesson, you will be able to:

• Implement a validation rule as a small function returning a structured (name, passed, observed, expected, severity) result, rather than a bare assert or a printed hope
• Write the four rule families on real data — a volume tolerance band, range/domain bounds, a referential left_anti integrity check, and nullability/uniqueness counts
• Run rules as a suite and print a pass/fail table with observed-versus-expected, the artifact a reviewer actually reads
• Explain why a passing referential rule can still hide a data problem — 36,760 trips resolve against the zone dimension yet name no real zone
• Calibrate a threshold so a rule is informative rather than decorative, and recognise that each rule’s count is a Spark action with a real cost that caching amortises

You’ll need pyspark with the JDK configured as in Module 1, Lesson 3, the three taxi-quarter Parquet files, and taxi_zone_lookup.csv.

What a Rule Actually Is#

Start the one session this lesson uses:

import warnings
warnings.filterwarnings("ignore")

from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import (StructType, StructField, IntegerType, LongType,
                               DoubleType, StringType, TimestampNTZType)

spark = (SparkSession.builder
         .appName("cityflow")
         .master("local[*]")
         .config("spark.ui.showConsoleProgress", "false")
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

Now the shape. The instinct is to write validation as a bare assert count > 2_000_000 somewhere in the job, and that instinct fails for three reasons: an assert that passes tells you nothing, an assert that fails gives a stack trace instead of a number, and neither can be collected into a report. A rule needs to hand back what it saw alongside its verdict, so a run log can say “the volume rule passed with 2,964,624 rows against a band of 2,250,000 to 3,750,000” rather than merely “OK”. Five fields carry that:

from collections import namedtuple

RuleResult = namedtuple("RuleResult", "name passed observed expected severity")

def rule_row_volume(name, observed, expected, tolerance=0.25, severity="critical"):
    """VOLUME: is the batch size within a tolerance band around what we expect?"""
    low, high = expected * (1 - tolerance), expected * (1 + tolerance)
    return RuleResult(name, low <= observed <= high,
                      f"{observed:,} rows",
                      f"{low:,.0f}-{high:,.0f}", severity)

demo = rule_row_volume("volume_demo", observed=300_000, expected=3_000_000)
print(demo)
print("passed?", demo.passed)

RuleResult(name='volume_demo', passed=False, observed='300,000 rows', expected='2,250,000-3,750,000', severity='critical')
passed? False

Read the fields, because every rule in this lesson returns exactly these. name identifies the rule in a log so an alert can say which check failed. passed is the verdict a gate reads. observed and expected together are the diagnosis — without them a failure means “something is wrong” and someone has to re-run the query by hand at 3 a.m. to find out what. severity is the field that separates this from a blunt assert: some violations mean abort the run, others mean note it and keep going, and encoding that in the rule rather than at the call site is what Lesson 4 will build its warn-versus-abort gate on.

Note also what the rule function does not do: it does not touch Spark. rule_row_volume takes a number. That separation makes rules unit-testable on hand-made inputs exactly the way Module 6, Lesson 2 made clean_trips testable on a four-row frame — the measurement happens in Spark, the judgment happens in plain Python.

Family 1: Volume — A Band, Not A Number#

The first question about any batch is the cheapest and the most valuable: did roughly the right amount of data arrive? Almost every catastrophic feed failure is visible in the row count alone, long before anyone inspects a column. The trick is that the expected count is never exact — February is short, March is long, and a rule demanding precisely 3,000,000 rows would fail every single month. So the rule takes a tolerance band. Declare the schema from Module 6 and count each month:

TAXI_SCHEMA = StructType([
    StructField("VendorID", IntegerType(), True),
    StructField("tpep_pickup_datetime", TimestampNTZType(), True),
    StructField("tpep_dropoff_datetime", TimestampNTZType(), True),
    StructField("passenger_count", LongType(), True),
    StructField("trip_distance", DoubleType(), True),
    StructField("RatecodeID", LongType(), True),
    StructField("store_and_fwd_flag", StringType(), True),
    StructField("PULocationID", IntegerType(), True),
    StructField("DOLocationID", IntegerType(), True),
    StructField("payment_type", LongType(), True),
    StructField("fare_amount", DoubleType(), True),
    StructField("extra", DoubleType(), True),
    StructField("mta_tax", DoubleType(), True),
    StructField("tip_amount", DoubleType(), True),
    StructField("tolls_amount", DoubleType(), True),
    StructField("improvement_surcharge", DoubleType(), True),
    StructField("total_amount", DoubleType(), True),
    StructField("congestion_surcharge", DoubleType(), True),
    StructField("Airport_fee", DoubleType(), True),
])

MONTHS = {"2024-01": "yellow_tripdata_2024-01.parquet",
          "2024-02": "yellow_tripdata_2024-02.parquet",
          "2024-03": "yellow_tripdata_2024-03.parquet"}

volume_results = []
for label, path in MONTHS.items():
    n = spark.read.schema(TAXI_SCHEMA).parquet(path).count()
    volume_results.append(rule_row_volume(f"volume_{label}", n, expected=3_000_000))

for r in volume_results:
    print(f"{r.name:<16} {'PASS' if r.passed else 'FIRED':<6} {r.observed:>18}   expected {r.expected}")

volume_2024-01   PASS       2,964,624 rows   expected 2,250,000-3,750,000
volume_2024-02   PASS       3,007,526 rows   expected 2,250,000-3,750,000
volume_2024-03   PASS       3,582,628 rows   expected 2,250,000-3,750,000

Three passes, and the numbers are Course 1’s monthly counts exactly: 2,964,624, 3,007,526, 3,582,628. Notice how the band earns its width — March is 19.4% above the 3,000,000 centre, which a ±10% tolerance would have flagged as an anomaly every March forever. A rule that cries wolf on normal seasonal variation gets muted by the on-call engineer within two weeks, and a muted rule is worse than no rule because it creates the illusion of coverage.

Now prove the rule can actually fail, on real data rather than a hypothetical. Take a genuine 300,000-row slice of February — exactly what a truncated export looks like — and run the same function over it:

truncated = spark.read.schema(TAXI_SCHEMA).parquet(MONTHS["2024-02"]).limit(300_000)
bad = rule_row_volume("volume_2024-02_truncated", truncated.count(), expected=3_000_000)
print(f"{bad.name:<26} {'PASS' if bad.passed else 'FIRED':<6} {bad.observed:>14}   expected {bad.expected}")

volume_2024-02_truncated   FIRED    300,000 rows   expected 2,250,000-3,750,000

FIRED, in one line, before a single downstream number was computed. This is the failure mode that motivates the whole module: the truncated file is perfectly valid — correct schema, clean types, plausible values in every column, and every other rule in this lesson would pass on it. Only the count knows anything is wrong. Ten percent of a month passes every structural check you can write and produces a revenue total that is quietly, catastrophically wrong.

Caching the Frame Every Rule Reads#

Before the remaining families, a Spark-specific fact that shapes how suites are built. Every rule’s count() or agg() is an action, and an action on an uncached DataFrame re-reads the source from disk. Fifteen rules over one frame means fifteen full scans of 160 MB unless you intervene. The intervention is the one Module 5, Lesson 3 measured: materialise once, reuse many times. Project down to only the columns the rules examine first — caching nine columns instead of nineteen is roughly half the memory for identical results:

CHECK_COLS = ["tpep_pickup_datetime", "trip_distance", "total_amount", "PULocationID",
              "passenger_count", "RatecodeID", "store_and_fwd_flag",
              "congestion_surcharge", "Airport_fee"]

quarter = spark.read.schema(TAXI_SCHEMA).parquet(*MONTHS.values())
checks_frame = quarter.select(*CHECK_COLS).cache()
print(f"quarter rows under test : {checks_frame.count():,}")
print(f"columns kept for checks : {len(checks_frame.columns)} of {len(quarter.columns)}")

quarter rows under test : 9,554,778
columns kept for checks : 9 of 19

9,554,778 rows — the full quarter, Course 1’s raw total — now held in memory across 9 of the 19 columns. That first count() is the action that populates the cache; every rule after it reads from memory. This is not a micro-optimisation: a validation suite is unusual precisely because it hits the same data a dozen times in a row for a dozen different reasons, which is the exact access pattern caching exists for.

Family 2: Range and Domain — And The First Real Failure#

Range rules assert that individual values are physically possible: distances non-negative and under some sane ceiling, money inside plausible bounds, timestamps inside the reporting window. Each of those is a separate rule, but they need not be separate passes over the data. Compute every extreme in a single agg, then let the rule functions judge the results in Python:

probe = checks_frame.agg(
    F.min("trip_distance").alias("dist_min"),
    F.max("trip_distance").alias("dist_max"),
    F.sum((F.col("trip_distance") > 200).cast("int")).alias("dist_over_200"),
    F.min("total_amount").alias("amt_min"),
    F.max("total_amount").alias("amt_max"),
    F.min("tpep_pickup_datetime").alias("pickup_min"),
    F.max("tpep_pickup_datetime").alias("pickup_max"),
    F.sum(((F.col("tpep_pickup_datetime") <  "2024-01-01") |
           (F.col("tpep_pickup_datetime") >= "2024-04-01")).cast("int")).alias("pickup_outside"),
).first().asDict()

for k, v in probe.items():
    print(f"{k:<16}: {v}")

dist_min        : 0.0
dist_max        : 312722.3
dist_over_200   : 170
amt_min         : -1000.0
amt_max         : 9792.0
pickup_min      : 2002-12-31 22:17:10
pickup_max      : 2024-04-01 00:34:55
pickup_outside  : 21

One action, eight measurements — and a sum of a cast boolean is how you count violating rows inside an aggregate instead of running a separate filtered count() per condition, no UDF required. Now the rules that judge them:

def rule_range(name, observed, low, high, unit="", severity="critical"):
    """RANGE/DOMAIN: is an observed extreme inside the plausible band?"""
    return RuleResult(name, low <= observed <= high,
                      f"{observed:,.1f}{unit}", f"{low:,.0f} to {high:,.0f}{unit}", severity)

def rule_zero_count(name, observed, severity="critical"):
    """DOMAIN: a violating-row count that must be exactly zero."""
    return RuleResult(name, observed == 0, f"{observed:,} rows", "0 rows", severity)

range_results = [
    rule_range("distance_min",  probe["dist_min"], 0, 200, " mi"),
    rule_range("distance_max",  probe["dist_max"], 0, 200, " mi"),
    rule_range("amount_min",    probe["amt_min"], -1000, 10000, ""),
    rule_range("amount_max",    probe["amt_max"], -1000, 10000, ""),
    rule_zero_count("pickup_in_window", probe["pickup_outside"]),
]
for r in range_results:
    print(f"{r.name:<20} {'PASS' if r.passed else 'FIRED':<6} {r.observed:>16}   expected {r.expected}")
print(f"\ntrips over 200 mi: {probe['dist_over_200']:,}")
print(f"pickup span      : {probe['pickup_min']}  ..  {probe['pickup_max']}")

distance_min         PASS             0.0 mi   expected 0 to 200 mi
distance_max         FIRED      312,722.3 mi   expected 0 to 200 mi
amount_min           PASS           -1,000.0   expected -1,000 to 10,000
amount_max           PASS            9,792.0   expected -1,000 to 10,000
pickup_in_window     FIRED           21 rows   expected 0 rows

trips over 200 mi: 170
pickup span      : 2002-12-31 22:17:10  ..  2024-04-01 00:34:55

Two rules fired, and both are telling the truth about this data. distance_max at 312,722.3 miles is the broken odometer Module 6 flagged — roughly twelve times around the equator, in a vehicle that bills by the mile. pickup_in_window at 21 rows is the out-of-quarter stray set, and the printed span shows both ends of it: a pickup stamped 2002-12-31 22:17:10 twenty-two years early, and one at 2024-04-01 00:34:55 just past the window’s close. These are the same 21 rows Module 6’s window filter removed and Lesson 2 will quarantine with reasons instead of deleting.

Three rules passed, and one of those passes deserves a second look. amount_min observed exactly −1,000.00 against a lower bound of −1,000 — it passed on the boundary. A bound picked one dollar tighter would have fired. That is not a lucky escape; it is a signal that −1,000.00 is a vendor-side floor rather than a natural minimum, and a threshold sitting exactly on a real value in the data is a threshold you chose by coincidence rather than by reasoning. Write down why a bound is where it is, or a future engineer will move it by one and not know what broke.

Family 3: Referential — The Rule That Passes And Lies#

Referential integrity asks whether every foreign key resolves against its dimension. In Spark the idiom is a left_anti join: keep only the left-side rows that found no match, which is precisely the set of orphans. The zone dimension has 265 rows, so it broadcasts, and the whole check is one narrow-ish pass:

zones = (spark.read.option("header", True).csv("taxi_zone_lookup.csv")
              .withColumn("LocationID", F.col("LocationID").cast("int")))

orphans = (checks_frame
           .join(F.broadcast(zones), checks_frame.PULocationID == zones.LocationID, "left_anti")
           .count())

referential = rule_zero_count("zone_referential", orphans)
print(f"zone dimension rows : {zones.count()}")
print(f"{referential.name:<20} {'PASS' if referential.passed else 'FIRED':<6} "
      f"{referential.observed:>16}   expected {referential.expected}")

zone dimension rows : 265
zone_referential     PASS                  0 rows   expected 0 rows

Zero orphans across 9,554,778 trips. Every single PULocationID in the quarter resolves against the 265-row lookup. By the strict definition of referential integrity, this dataset is flawless, and a suite containing only this rule would report a clean bill of health.

Now look at what the dimension actually contains at the top of its ID range:

zones.filter(F.col("LocationID").isin(264, 265)).show(truncate=False)

junk = checks_frame.filter(F.col("PULocationID").isin(264, 265)).count()
meaningful = rule_zero_count("zone_meaningful", junk, severity="warn")
print(f"{meaningful.name:<20} {'PASS' if meaningful.passed else 'FIRED':<6} "
      f"{meaningful.observed:>16}   expected {meaningful.expected}")
print(f"share of quarter    : {junk / 9_554_778:.2%}")

+----------+-------+--------------+------------+
|LocationID|Borough|Zone          |service_zone|
+----------+-------+--------------+------------+
|264       |Unknown|NULL          |NULL        |
|265       |NULL   |Outside of NYC|NULL        |
+----------+-------+--------------+------------+

zone_meaningful      FIRED       36,760 rows   expected 0 rows
share of quarter    : 0.38%

Here is the honest nuance the whole family turns on. 264 is “Unknown” with a null zone name, and 265 is “Outside of NYC” with a null borough. Neither is a place. They are the dimension’s way of saying we could not determine the location, and because they are legitimate rows in the lookup table, 36,760 trips — 0.38% of the quarter — join successfully to them and sail past the referential rule. Those trips have a zone id, a zone row, and no zone.

Referential-valid does not mean meaningful. The left_anti check proves only that the key exists in the dimension; it says nothing about whether the dimension row carries information. Any downstream report that groups by Zone will produce a bucket named NULL holding 36,760 trips, or — worse — will drop them in a chart that filters out nulls, and the totals will silently stop reconciling. So the family needs two rules: the classic integrity check, which passes, and a domain rule asserting that keys resolve to substantive dimension rows, which fires at warn severity. The second rule is the one that would have caught this, and you only think to write it after you have looked at what the dimension actually says.

Family 4: Nullability and Uniqueness#

The last family covers the two structural properties a table is supposed to guarantee: columns that should never be null, and rows that should never repeat. Nullability first, as one aggregate over the cached frame:

NEVER_NULL = ["passenger_count", "RatecodeID", "store_and_fwd_flag",
              "congestion_surcharge", "Airport_fee"]

null_counts = checks_frame.agg(
    *[F.sum(F.col(c).isNull().cast("int")).alias(c) for c in NEVER_NULL]
).first().asDict()

null_results = [rule_zero_count(f"not_null_{c}", n, severity="warn")
                for c, n in null_counts.items()]
for r in null_results:
    print(f"{r.name:<30} {'PASS' if r.passed else 'FIRED':<6} {r.observed:>16}")

co_located = checks_frame.filter(
    " AND ".join(f"{c} IS NULL" for c in NEVER_NULL)).count()
print(f"\nrows where ALL FIVE are null: {co_located:,}")

not_null_passenger_count       FIRED      751,962 rows
not_null_RatecodeID            FIRED      751,962 rows
not_null_store_and_fwd_flag    FIRED      751,962 rows
not_null_congestion_surcharge  FIRED      751,962 rows
not_null_Airport_fee           FIRED      751,962 rows

rows where ALL FIVE are null: 751,962

Five rules fire with the identical count, 751,962 each — and the follow-up query proves they are not five independent problems. All five columns are null in the same 751,962 rows. That co-location is diagnostic: five unrelated fields do not go missing together by chance, so this is one upstream cause (a vendor feed that omits the optional metadata block) affecting 7.9% of trips, not five separate data-quality incidents. The trips themselves are complete — they have timestamps, distances, locations, and money.

This is why these rules carry severity="warn" rather than critical. The finding is real and worth logging on every run, but aborting a pipeline over it would mean never processing this dataset at all, and dropping the rows would delete three quarters of a million genuine trips. The rule’s job is to make the gap visible and quantified, so that if the count ever jumps to three million someone notices the upstream change. A rule’s severity encodes what you intend to do about it, and “keep processing, but tell me” is a legitimate intention that a bare assert cannot express.

Uniqueness is the other half, and it needs the full row rather than the projection — two trips are duplicates only if every column matches:

dupe_results = []
for label, path in MONTHS.items():
    month = spark.read.schema(TAXI_SCHEMA).parquet(path)
    excess = month.count() - month.distinct().count()
    dupe_results.append(rule_zero_count(f"no_duplicate_rows_{label}", excess, severity="warn"))

for r in dupe_results:
    print(f"{r.name:<30} {'PASS' if r.passed else 'FIRED':<6} {r.observed:>16}")

no_duplicate_rows_2024-01      PASS             0 rows
no_duplicate_rows_2024-02      FIRED            1 rows
no_duplicate_rows_2024-03      PASS             0 rows

January and March are clean. February carries one excess row — meaning one pair of byte-identical trips, matching Course 1’s finding of two duplicate rows in February (the pair; this rule counts the surplus, which is one). Be precise about which convention your rule reports, because “2 duplicates” and “1 excess row” describe the same fact and a reconciliation check will disagree with you if you mix them. Note the cost, too: distinct() on full rows is a shuffle over all 19 columns and is by far the most expensive rule in this suite — genuine, but not something to run per-batch without thinking about it.

Running The Suite#

Individual rules are diagnostics; a suite is the artifact a reviewer reads. Collect every result and print one table, with the summary line a gate will later branch on:

def run_suite(results):
    header = f"{'rule':<30}{'severity':<10}{'result':<8}{'observed':>20}   expected"
    print(header)
    print("-" * len(header))
    for r in results:
        print(f"{r.name:<30}{r.severity:<10}{'PASS' if r.passed else 'FIRED':<8}"
              f"{r.observed:>20}   {r.expected}")
    fired = [r for r in results if not r.passed]
    critical = [r for r in fired if r.severity == "critical"]
    print("-" * len(header))
    print(f"{len(results)} rules  |  {len(results) - len(fired)} passed  |  "
          f"{len(fired)} fired  |  {len(critical)} critical")
    return fired, critical

suite = volume_results + range_results + [referential, meaningful] + null_results + dupe_results
fired, critical = run_suite(suite)

rule                          severity  result              observed   expected
-------------------------------------------------------------------------------
volume_2024-01                critical  PASS          2,964,624 rows   2,250,000-3,750,000
volume_2024-02                critical  PASS          3,007,526 rows   2,250,000-3,750,000
volume_2024-03                critical  PASS          3,582,628 rows   2,250,000-3,750,000
distance_min                  critical  PASS                  0.0 mi   0 to 200 mi
distance_max                  critical  FIRED           312,722.3 mi   0 to 200 mi
amount_min                    critical  PASS                -1,000.0   -1,000 to 10,000
amount_max                    critical  PASS                 9,792.0   -1,000 to 10,000
pickup_in_window              critical  FIRED                21 rows   0 rows
zone_referential              critical  PASS                  0 rows   0 rows
zone_meaningful               warn      FIRED            36,760 rows   0 rows
not_null_passenger_count      warn      FIRED           751,962 rows   0 rows
not_null_RatecodeID           warn      FIRED           751,962 rows   0 rows
not_null_store_and_fwd_flag   warn      FIRED           751,962 rows   0 rows
not_null_congestion_surcharge warn      FIRED           751,962 rows   0 rows
not_null_Airport_fee          warn      FIRED           751,962 rows   0 rows
no_duplicate_rows_2024-01     warn      PASS                  0 rows   0 rows
no_duplicate_rows_2024-02     warn      FIRED                 1 rows   0 rows
no_duplicate_rows_2024-03     warn      PASS                  0 rows   0 rows
-------------------------------------------------------------------------------
18 rules  |  9 passed  |  9 fired  |  2 critical

Eighteen rules, nine passed, nine fired, two of them critical — on the very dataset this course has been computing correct answers from for seven modules. Nothing here is new damage; every defect the suite names has been in the files the whole time. Module 6’s pipeline reads all of it and produces $256,692,373.14 without a murmur, because a pipeline that does not assert cannot object. What changed in this lesson is that the defects now have names, counts, and severities, printed on every run.

The fired and critical lists are the handoff to the rest of the module. fired is what Lesson 3 writes to the run log so tomorrow you can explain last night. critical is what Lesson 4 turns into an exit code — and note that with 2 criticals firing, a naive “abort on any critical failure” gate would refuse to process this quarter at all, which tells you the severities need the same calibration the thresholds do.

Calibration: The Rule That Never Fires#

Everything above rests on one claim, and it deserves to be demonstrated rather than asserted. Take the rule that fired hardest — the 312,722.3-mile distance — and “fix” it the way a tired engineer fixes a noisy check, by raising the ceiling until it stops complaining:

recalibrated = rule_range("distance_max_calibrated", probe["dist_max"], 0, 400_000, " mi")
print(f"{recalibrated.name:<26} {'PASS' if recalibrated.passed else 'FIRED':<6} "
      f"{recalibrated.observed:>16}   expected {recalibrated.expected}")
print()
for cutoff in [50, 100, 200, 500, 1000]:
    n = checks_frame.filter(F.col("trip_distance") > cutoff).count()
    print(f"  rows over {cutoff:>5} mi : {n:>8,}   ({n / 9_554_778:.4%} of the quarter)")

distance_max_calibrated    PASS       312,722.3 mi   expected 0 to 400,000 mi

  rows over    50 mi :    1,263   (0.0132% of the quarter)
  rows over   100 mi :      252   (0.0026% of the quarter)
  rows over   200 mi :      170   (0.0018% of the quarter)
  rows over   500 mi :      162   (0.0017% of the quarter)
  rows over  1000 mi :      158   (0.0017% of the quarter)

The recalibrated rule passes. It passes on a taxi that allegedly drove a third of a million miles. It will pass on every batch forever, it will never generate an alert, and a dashboard of rule results will show it green next to the rules doing real work. That rule is not validation — it is a comment with a passed field, and it is exactly what a suite decays into when failures are silenced by moving the threshold rather than investigated.

The sweep shows where the real boundary sits, and it is remarkably sharp. Between 200 and 1,000 miles the count barely moves — 170 rows down to 158 — because there is no continuum of increasingly-implausible trips there. There is a small population of records with garbage odometer values, and they are all far beyond any threshold in that range. Below 200, the count starts climbing into real trips: 1,263 rows over 50 miles is largely genuine long-haul work (airport runs to the far suburbs, out-of-city fares), so a 50-mile ceiling would fire on legitimate data and train everyone to ignore it. A well-calibrated threshold sits in the flat part of that curve, where moving it a bit either way changes almost nothing — which is the empirical definition of a boundary the data itself put there. The 200-mile ceiling flags 0.0018% of the quarter: rare enough to be worth reading, common enough to prove the rule executes.

So the lesson’s spine, stated plainly: a rule that never fires is either describing perfect data or was written wrong. Perfect data does not exist at 9.5 million rows. When a suite reports all-green for weeks, the correct reaction is not satisfaction but suspicion — go and break something on purpose, the way this lesson truncated February to 300,000 rows, and confirm the rules still have teeth. A validation suite is code, and untested code does not work; the difference is that a broken test fails loudly while a broken rule passes quietly forever.

Shutting Down#

Release the cache and the session:

import shutil, os
checks_frame.unpersist()
shutil.rmtree("spark-warehouse", ignore_errors=True)
print("cached frame released; scratch cleaned")
spark.stop()
print("session stopped")

cached frame released; scratch cleaned
session stopped

Practice Exercises#

Exercise 1 — Add a distribution rule to the suite. The four families here all check extremes or counts, which misses a whole class of failure: a feed where every value is individually plausible but the shape has shifted. Write rule_mean_shift(name, observed, baseline, tolerance) asserting that the quarter’s mean trip_distance sits within 20% of a baseline you measure yourself from January alone, then run it against February and March. Add it to suite and re-run run_suite. Then break it deliberately: filter the frame to trips over 5 miles and confirm the rule fires.

Exercise 2 — Make the referential rule report which keys are the problem. zone_meaningful fired with 36,760 rows but does not say how those trips split between code 264 and code 265, and an engineer investigating needs that. Extend the rule so its observed field carries the per-code breakdown, then go further: write a rule asserting that no single zone accounts for more than 10% of the quarter’s trips, and check it against the busiest zones (161 Midtown Center at 453,825 trips, 237 Upper East Side South at 439,138, 132 JFK at 429,745).

Exercise 3 — Unit-test the rule functions on hand-made inputs. rule_row_volume, rule_range, and rule_zero_count are plain Python taking plain numbers, so they can be tested in microseconds with no Spark at all. Write test_rules() with plain asserts covering the cases that matter: a volume exactly on each band edge (does your rule use <= or <?), a zero-count rule given 0 and given 1, and a range rule given a value exactly equal to high. Then add the case the suite has not handled: what should a rule return when its observed value is None because the column was entirely null?

Summary#

A validation rule is an executable assertion about the data, evaluated every run — a small function returning (name, passed, observed, expected, severity) rather than a bare assert, because a verdict without the observed and expected values leaves whoever gets paged with no diagnosis. This lesson built that shape and filled out four families on the real quarter. Volume used a tolerance band (3,000,000 ±25%) that passes January’s 2,964,624, February’s 3,007,526, and March’s 3,582,628 — March runs 19.4% above centre, so a tighter band would cry wolf every spring — while a genuine 300,000-row truncated slice of February FIRED immediately, the one failure mode no structural check can see. Range and domain computed eight statistics in a single agg over the cached frame: trip_distance min 0.0 passed, max 312,722.3 mi FIRED, total_amount spanning −1,000.00 to 9,792.00 passed (the minimum landing exactly on the boundary), and 21 pickups outside the quarter window FIRED, spanning 2002-12-31 22:17:10 to 2024-04-01 00:34:55. Referential used a left_anti join against the 265-row zone dimension and found 0 orphans — then the honest nuance that defines the family: junk codes 264 “Unknown” and 265 “Outside of NYC” resolve perfectly while naming no real place, so 36,760 trips (0.38%) are referentially valid and analytically meaningless, and only a second rule catches them. Nullability fired five times at 751,962 each, co-located in the same 751,962 rows — one upstream cause, not five incidents — and uniqueness found 1 excess row in February against 0 in January and March. Run as a suite: 18 rules, 9 passed, 9 fired, 2 critical, on data this course has computed correct answers from for seven modules. The calibration sweep closed the argument: raising the distance ceiling to 400,000 miles makes the rule pass forever on a broken odometer, while the row counts over 50/100/200/500/1,000 miles (1,263 / 252 / 170 / 162 / 158) show the real boundary is flat between 200 and 1,000 — the empirical signature of a threshold the data itself put there.

Key Concepts#

• A rule is a function, not a comment — returning (name, passed, observed, expected, severity) so a failure carries its own diagnosis; the measurement happens in Spark, the judgment in plain Python, which makes rules unit-testable without a session.
• Volume checks use a band, not a number — 3,000,000 ±25% passes all three real months including March at +19.4%, yet fires on a 300,000-row truncated export, the failure that every column-level check is blind to.
• Referential-valid ≠ meaningful — left_anti against the 265-row dimension returns 0 orphans, but sentinel rows 264 and 265 absorb 36,760 trips that have an id, a dimension row, and no zone; every referential rule needs a companion rule excluding sentinels.
• Severity encodes intent — critical means abort the run, warn means log it and continue; the 751,962 co-located nulls are a warn because aborting would mean never processing this feed and dropping would delete 7.9% of genuine trips.
• A rule that never fires is a rule written wrong — a 400,000-mile ceiling passes a broken odometer forever; calibrate into the flat part of the threshold curve (200–1,000 mi moves the count only 170→158) so a rule is informative rather than decorative.

Why This Matters#

The gap between a pipeline that runs and a pipeline you can trust is not sophistication — it is whether the job has any opinion about what it is reading. Module 6’s ETL is well-structured, well-tested, and completely credulous: hand it a tenth of a month and it computes a tenth of the revenue with total confidence, because nowhere in it does anything claim what the data should be. The eighteen rules here are that claim, written down in a form that executes at 2 a.m. whether or not anyone is watching, and they immediately found nine real defects in a dataset seven modules of correct answers have been computed from. That is the ordinary state of production data: not broken, but never clean, and silently degrading in ways only an assertion catches.

Everything the rest of the module builds needs this lesson’s output as its input. A rule that fires has to go somewhere — the 21 out-of-window strays cannot simply be filtered into oblivion, because a row deleted without a record is a question nobody can answer later. A run’s results have to be recorded in a form you can grep next Tuesday, which means the RuleResult fields become log fields. And the critical subset has to become a decision: warn and continue, or abort before writing and exit nonzero so a scheduler notices. The suite you just wrote is the sensor; the next three lessons build the response. And the habit worth carrying past this course is the suspicious one: when the table comes back all green, do not celebrate — go break something on purpose and check that the rules still have teeth.

Continue Building Your Skills#

You now have a suite that names nine real defects in the quarter, and an uncomfortable question it leaves open: the 21 out-of-window strays failed a critical rule, so what happens to them? Module 6’s answer was filter(...) — the rows vanish, the counts no longer reconcile with the input, and nobody can ever inspect what was rejected or decide whether the rule or the data was wrong. Lesson 2, Quarantine Patterns, replaces that silent deletion with a discipline: split each batch into GOOD and QUARANTINED, attach a reason column naming the rule that rejected each row, and write both — the quarantine to its own partitioned path keyed by reason and run id. The payoff is an identity a reviewer can check on any run, and on this quarter it is exact: 9,554,757 good plus 21 quarantined equals the 9,554,778 that arrived. You’ll also see how a quarantine table gets reviewed and selectively re-ingested after an upstream fix, and why the 115,894 reversals must stay firmly in GOOD. Rules tell you what is wrong; quarantine is what you do about it without losing the evidence.