Lesson 5 - Guided Project: A Guarded ETL Run

Welcome to A Guarded ETL Run#

Module 6 ended with a job you could hand to a scheduler: quarterly_etl.py read three months of taxi data under an explicit schema, cleaned it, aggregated it to the zone-hour report, and wrote it idempotently so a retry could never double the books. It was correct, and it was repeatable. It was also, in one specific and dangerous way, naive — it assumed the data arriving each night would look like the data that arrived tonight. Hand that job a March export that failed halfway through, and it will cheerfully process the fragment it received, overwrite the published table with a third of a quarter, and exit 0. The dashboards will be wrong by morning, and nothing in the run will have said so.

This project fixes that. You will take the same ETL and wrap it in everything this module taught: the executable validation rules from Lesson 1, the quarantine with reasons from Lesson 2, the structured run summary from Lesson 3, and the warn-versus-critical gate with a real exit code from Lesson 4. The result is guarded_etl.py — a job safe to run unattended. Then comes the part that makes this a proof rather than a description: you will run the script as an actual subprocess, twice, and read its real exit codes back into Python. The good run publishes and exits 0. The bad run — a deliberately truncated March file standing in for a half-failed export — trips a critical rule, aborts before the write, exits 2, and leaves the previously published table completely untouched. That last sentence is the whole point of the module, and by the end of this lesson you will have watched it happen with real numbers.

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

• Assemble input validation, quarantine, output reconciliation, gating, and structured logging into one reusable guarded ETL script
• Order the stages so the quality gate fires before the write, making a bad run incapable of publishing bad data
• Emit a machine-readable JSON run summary with per-stage counts and durations that answers “what did last night’s run do?”
• Return a real process exit code and verify it from a calling process, the way a scheduler will
• Prove that a failed run leaves the previously published table byte-identical, and carry away a six-point data-quality checklist

You’ll need pyspark with its Java runtime; everything else is Python’s standard library. Let’s build the guarded job.

The Session, the Contract, and the Thresholds#

One session for the whole project. Alongside the schema from Module 6, the guarded job needs something new: the numbers that define “normal.” A validation rule is only as good as the threshold it compares against, and those thresholds are a judgment call you make once, in daylight, so the job can make the decision at 2 a.m. without you. We declare three: a minimum row count for a quarter (MIN_ROWS), and a plausible revenue band (REV_LO, REV_HI). They live at the top of the file where a reviewer can argue with them.

import warnings; warnings.filterwarnings("ignore")
import datetime as dt, glob, json, os, re, shutil, subprocess, sys, time
from pathlib import Path
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")

FILES = [f"yellow_tripdata_2024-{m}.parquet" for m in ("01", "02", "03")]
LO, HI = dt.datetime(2024, 1, 1), dt.datetime(2024, 4, 1)
MIN_ROWS = 9_000_000
REV_LO, REV_HI = 100_000_000.0, 500_000_000.0

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

zones = (spark.read.option("header", True).csv("taxi_zone_lookup.csv")
         .select(F.col("LocationID").cast("int").alias("LocationID"), "Zone"))
raw = spark.read.schema(taxi_schema()).parquet(*FILES)
print("schema fields :", len(taxi_schema().fields))
print("zone dimension:", zones.count(), "rows")

schema fields : 19
zone dimension: 265 rows

Nineteen typed fields and 265 zones — the same contract and dimension the job has used since Module 6. What changes in this lesson is everything that happens around them.

Stage 1: Validate the Input Before You Trust It#

Lesson 1 established what a validation rule is: a small function returning a name, a level, whether it passed, what it observed, and what it expected. The guarded job runs a suite of them over the raw input before a single row is transformed. Four rule families cover the ways a batch feed goes wrong:

• Volume — did we receive roughly a quarter’s worth of data? A quarter is about 9.5 million trips; anything under 9 million means a file is missing or truncated. This is critical.
• Referential — does every PULocationID in the input exist in the 265-row zone dimension? Checked with a left_anti join against the distinct codes, which is cheap because there are only a few hundred of them. Also critical.
• Nullability — are there rows we cannot key at all, with a null pickup zone? critical.
• Range/domain — how far outside the window do pickups stray, and how absurd does trip_distance get? These are warn: they describe dirt we already know how to handle, not a broken feed.

The important engineering detail is that four of the five metrics come from one aggregate pass over the input. Validation costs actions, and a naive suite that calls .count() once per rule re-scans the data five times. Rolling the counts into a single agg with F.sum(F.when(...)) conditionals gets them all for the price of one scan.

def rule(name, level, passed, observed, expected):
    return {"rule": name, "level": level, "passed": bool(passed),
            "observed": observed, "expected": expected}

def show_rules(rules):
    print(f"{'rule':<30}{'level':<10}{'result':<8}{'observed':<16}expected")
    for r in rules:
        print(f"{r['rule']:<30}{r['level']:<10}"
              f"{'PASS' if r['passed'] else 'FAIL':<8}{str(r['observed']):<16}{r['expected']}")

def out_of_window():
    return ((F.col("tpep_pickup_datetime") < F.lit(LO)) |
            (F.col("tpep_pickup_datetime") >= F.lit(HI)))

def validate_input(raw, zones):
    m = raw.agg(
        F.count(F.lit(1)).alias("rows"),
        F.sum(F.when(out_of_window(), 1).otherwise(0)).alias("out_of_window"),
        F.sum(F.when(F.col("PULocationID").isNull(), 1).otherwise(0)).alias("null_pickup_zone"),
        F.max("trip_distance").alias("max_distance"),
    ).collect()[0].asDict()
    codes = raw.select("PULocationID").distinct()
    m["orphan_zone_codes"] = codes.join(
        zones, codes.PULocationID == zones.LocationID, "left_anti").count()
    rows = m["rows"]
    return [
        rule("input.volume", "critical", rows >= MIN_ROWS, rows, f">= {MIN_ROWS:,}"),
        rule("input.referential_zone", "critical",
             m["orphan_zone_codes"] == 0, m["orphan_zone_codes"], "0 orphan codes"),
        rule("input.pickup_zone_not_null", "critical",
             m["null_pickup_zone"] == 0, m["null_pickup_zone"], "0 nulls"),
        rule("input.pickup_in_window", "warn",
             m["out_of_window"] <= rows * 0.001, m["out_of_window"], "<= 0.1% of rows"),
        rule("input.distance_range", "warn",
             m["max_distance"] <= 200, round(m["max_distance"], 1), "<= 200 miles"),
    ], m

t = time.time()
input_rules, metrics = validate_input(raw, zones)
show_rules(input_rules)
print(f"\none pass over the quarter: {time.time() - t:.1f}s")

rule                          level     result  observed        expected
input.volume                  critical  PASS    9554778         >= 9,000,000
input.referential_zone        critical  PASS    0               0 orphan codes
input.pickup_zone_not_null    critical  PASS    0               0 nulls
input.pickup_in_window        warn      PASS    21              <= 0.1% of rows
input.distance_range          warn      FAIL    312722.3        <= 200 miles

one pass over the quarter: 2.1s

Five rules, 2.1 seconds, and an honest result: four pass and one fails. The quarter has 9,554,778 rows, comfortably over the 9-million floor; zero orphan zone codes and zero null pickup zones, so every row can be keyed and joined; and 21 out-of-window pickups, far under the 0.1% tolerance, so that rule passes as a warn-level observation rather than a complaint.

The failure is the interesting one. input.distance_range observes a maximum trip_distance of 312,722.3 miles — a taxi ride roughly to the moon and back, and Lesson 1’s canonical example of a rule that actually fires. It is marked warn deliberately: a single absurd odometer reading in 9.5 million rows is a known data defect that the downstream aggregation tolerates, not evidence that the feed is broken. A rule suite where everything always passes is a suite you should distrust; this one earns its keep by telling you something true and letting you decide it isn’t fatal.

Stage 2: Clean by Quarantining, and Reconcile#

Lesson 2’s rule was blunt: never silently drop a row. The clean stage splits the batch into good and quarantined, attaches a quarantine_reason to every held-back row, and the two counts must add back up to the input. That last identity is the one a reviewer can check in five seconds without reading any of your code.

def clean(raw):
    keyed = ~out_of_window() & F.col("PULocationID").isNotNull()
    good = raw.filter(keyed)
    bad = raw.filter(~keyed).withColumn(
        "quarantine_reason",
        F.when(F.col("PULocationID").isNull(), F.lit("null_pickup_zone"))
         .otherwise(F.lit("pickup_outside_window")))
    return good, bad

good, bad = clean(raw)
counts = {"input": metrics["rows"], "good": good.count(), "quarantined": bad.count()}
print("input       :", f"{counts['input']:,}")
print("good        :", f"{counts['good']:,}")
print("quarantined :", counts["quarantined"])
print("reconciles  :", counts["good"] + counts["quarantined"] == counts["input"],
      f"({counts['good']:,} + {counts['quarantined']} = {counts['good'] + counts['quarantined']:,})")
bad.groupBy("quarantine_reason").count().show(truncate=False)
bad.select("tpep_pickup_datetime", "PULocationID", "total_amount",
           "quarantine_reason").orderBy("tpep_pickup_datetime").show(3, truncate=False)

input       : 9,554,778
good        : 9,554,757
quarantined : 21
reconciles  : True (9,554,757 + 21 = 9,554,778)
+---------------------+-----+
|quarantine_reason    |count|
+---------------------+-----+
|pickup_outside_window|21   |
+---------------------+-----+

+--------------------+------------+------------+---------------------+
|tpep_pickup_datetime|PULocationID|total_amount|quarantine_reason    |
+--------------------+------------+------------+---------------------+
|2002-12-31 22:17:10 |50          |18.0        |pickup_outside_window|
|2002-12-31 22:59:39 |170         |-10.5       |pickup_outside_window|
|2002-12-31 22:59:39 |170         |10.5        |pickup_outside_window|
+--------------------+------------+------------+---------------------+
only showing top 3 rows

9,554,757 + 21 = 9,554,778. Nothing vanished. All 21 held-back rows carry the same reason, pickup_outside_window, and the first three are the module’s familiar strays — pickups stamped 2002-12-31, a decade before the quarter began. Note the second and third rows: a −$10.50 and a +$10.50 on the same timestamp and zone, a reversal pair that happens to sit outside the window. Because they are quarantined rather than deleted, an analyst can look at them tomorrow, decide they were a clock fault at one vendor, and re-ingest them with a corrected timestamp. A filter() that had thrown them away would have left nothing to look at.

Stage 3 and 4: Transform, Then Validate What You Produced#

Validating the input tells you the data arriving was plausible. It says nothing about whether your own transform was correct. A join that silently drops rows, an aggregation keyed on the wrong column, a rounding drift — none of those are input problems, and all of them produce a confidently wrong table. So the guarded job validates its output too, and the strongest output rules are reconciliation rules: arithmetic identities that must hold if the transform did what it claimed.

Two of them matter here. The aggregated trips total must equal the cleaned row count — if the broadcast join dropped a single trip, that equality breaks. And the good-plus-quarantined identity must still hold. Two more sanity rules check that the report isn’t empty and that revenue lands inside the plausible band.

def transform(good, zones):
    hourly = good.withColumn("pickup_hour", F.date_trunc("hour", "tpep_pickup_datetime"))
    joined = hourly.join(F.broadcast(zones), hourly.PULocationID == zones.LocationID, "inner")
    return (joined.groupBy("PULocationID", "Zone", "pickup_hour")
            .agg(F.count(F.lit(1)).alias("trips"),
                 F.round(F.sum("total_amount"), 2).alias("revenue"))
            .withColumn("pickup_date", F.to_date("pickup_hour")))

def validate_output(report, counts):
    agg = report.agg(F.count(F.lit(1)).alias("buckets"),
                     F.sum("trips").alias("trips"),
                     F.round(F.sum("revenue"), 2).alias("revenue")).collect()[0].asDict()
    return [
        rule("recon.good_plus_quarantined", "critical",
             counts["good"] + counts["quarantined"] == counts["input"],
             f"{counts['good']}+{counts['quarantined']}", counts["input"]),
        rule("recon.trips_equal_cleaned", "critical",
             agg["trips"] == counts["good"], agg["trips"], counts["good"]),
        rule("output.buckets_nonzero", "critical", agg["buckets"] > 0, agg["buckets"], "> 0"),
        rule("output.revenue_range", "critical",
             REV_LO <= agg["revenue"] <= REV_HI, agg["revenue"],
             f"{REV_LO:,.0f}..{REV_HI:,.0f}"),
    ], agg

report = transform(good, zones)
output_rules, agg = validate_output(report, counts)
show_rules(output_rules)
print(f"\nbuckets {agg['buckets']:,} | trips {agg['trips']:,} | revenue ${agg['revenue']:,.2f}")

rule                          level     result  observed        expected
recon.good_plus_quarantined   critical  PASS    9554757+21      9554778
recon.trips_equal_cleaned     critical  PASS    9554757         9554757
output.buckets_nonzero        critical  PASS    240917          > 0
output.revenue_range          critical  PASS    256692373.14    100,000,000..500,000,000

buckets 240,917 | trips 9,554,757 | revenue $256,692,373.14

All four output rules pass, and the report is the one this course keeps returning to: 240,917 zone-hour buckets, 9,554,757 trips, $256,692,373.14 in revenue — Course 1’s verified ground truth, reproduced by a different engine. The recon.trips_equal_cleaned rule is the quiet hero: 9,554,757 aggregated trips against 9,554,757 cleaned rows is arithmetic proof that the broadcast join matched every single trip to a zone. If someone later “optimizes” the join into an inner join on the zone name, the two “Corona” zones will start duplicating and dropping rows, and this rule will fail on the very next run instead of six weeks later in a board meeting.

The Gate: Warn Versus Critical#

Now the decision Lesson 4 built. A gate takes the rule results and answers one question: may this run write? Warn-level failures are logged and the job continues. Critical failures stop it. The gate is deliberately trivial code — the intelligence lives in how each rule was classified, which is a human judgment recorded once in the rule definitions.

def gate(rules):
    return [r for r in rules if not r["passed"] and r["level"] == "critical"]

all_rules = input_rules + output_rules
print("rules evaluated  :", len(all_rules))
print("warn-level failed:", [r["rule"] for r in all_rules
                             if not r["passed"] and r["level"] == "warn"])
print("critical failed  :", [r["rule"] for r in gate(all_rules)])
print("gate decision    :", "ABORT" if gate(all_rules) else "PROCEED TO WRITE")

pretend = all_rules + [rule("input.volume", "critical", False, 6_072_150, ">= 9,000,000")]
print("if volume had failed:", "ABORT" if gate(pretend) else "PROCEED TO WRITE",
      "->", [r["rule"] for r in gate(pretend)])

rules evaluated  : 9
warn-level failed: ['input.distance_range']
critical failed  : []
gate decision    : PROCEED TO WRITE
if volume had failed: ABORT -> ['input.volume']

Nine rules, one warn-level failure, zero critical failures — proceed. The 312,722-mile trip is noted in the log and does not stop the run, which is exactly right: it is a known defect in one row out of nine and a half million. The second line shows the same gate handed a hypothetical volume failure, and the answer flips to ABORT. Same code, same rules list, different classification of one entry.

The Deliverable: guarded_etl.py#

Everything above, assembled into the file you keep. This is a real script with a main(), command-line arguments for the input files and both output paths, Python’s logging configured the way Lesson 3 established, and — the piece that makes it a job rather than a notebook — sys.exit() with a meaningful code. Note the ordering inside main(): validate_input and its gate run before clean; validate_output and its gate run before the write. Every gate sits upstream of the thing it protects.

The block below writes the script to disk so we can run it. (We namespace the file for this module’s scratch directory; in your own repo, save it as guarded_etl.py.)

SCRIPT = r'''"""guarded_etl.py - CityFlow's quarterly zone-hour ETL, with quality gates.

Exit codes: 0 = published, 2 = aborted by a critical rule (nothing written).
"""
import warnings; warnings.filterwarnings("ignore")
import argparse, datetime as dt, json, logging, sys, time, uuid
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import (StructType, StructField, IntegerType, LongType,
                               DoubleType, StringType, TimestampNTZType)

LO, HI = dt.datetime(2024, 1, 1), dt.datetime(2024, 4, 1)
MIN_ROWS = 9_000_000
REV_LO, REV_HI = 100_000_000.0, 500_000_000.0

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s",
                    datefmt="%H:%M:%S", stream=sys.stdout)
log = logging.getLogger("guarded-etl")


def rule(name, level, passed, observed, expected):
    return {"rule": name, "level": level, "passed": bool(passed),
            "observed": observed, "expected": expected}


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


def build_session():
    spark = (SparkSession.builder.appName("cityflow-guarded-etl")
             .master("local[*]")
             .config("spark.ui.showConsoleProgress", "false")
             .getOrCreate())
    spark.sparkContext.setLogLevel("ERROR")
    spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
    return spark


def out_of_window():
    return ((F.col("tpep_pickup_datetime") < F.lit(LO)) |
            (F.col("tpep_pickup_datetime") >= F.lit(HI)))


def validate_input(raw, zones):
    """One pass over the input; returns (rules, metrics)."""
    m = raw.agg(
        F.count(F.lit(1)).alias("rows"),
        F.sum(F.when(out_of_window(), 1).otherwise(0)).alias("out_of_window"),
        F.sum(F.when(F.col("PULocationID").isNull(), 1).otherwise(0)).alias("null_pickup_zone"),
        F.max("trip_distance").alias("max_distance"),
    ).collect()[0].asDict()
    codes = raw.select("PULocationID").distinct()
    orphans = codes.join(zones, codes.PULocationID == zones.LocationID, "left_anti").count()
    m["orphan_zone_codes"] = orphans
    rows = m["rows"]
    rules = [
        rule("input.volume", "critical", rows >= MIN_ROWS, rows, f">= {MIN_ROWS:,}"),
        rule("input.referential_zone", "critical", orphans == 0, orphans, "0 orphan codes"),
        rule("input.pickup_zone_not_null", "critical",
             m["null_pickup_zone"] == 0, m["null_pickup_zone"], "0 nulls"),
        rule("input.pickup_in_window", "warn",
             m["out_of_window"] <= rows * 0.001, m["out_of_window"], "<= 0.1% of rows"),
        rule("input.distance_range", "warn",
             m["max_distance"] <= 200, round(m["max_distance"], 1), "<= 200 miles"),
    ]
    return rules, m


def clean(raw):
    keyed = ~out_of_window() & F.col("PULocationID").isNotNull()
    good = raw.filter(keyed)
    bad = raw.filter(~keyed).withColumn(
        "quarantine_reason",
        F.when(F.col("PULocationID").isNull(), F.lit("null_pickup_zone"))
         .otherwise(F.lit("pickup_outside_window")))
    return good, bad


def transform(good, zones):
    hourly = good.withColumn("pickup_hour", F.date_trunc("hour", "tpep_pickup_datetime"))
    joined = hourly.join(F.broadcast(zones), hourly.PULocationID == zones.LocationID, "inner")
    return (joined.groupBy("PULocationID", "Zone", "pickup_hour")
            .agg(F.count(F.lit(1)).alias("trips"),
                 F.round(F.sum("total_amount"), 2).alias("revenue"))
            .withColumn("pickup_date", F.to_date("pickup_hour")))


def validate_output(report, counts):
    agg = report.agg(F.count(F.lit(1)).alias("buckets"),
                     F.sum("trips").alias("trips"),
                     F.round(F.sum("revenue"), 2).alias("revenue")).collect()[0].asDict()
    recon_in = counts["good"] + counts["quarantined"] == counts["input"]
    rules = [
        rule("recon.good_plus_quarantined", "critical", recon_in,
             f"{counts['good']} + {counts['quarantined']}", counts["input"]),
        rule("recon.trips_equal_cleaned", "critical", agg["trips"] == counts["good"],
             agg["trips"], counts["good"]),
        rule("output.buckets_nonzero", "critical", agg["buckets"] > 0, agg["buckets"], "> 0"),
        rule("output.revenue_range", "critical",
             REV_LO <= agg["revenue"] <= REV_HI, agg["revenue"],
             f"{REV_LO:,.0f}..{REV_HI:,.0f}"),
    ]
    return rules, agg


def gate(rules):
    """Return the list of failed CRITICAL rules; warn-level failures only log."""
    for r in rules:
        if r["passed"]:
            continue
        log.warning("RULE FAILED [%s] %s observed=%s expected=%s",
                    r["level"], r["rule"], r["observed"], r["expected"])
    return [r for r in rules if not r["passed"] and r["level"] == "critical"]


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--files", nargs="+", required=True)
    ap.add_argument("--out", required=True)
    ap.add_argument("--quarantine", required=True)
    args = ap.parse_args()

    run_id = uuid.uuid4().hex[:8]
    t0 = time.time()
    stages, counts = {}, {}
    spark = build_session()
    exit_code = 0
    try:
        zones = (spark.read.option("header", True).csv("taxi_zone_lookup.csv")
                 .select(F.col("LocationID").cast("int").alias("LocationID"), "Zone"))
        raw = spark.read.schema(taxi_schema()).parquet(*args.files)

        log.info("run_id=%s stage=validate_input files=%d", run_id, len(args.files))
        s = time.time()
        rules, m = validate_input(raw, zones)
        stages["validate_input"] = round(time.time() - s, 1)
        counts["input"] = m["rows"]
        failed = gate(rules)
        if failed:
            raise SystemExit(2)

        log.info("run_id=%s stage=clean_quarantine", run_id)
        s = time.time()
        good, bad = clean(raw)
        counts["good"], counts["quarantined"] = good.count(), bad.count()
        bad.write.mode("overwrite").partitionBy("quarantine_reason").parquet(args.quarantine)
        stages["clean_quarantine"] = round(time.time() - s, 1)

        log.info("run_id=%s stage=transform", run_id)
        s = time.time()
        report = transform(good, zones)
        out_rules, agg = validate_output(report, counts)
        rules += out_rules
        counts["buckets"] = agg["buckets"]
        stages["transform_validate"] = round(time.time() - s, 1)
        failed = gate(out_rules)
        if failed:
            raise SystemExit(2)

        log.info("run_id=%s stage=write path=%s", run_id, args.out)
        s = time.time()
        (report.repartition("pickup_date").write.mode("overwrite")
               .partitionBy("pickup_date").parquet(args.out))
        stages["write"] = round(time.time() - s, 1)

        back = spark.read.parquet(args.out).agg(
            F.count(F.lit(1)).alias("buckets"), F.sum("trips").alias("trips"),
            F.round(F.sum("revenue"), 2).alias("revenue")).collect()[0].asDict()
        summary = {"run_id": run_id, "status": "published", "counts": counts,
                   "read_back": back, "stage_seconds": stages,
                   "seconds": round(time.time() - t0, 1),
                   "rules_passed": sum(r["passed"] for r in rules), "rules": len(rules)}
    except SystemExit:
        exit_code = 2
        summary = {"run_id": run_id, "status": "aborted", "counts": counts,
                   "stage_seconds": stages, "seconds": round(time.time() - t0, 1),
                   "failed_critical": [r["rule"] for r in rules
                                       if not r["passed"] and r["level"] == "critical"],
                   "rules_passed": sum(r["passed"] for r in rules), "rules": len(rules)}
    finally:
        spark.stop()

    log.info("RUN_SUMMARY %s", json.dumps(summary))
    log.info("exiting with code %d", exit_code)
    sys.exit(exit_code)


if __name__ == "__main__":
    main()
'''
Path("c2m7l5_guarded_etl.py").write_text(SCRIPT)
print("wrote c2m7l5_guarded_etl.py:", len(SCRIPT.strip().splitlines()), "lines")

wrote c2m7l5_guarded_etl.py: 209 lines

Three details are worth pausing on. The try/except SystemExit/finally structure means the run summary is emitted on both paths — a failed run logs just as carefully as a successful one, which is precisely when you need the log most. The stages dict accumulates per-stage durations as the job goes, so an aborted run’s summary shows how far it got. And spark.stop() lives in the finally, so the session is released whether the job published or died.

Path A: The Good Run, as a Real Subprocess#

Time to run it for real. We invoke the script with subprocess.run — a genuinely separate process with its own Spark session — and read back the actual returncode, the same integer a scheduler would see. The helper filters the child’s stdout down to our logger’s lines (JVM startup chatter is noise here).

We also build the fault we’ll need in a moment: a truncated March file, 100,000 rows instead of 3,582,628, standing in for an export that died a few seconds in. Real feeds fail this way constantly — a network blip mid-upload leaves a file that is perfectly valid Parquet and completely wrong.

TRUNC = "c2m7l5_march_truncated.parquet"
(spark.read.schema(taxi_schema()).parquet("yellow_tripdata_2024-03.parquet")
 .limit(100_000).coalesce(1).write.mode("overwrite").parquet(TRUNC))
print("March arrived with", f"{spark.read.parquet(TRUNC).count():,}", "rows instead of 3,582,628")

def run_job(files):
    t0 = time.time()
    p = subprocess.run([sys.executable, "c2m7l5_guarded_etl.py", "--files", *files,
                        "--out", "c2m7l5_zone_hour", "--quarantine", "c2m7l5_quarantine"],
                       capture_output=True, text=True)
    log_lines = [ln for ln in p.stdout.splitlines() if re.match(r"^\d\d:\d\d:\d\d ", ln)]
    return p.returncode, log_lines, round(time.time() - t0, 1)

code_good, log_good, secs_good = run_job(FILES)
print("\n".join(log_good))
print("\nEXIT CODE:", code_good, f"  (wall clock {secs_good}s)")

March arrived with 100,000 rows instead of 3,582,628
14:33:29 INFO run_id=6296813c stage=validate_input files=3
14:33:31 WARNING RULE FAILED [warn] input.distance_range observed=312722.3 expected=<= 200 miles
14:33:31 INFO run_id=6296813c stage=clean_quarantine
14:33:34 INFO run_id=6296813c stage=transform
14:33:37 INFO run_id=6296813c stage=write path=c2m7l5_zone_hour
14:33:42 INFO RUN_SUMMARY {"run_id": "6296813c", "status": "published", "counts": {"input": 9554778, "good": 9554757, "quarantined": 21, "buckets": 240917}, "read_back": {"buckets": 240917, "trips": 9554757, "revenue": 256692373.14}, "stage_seconds": {"validate_input": 2.3, "clean_quarantine": 2.5, "transform_validate": 3.3, "write": 3.8}, "seconds": 22.5, "rules_passed": 8, "rules": 9}
14:33:42 INFO exiting with code 0
14:33:42 INFO Closing down clientserver connection

EXIT CODE: 0   (wall clock 24.2s)

That is a complete unattended run in 22.5 seconds, and every question you’d ask the next morning is answered by those seven lines. The job walked all four stages in order. It logged one warn-level rule failure — the 312,722-mile trip again — and did not stop for it. Then it emitted the run summary as a single JSON line: input 9,554,778, good 9,554,757, quarantined 21, 240,917 buckets, read back from disk as 9,554,757 trips and $256,692,373.14, with per-stage seconds (validate 2.3, quarantine 2.5, transform 3.3, write 3.8) and 8 of 9 rules passing. Finally, exiting with code 0, and Python confirms p.returncode == 0.

The run_id (6296813c here) is a fresh random hex string on every run — yours will differ, and that’s the point. It is the key you grep by when someone asks what happened at 02:14 last Thursday. And because the summary is JSON rather than prose, a monitoring system can parse status, counts.quarantined, and read_back.revenue straight out of the log stream without anyone writing a regular expression.

Path B: The Bad Run, and the Table That Survived It#

Now the run that matters more. We hand the same script January, February, and the truncated March — a half-failed export — and then check something no amount of prose can substitute for: is the table published thirty seconds ago still exactly as it was? We fingerprint all 91 partition files (path, modification time, size) before the bad run and compare afterward.

pattern = "c2m7l5_zone_hour/pickup_date=*/*.parquet"
before = {f: (os.path.getmtime(f), os.path.getsize(f)) for f in glob.glob(pattern)}
code_bad, log_bad, secs_bad = run_job([FILES[0], FILES[1], TRUNC])
print("\n".join(log_bad))
print("\nEXIT CODE:", code_bad, f"  (wall clock {secs_bad}s)")
after = {f: (os.path.getmtime(f), os.path.getsize(f)) for f in glob.glob(pattern)}
print("\npartition files before / after the failed run:", len(before), "/", len(after))
print("published table byte-identical:", before == after)
back = spark.read.parquet("c2m7l5_zone_hour")
a = back.agg(F.count(F.lit(1)).alias("buckets"), F.sum("trips").alias("trips"),
             F.round(F.sum("revenue"), 2).alias("revenue")).collect()[0]
print(f"still reads: {a['buckets']:,} buckets | {a['trips']:,} trips | ${a['revenue']:,.2f}")
for r in (back.groupBy("PULocationID", "Zone").agg(F.sum("trips").alias("trips"))
          .orderBy(F.desc("trips")).limit(5).collect()):
    print(f"  {r['PULocationID']:>3}  {r['Zone']:<24}{r['trips']:>9,}")
q = spark.read.parquet("c2m7l5_quarantine")
print("quarantine still holds:", q.count(), "rows")

for p_ in ["c2m7l5_zone_hour", "c2m7l5_quarantine", TRUNC,
           "c2m7l5_guarded_etl.py", "spark-warehouse"]:
    shutil.rmtree(p_) if os.path.isdir(p_) else os.path.isfile(p_) and os.remove(p_)
print("temporary outputs removed")
spark.stop()

14:33:55 INFO run_id=e9a5eb5f stage=validate_input files=3
14:33:57 WARNING RULE FAILED [critical] input.volume observed=6072150 expected=>= 9,000,000
14:33:57 WARNING RULE FAILED [warn] input.distance_range observed=312722.3 expected=<= 200 miles
14:33:58 INFO RUN_SUMMARY {"run_id": "e9a5eb5f", "status": "aborted", "counts": {"input": 6072150}, "stage_seconds": {"validate_input": 2.4}, "seconds": 13.8, "failed_critical": ["input.volume"], "rules_passed": 3, "rules": 5}
14:33:58 INFO exiting with code 2
14:33:58 INFO Closing down clientserver connection

EXIT CODE: 2   (wall clock 15.1s)

partition files before / after the failed run: 91 / 91
published table byte-identical: True
still reads: 240,917 buckets | 9,554,757 trips | $256,692,373.14
  161  Midtown Center            453,825
  237  Upper East Side South     439,138
  132  JFK Airport               429,745
  236  Upper East Side North     416,508
  162  Midtown East              336,460
quarantine still holds: 21 rows
temporary outputs removed

Read the log line by line, because every line is doing work.

The job started, validated its input, and found 6,072,150 rows against an expected minimum of 9,000,000 — the truncated March file is short by 3,482,628 trips. That rule is classified critical, so the gate aborted at stage 1. Look at what is missing from the log: there is no stage=clean_quarantine, no stage=transform, no stage=write. The run summary confirms it — "status": "aborted", only validate_input in stage_seconds, only 5 rules evaluated instead of 9 (the four output rules never ran, because there was no output), and failed_critical: ["input.volume"] naming the culprit. The whole thing took 13.8 seconds and touched nothing.

Then exiting with code 2, and the calling process reads exit code 2. That integer is the entire interface between this job and a scheduler. An orchestrator does not read your log prose; it reads that number, and a nonzero value is what turns a silent 2 a.m. failure into a page, a retry, and a downstream job that doesn’t run on stale data. A job that catches its own exceptions and exits 0 is actively worse than one that crashes, because it has lied to the only thing that was listening.

And the survival proof: 91 partition files before, 91 after, and the two fingerprint dictionaries compare equal — same paths, same sizes, same modification times, to the last file. The published table still reads back to 240,917 buckets, 9,554,757 trips, and $256,692,373.14, with the podium unchanged (Midtown Center 453,825, Upper East Side South 439,138, JFK 429,745, Upper East Side North 416,508, Midtown East 336,460). The quarantine still holds its 21 rows. Yesterday’s correct answer stayed on the dashboards while the broken run failed loudly beside it. That outcome exists for exactly one reason: the gate was placed before the write. Move those four lines below the write, and this same failure would have overwritten a quarter of correct data with two-thirds of a quarter of it, and exited 2 to tell you so afterward — too late.

Practice Exercises#

Exercise 1 — Break the transform, not the input, and catch it. The bad run above failed an input rule. Prove the output rules earn their keep too: change transform to join on the zone name instead of LocationID (hourly.join(F.broadcast(zones), hourly.Zone == zones.Zone) after adding a Zone column, or simply switch the inner join to a "left" join on a deliberately wrong key), run the job, and confirm recon.trips_equal_cleaned fires as critical and aborts before the write. Report the observed trips count against the expected 9,554,757, and note that the input rules all passed — this failure was entirely your own code’s fault, which is exactly the class of bug input validation cannot catch.

Exercise 2 — Add a rule with a threshold you have to justify. The current suite has no rule about revenue per trip. Add input.fare_plausible computing the mean total_amount over the input, and decide a band. Run it against the real quarter to see the observed value, then argue for the band in a comment: how tight can you make it before a legitimate quarter (a fare hike, a snowy month with more short trips) would trip it? Classify it warn or critical and defend the choice in one sentence.

Exercise 3 — Make the run summaries queryable. A single JSON line per run is only useful if you can read many of them together. Change main() to also append its summary dict as one JSON line to a run_history.jsonl file, then run the job three or four times with different inputs (the full quarter, the truncated one, a single month). Load the file back with json.loads per line into a small table and answer: how many runs aborted, which rules fired most often, and how the validate_input stage duration varied. That table is the beginning of a pipeline-health dashboard.

Summary#

You wrapped Module 6’s quarterly ETL in every guard this module taught and proved both of its paths as real subprocesses. The guarded job runs nine rules: five over the input (volume, referential, nullability, window, distance range) evaluated in one 2.1-second pass over the quarter, and four reconciliation rules over the output. It quarantines rather than drops, holding back the 21 out-of-window strays with a pickup_outside_window reason and reconciling 9,554,757 + 21 = 9,554,778. It validates its own transform with an equality that cannot be fudged — 9,554,757 aggregated trips against 9,554,757 cleaned rows — and only then consults the gate. On the good path it published 240,917 buckets, 9,554,757 trips, and $256,692,373.14 in 22.5 seconds, logged a single JSON run summary with per-stage durations, passed 8 of 9 rules with one honest warn (a 312,722-mile trip), and returned exit code 0. Handed a truncated March export of 100,000 rows, the same script observed 6,072,150 input rows against a 9-million minimum, classified it critical, aborted at stage 1 in 13.8 seconds, and returned exit code 2 — with no clean stage, no transform, and no write in the log. All 91 of 91 partition files of the previously published table were byte-identical afterward, still reading $256,692,373.14 with Midtown Center on top at 453,825 trips.

Key Concepts#

• Validate input before you trust it — five rules in a single aggregate pass caught a truncated feed in 2.4 seconds; folding the counts into one agg means the whole suite costs one scan instead of five.
• Quarantine with a reason, never drop — the 21 held-back rows carry pickup_outside_window and reconcile exactly to the input, so a reviewer can verify nothing vanished and an analyst can re-ingest them after a fix.
• Reconciliation is the strongest output rule — aggregated trips == cleaned rows is arithmetic that only holds if the join matched every row, catching a class of bug that input validation structurally cannot see.
• The gate goes before the write — ordering, not cleverness, is what kept 91 published files byte-identical through a failed run; a gate after the write detects the disaster instead of preventing it.
• The exit code is the interface — a scheduler reads an integer, not your prose, so sys.exit(2) on a data-contract failure is what separates a loud failure from a silent one that publishes wrong numbers and reports success.

Why This Matters#

Every job CityFlow schedules will eventually be handed data it was not designed for: a file that uploaded halfway, a source that renamed a column, a month that arrives twice or not at all. The job you built here does not need a human awake to handle that. It states its expectations as executable rules, checks them against what actually arrived, keeps the rows it can’t process instead of erasing the evidence, proves its own arithmetic on the way out, refuses to publish when the contract is broken, writes down what it did in a form a machine can read, and tells the scheduler the truth with an integer. None of those guards is difficult — the whole suite is a few dozen lines — but their ordering is load-bearing, and the difference between having them and not is the difference between a table your organization can act on and a table nobody quite trusts. The 91 untouched files after that failed run are what trust looks like when you can actually measure it.

Continue Building Your Skills#

You now have a job that is correct, idempotent, and guarded — one that publishes when the data is good and refuses when it isn’t. There is exactly one thing left, and it’s the thing that turns a script into a pipeline: something has to run it. Right now you are the scheduler. You choose the input files, you type the command, you read the exit code. Module 8, the course capstone, closes that gap: it takes this guarded_etl.py and orchestrates it — dependencies between jobs, a schedule that fires without you, retries that know the difference between exit 1 and exit 2, backfills that reprocess a single day using the dynamic partition overwrite you built in Module 6, and the alerting that makes a nonzero exit reach a human instead of a log file nobody reads. Everything you have built across these seven modules — the schema contract, the execution-plan literacy, the shuffle and join intuition, the caching decisions, the ETL structure, and now the quality gates — comes together into one operated pipeline. The exit code you just proved is the wire that connects them.