Capstone Project - CityFlow's Production Spark Pipeline

Welcome to the Capstone#

Go back to Module 1 and remember what you could not do. You had a laptop and a hard, honest measurement: at this scale, Spark was not faster than the tuned streaming pandas pipeline the previous course had already built. Nothing in that measurement has changed, and this lesson will not pretend otherwise. What has changed is everything around it. In Module 1 you could run a Spark job; you could not tell why it was slow, whether its plan was reasonable, where its shuffles were, whether a retry would double your numbers, or what it would do at 2 a.m. on the night a source file uploaded halfway. Seven modules later, every one of those is a question you can answer by reading output — and this project is where you answer all of them at once, about one job, with real numbers.

That job is CityFlow’s production zone-hour pipeline: a single file a scheduler could invoke tonight. It configures and sizes its own session, declares a nineteen-field schema as a contract rather than inferring one, validates the quarter it was handed before trusting a single row of it, cleans by quarantining violations with reasons instead of deleting them, transforms with a broadcast join and not one UDF, validates its own arithmetic on the way out, refuses to publish if a critical rule fails, writes partitioned Parquet idempotently so a retry replaces partitions rather than doubling them, and finishes by emitting a machine-readable run summary and an exit code. Nine stages, and every one of them traces back to a specific lesson you have already done.

Then you will prove it three ways, because a pipeline you merely believe in is not a pipeline you can leave alone. Correctness: the published table reads back to 240,917 buckets, 9,554,757 trips, and $256,692,373.14 — the numbers Course 1 derived with pandas and multiprocessing, matched to the cent here by a distributed engine that shares no code with it. Idempotency: run it twice, and nothing moves. Guarding: point it at a deliberately broken feed and watch it abort before the write, exit nonzero, and leave the last good table byte-for-byte untouched. All three runs happen as genuine subprocesses so the exit codes are the real integers a scheduler would read. Let’s build it.

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

• Assemble every technique in this course — schema contract, validation, quarantine, broadcast join, tuned partitions, idempotent write, structured logging, exit codes — into one reusable production Spark job
• Cross-verify a distributed result against an independently derived ground truth and treat any mismatch as a failure rather than a rounding detail
• Prove idempotency by running the same job twice and demonstrating that dynamic partition overwrite replaces rather than appends
• Prove that a guarded job aborts before the write on a broken feed, returns a nonzero exit code, and leaves the previously published table intact
• Read your own finished job’s execution plan and recognize the pruning, pushdown, broadcast, single exchange, and AQE coalesce that Catalyst applied

You’ll need pyspark with a Java runtime; everything else is Python’s standard library.

The Acceptance Criteria#

Before writing a job, write down what “done” means, in numbers a reviewer can check without reading your code. CityFlow’s pipeline has to satisfy six things:

1. It publishes the right table. Reading the output back must give exactly 240,917 zone-hour buckets, 9,554,757 trips, and $256,692,373.14 — Course 1’s verified ground truth — with the podium unchanged: 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.
2. Nothing vanishes. The 21 out-of-window strays are quarantined with a reason, and 9,554,757 + 21 = 9,554,778 reconciles against the input.
3. A retry is safe. Running the identical job twice changes no total and creates no duplicate partitions.
4. A bad feed cannot publish. A truncated input trips a critical rule, aborts before the write, and returns a nonzero exit code while the previously published table stays byte-identical.
5. It logs like a job, not a notebook. One machine-readable summary line per run, on both the success and failure paths, plus per-stage durations.
6. Its plan is defensible. Pruned read schema, pushed filters, a broadcast join with no shuffle on the large side, exactly one exchange for the group-by, and AQE reshaping the post-shuffle partitions.

Here is the session the whole project shares. Every configuration on it is a decision from an earlier module rather than a default: local[*] from Module 1’s driver-and-executors model, spark.sql.shuffle.partitions sized deliberately from Module 5 (200 is absurd for ten cores and 240,917 output rows), adaptive execution left on so it can reshape what we asked for, and dynamic partition overwrite from Module 6 Lesson 4, which is what makes stage 8 idempotent.

import warnings; warnings.filterwarnings("ignore")
import datetime as dt, glob, hashlib, json, os, re, shutil, subprocess, sys, time
from pathlib import Path
from pyspark.sql import SparkSession, functions as F

FILES = [f"yellow_tripdata_2024-{m}.parquet" for m in ("01", "02", "03")]
OUT, QUAR = "c2m8_zone_hour", "c2m8_quarantine"

spark = (SparkSession.builder
         .appName("cityflow-capstone")
         .master("local[*]")
         .config("spark.sql.shuffle.partitions", 16)
         .config("spark.sql.adaptive.enabled", "true")
         .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
         .config("spark.ui.showConsoleProgress", "false")
         .getOrCreate())
spark.sparkContext.setLogLevel("ERROR")

print("cores (defaultParallelism) :", spark.sparkContext.defaultParallelism)
print("spark.sql.shuffle.partitions:", spark.conf.get("spark.sql.shuffle.partitions"))
print("adaptive query execution   :", spark.conf.get("spark.sql.adaptive.enabled"))
print("partition overwrite mode   :", spark.conf.get("spark.sql.sources.partitionOverwriteMode"))
print("input                      :",
      f"{sum(os.path.getsize(f) for f in FILES)/1e6:.1f} MB across {len(FILES)} monthly files")

cores (defaultParallelism) : 10
spark.sql.shuffle.partitions: 16
adaptive query execution   : true
partition overwrite mode   : dynamic
input                      : 160.4 MB across 3 monthly files

Ten cores, sixteen shuffle partitions, AQE on, dynamic overwrite armed, and 160.4 MB of Parquet holding a quarter of New York’s yellow-taxi trips. That is the entire operating environment, stated in five lines that a reviewer can argue with.

The Deliverable: cityflow_pipeline.py#

Everything the course taught, in one file. Read it as nine stages, because that is how it is written — and notice that the interesting engineering is almost entirely in the ordering, not the cleverness of any single function.

Two details deserve flagging before you read it. First, there are no UDFs anywhere, deliberately: Module 4 Lesson 4 measured a Python UDF running 7.1x slower than the equivalent built-in because Catalyst cannot see inside it, and this entire transform is expressible with pyspark.sql.functions. Second, clean() projects the good rows down to the three columns the report actually needs before anything caches them. That is not cosmetic. Caching the full nineteen-column frame at this scale throws java.lang.OutOfMemoryError: Java heap space in the executor’s Parquet reader — the exact failure Module 5 Lesson 3 measured, where the naive cache of a wide frame blew the heap and a narrow projection fit comfortably. The projection is what makes the cache on line good.persist() safe.

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

SCRIPT = r'''"""cityflow_pipeline.py - CityFlow's production zone-hour Spark pipeline.

Nine stages: configure, schema, validate input, clean+quarantine, transform,
validate output, gate, write idempotently, summarize.

Exit codes
    0  published
    2  aborted by a critical data-quality rule (nothing written)
    1  unhandled error (the code broke, not the data)
"""
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)

# --- thresholds: the definition of "normal", argued once in daylight ---------
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
MIN_BUCKETS = 100_000
SHUFFLE_PARTITIONS = 16

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


# --- M6 L1: the schema is a contract, not a guess ---------------------------
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()),
    ])


# --- M1 + M5: one session, sized on purpose ---------------------------------
def build_session(app="cityflow-pipeline"):
    spark = (SparkSession.builder.appName(app)
             .master("local[*]")
             .config("spark.sql.shuffle.partitions", SHUFFLE_PARTITIONS)
             .config("spark.sql.adaptive.enabled", "true")
             .config("spark.sql.sources.partitionOverwriteMode", "dynamic")
             .config("spark.ui.showConsoleProgress", "false")
             .getOrCreate())
    spark.sparkContext.setLogLevel("ERROR")
    return spark


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


def gate(rules):
    """Log every failure; return only the CRITICAL ones. Warns never stop a run."""
    for r in rules:
        if not r["passed"]:
            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 out_of_window():
    return ((F.col("tpep_pickup_datetime") < F.lit(LO)) |
            (F.col("tpep_pickup_datetime") >= F.lit(HI)))


# --- M7 L1: validate the input, one pass ------------------------------------
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"]
    rules = [
        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"),
    ]
    return rules, m


# --- M6 L2 + M7 L2: clean by quarantining -----------------------------------
def clean(raw):
    """Good rows are projected to the three columns the report needs (M2/M3 pruning),
    so the cache in main() holds ~20 bytes/row instead of all 19 columns."""
    keyed = ~out_of_window() & F.col("PULocationID").isNotNull()
    good = raw.filter(keyed).select(
        "PULocationID", "total_amount",
        F.date_trunc("hour", "tpep_pickup_datetime").alias("pickup_hour"))
    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


# --- M3 + M4: the transform. Built-in functions only; no UDFs ---------------
def transform(good, zones):
    joined = good.join(F.broadcast(zones), good.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")))


# --- M7 L1/L5: validate what you produced -----------------------------------
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()
    rules = [
        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_plausible", "critical", agg["buckets"] >= MIN_BUCKETS,
             agg["buckets"], f">= {MIN_BUCKETS:,}"),
        rule("output.revenue_range", "critical", REV_LO <= agg["revenue"] <= REV_HI,
             agg["revenue"], f"{REV_LO:,.0f}..{REV_HI:,.0f}"),
    ]
    return rules, agg


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--files", nargs="+", required=True)
    ap.add_argument("--zones", default="taxi_zone_lookup.csv")
    ap.add_argument("--out", required=True)
    ap.add_argument("--quarantine", required=True)
    ap.add_argument("--explain", action="store_true")
    args = ap.parse_args()

    run_id = uuid.uuid4().hex[:8]
    t0 = time.time()
    stages, counts, rules = {}, {}, []
    spark = build_session()
    exit_code = 0
    try:
        # 1-2. session + schema contract
        log.info("run_id=%s stage=start cores=%d shuffle_partitions=%s",
                 run_id, spark.sparkContext.defaultParallelism,
                 spark.conf.get("spark.sql.shuffle.partitions"))
        zones = (spark.read.option("header", True).csv(args.zones)
                 .select(F.col("LocationID").cast("int").alias("LocationID"), "Zone"))
        raw = spark.read.schema(taxi_schema()).parquet(*args.files)

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

        # 4. clean + quarantine + reconcile
        log.info("run_id=%s stage=clean_quarantine", run_id)
        s = time.time()
        good, bad = clean(raw)
        good = good.persist()
        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)

        # 5-6. transform + validate output
        log.info("run_id=%s stage=transform", run_id)
        s = time.time()
        report = transform(good, zones)
        if args.explain:
            report.explain(mode="formatted")
        out_rules, agg = validate_output(report, counts)
        rules += out_rules
        counts["buckets"] = agg["buckets"]
        stages["transform_validate"] = round(time.time() - s, 1)

        # 7. GATE, before the write
        if gate(out_rules):
            raise SystemExit(2)

        # 8. idempotent partitioned write
        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)
        good.unpersist()

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

    # 9. structured run summary + exit code
    log.info("RUN_SUMMARY %s", json.dumps(summary))
    log.info("exit=%d", exit_code)
    sys.exit(exit_code)


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

wrote c2m8_cityflow_pipeline.py: 244 lines

Two hundred and forty-four lines, and every block in it has a lesson behind it.

Stage 1, the session (Module 1 Lesson 2 and Module 5 Lesson 1): local[*] gives ten in-process executors — the same engine a cluster runs, just with the network removed. spark.sql.shuffle.partitions=16 replaces the default 200, which Module 5 showed produces mostly-empty tasks and pure overhead at this scale.

Stage 2, the schema (Module 6 Lesson 1): nineteen fields typed explicitly. Not for speed — although skipping the inference pass is free — but because a schema is a contract. If next month’s export renames Airport_fee or ships total_amount as a string, the job fails at read time with a clear complaint instead of silently producing nulls that a SUM will happily add to zero.

Stage 3, input validation (Module 7 Lesson 1): five rules, four of whose metrics come from one aggregate pass — a naive suite calling .count() per rule would scan the quarter five times. Severity is a human judgment recorded in code: volume, referential integrity, and null pickup zones are critical; the window and distance rules are warn.

Stage 4, clean and quarantine (Module 6 Lesson 2 and Module 7 Lesson 2): the split is by a single predicate, and the rejected rows are written, with a quarantine_reason, not deleted. Note that good keeps only PULocationID, total_amount, and the truncated pickup_hour — three columns of nineteen, which is both Module 3’s column pruning and what makes the subsequent persist() fit in memory.

Stage 5, the transform (Modules 3 and 4): date_trunc for the hour key, F.broadcast on the 265-row zone dimension, and a groupBy on (PULocationID, Zone, pickup_hour). Joining on LocationID and never on the zone name is the data-quality decision Course 1 made before it read a single trip, because two distinct Queens zones are both named “Corona.”

Stage 6, output validation (Module 7 Lesson 1): reconciliation rules, which are the strongest kind — aggregated trips == cleaned rows is arithmetic that only holds if the join matched every row. Input validation structurally cannot catch a bug in your own transform; this can.

Stage 7, the gate (Module 7 Lesson 4): three lines, deliberately stupid. Any critical failure aborts. All the intelligence lives in how each rule was classified, where a code review can see it.

Stage 8, the write (Module 6 Lessons 3 and 4): partitioned Parquet by pickup_date, mode("overwrite") combined with partitionOverwriteMode=dynamic, so a re-run replaces the partitions it produces and leaves everything else alone. That is the difference between a retry that is safe and one that doubles the books.

Stage 9, the summary (Module 7 Lesson 3 and 4): one JSON line and one integer. Note the try/except SystemExit/finally shape — the summary is emitted on both paths, and spark.stop() runs either way.

Proof 1: Correctness#

Time to run it for real. subprocess.run invokes the script as a genuinely separate process with its own Spark session, and hands back the actual returncode — the same integer a scheduler reads. The helper filters the child’s stdout down to our logger’s lines, because JVM startup chatter is noise here.

def run_pipeline(files, out=OUT, quarantine=QUAR):
    """Invoke the pipeline as a real subprocess and return its REAL exit code."""
    t0 = time.time()
    p = subprocess.run([sys.executable, "c2m8_cityflow_pipeline.py",
                        "--files", *files, "--out", out, "--quarantine", quarantine],
                       capture_output=True, text=True)
    lines = [ln for ln in p.stdout.splitlines() if re.match(r"^\d\d:\d\d:\d\d ", ln)]
    return p.returncode, lines, round(time.time() - t0, 1)

code_1, log_1, wall_1 = run_pipeline(FILES)
print("\n".join(log_1))
print("\nEXIT CODE:", code_1, f"  (wall clock {wall_1}s)")

18:42:41 INFO run_id=05348908 stage=start cores=10 shuffle_partitions=16
18:42:43 INFO run_id=05348908 stage=validate_input files=3
18:42:46 WARNING RULE FAILED [warn] input.distance_range observed=312722.3 expected=<= 200 miles
18:42:46 INFO run_id=05348908 stage=clean_quarantine
18:42:49 INFO run_id=05348908 stage=transform
18:42:51 INFO run_id=05348908 stage=write path=c2m8_zone_hour
18:42:54 INFO RUN_SUMMARY {"run_id": "05348908", "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.1, "clean_quarantine": 3.6, "transform_validate": 2.0, "write": 2.0}, "seconds": 16.7, "rules_passed": 8, "rules": 9}
18:42:54 INFO exit=0

EXIT CODE: 0   (wall clock 18.0s)

A complete unattended run in 16.7 seconds, and every question you would ask the next morning is answered by those eight lines. All four stages ran in order. One warn-level rule failed — a trip_distance of 312,722.3 miles, roughly to the moon and back, the course’s canonical example of a rule that actually fires — and the job correctly did not stop for it. The summary reports input 9,554,778, good 9,554,757, quarantined 21, 240,917 buckets, per-stage seconds, 8 of 9 rules passing, and exit=0. The run_id is a fresh hex string each run; yours will differ, and that is the point — it is what you grep by when someone asks what happened at 02:14 last Thursday.

Now the check that licenses everything else. The job’s own read-back is a good habit, but it is the job checking its own homework. We open the published table independently and compare it against numbers that were produced in a different course by a different engine — pandas, multiprocessing, and SQLite, sharing not one line of code with Spark.

published = spark.read.parquet(OUT)
a = published.agg(F.count(F.lit(1)).alias("buckets"),
                  F.sum("trips").alias("trips"),
                  F.round(F.sum("revenue"), 2).alias("revenue")).collect()[0]
quarantined = spark.read.parquet(QUAR).count()

print(f"buckets  : {a['buckets']:,}   (Course 1: 240,917)      match:", a["buckets"] == 240_917)
print(f"trips    : {a['trips']:,}   (Course 1: 9,554,757)    match:", a["trips"] == 9_554_757)
print(f"revenue  : ${a['revenue']:,.2f}   (Course 1: $256,692,373.14)  match:",
      a["revenue"] == 256_692_373.14)
print(f"quarantined: {quarantined}")
print(f"reconciles : {a['trips']:,} + {quarantined} = {a['trips'] + quarantined:,}",
      "==", f"{9_554_778:,}", "->", a["trips"] + quarantined == 9_554_778)
print("\nbusiest pickup zones over the quarter:")
for r in (published.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,}")

buckets  : 240,917   (Course 1: 240,917)      match: True
trips    : 9,554,757   (Course 1: 9,554,757)    match: True
revenue  : $256,692,373.14   (Course 1: $256,692,373.14)  match: True
quarantined: 21
reconciles : 9,554,757 + 21 = 9,554,778 == 9,554,778 -> True

busiest pickup zones over the quarter:
  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

Three matches out of three, and the podium is identical. This is the strongest verification in the entire path, and it is worth being precise about why. The floating-point sum of 9.5 million doubles is order-dependent — Spark’s ten executors add their partitions in a different order than pandas did, so the raw sums differ in their last bits. Rounded to the cent, they agree exactly: $256,692,373.14. Two independent implementations, two entirely different execution models, one answer. And the reconciliation identity closes the loop from the other side: 9,554,757 + 21 = 9,554,778, so every row of the input is accounted for as either published or quarantined. Nothing was silently dropped to make the numbers pretty.

The quarantine is not a wastebasket. Those 21 rows are the strays stamped 2002-12-31, a decade before the quarter began, and they sit in their own partitioned Parquet directory with quarantine_reason=pickup_outside_window. An analyst can look at them next week, conclude they were a clock fault at one vendor, and re-ingest them with corrected timestamps. A filter() that had thrown them away would have left nothing to look at and no way to notice they existed.

Proof 2: Idempotency#

Pipelines get retried. A network hiccup, a scheduler restart, a human running the same backfill twice — every one of those means your job runs again on data it has already processed, and the only acceptable outcome is nothing changes. Module 6 Lesson 4 built this guarantee out of two pieces: partitioning the output by pickup_date, and setting spark.sql.sources.partitionOverwriteMode=dynamic so mode("overwrite") replaces only the partitions this run produced instead of nuking the whole directory or appending beside it.

So we run the identical command a second time and compare everything.

files_before = sorted(glob.glob(f"{OUT}/pickup_date=*/*.parquet"))
code_2, log_2, wall_2 = run_pipeline(FILES)
summary_2 = json.loads([ln for ln in log_2 if "RUN_SUMMARY" in ln][0].split("RUN_SUMMARY ")[1])
b = spark.read.parquet(OUT).agg(F.count(F.lit(1)).alias("buckets"),
                                F.sum("trips").alias("trips"),
                                F.round(F.sum("revenue"), 2).alias("revenue")).collect()[0]
files_after = sorted(glob.glob(f"{OUT}/pickup_date=*/*.parquet"))
print("second run exit code:", code_2, f"({wall_2}s)")
print("second run status   :", summary_2["status"], "| read_back:", summary_2["read_back"])
print(f"\n{'':<12}{'run 1':>16}{'run 2':>16}")
print(f"{'buckets':<12}{a['buckets']:>16,}{b['buckets']:>16,}")
print(f"{'trips':<12}{a['trips']:>16,}{b['trips']:>16,}")
print(f"{'revenue':<12}{a['revenue']:>16,.2f}{b['revenue']:>16,.2f}")
print(f"{'partitions':<12}{len(files_before):>16}{len(files_after):>16}")
print("\nno doubling, totals identical:",
      (a["buckets"], a["trips"], a["revenue"]) == (b["buckets"], b["trips"], b["revenue"]))

second run exit code: 0 (16.9s)
second run status   : published | read_back: {'buckets': 240917, 'trips': 9554757, 'revenue': 256692373.14}

                       run 1           run 2
buckets              240,917         240,917
trips              9,554,757       9,554,757
revenue       256,692,373.14  256,692,373.14
partitions                91              91

no doubling, totals identical: True

Nothing moved. Same 240,917 buckets, same 9,554,757 trips, same $256,692,373.14, same 91 date partitions before and after. That is what idempotency looks like when you can measure it, and it is worth understanding exactly which line bought it. Change mode("overwrite") to mode("append") and the second run would leave you with 481,834 rows, 19,109,514 trips, and half a billion dollars of revenue that never existed — and no error, anywhere, to tell you so. The totals would simply be double, and the dashboards would show a record quarter.

Reading the Finished Job’s Plan#

The job is correct and repeatable. Is it reasonable? Module 4 taught you to read .explain() and Module 5 taught you what to look for in it, and this is the moment those pay off: not on a toy query someone else wrote, but on your own production job.

One subtlety about how to get the plan. Calling .explain() on an unexecuted DataFrame prints isFinalPlan=false, because adaptive query execution has not had a chance to look at any runtime statistics yet — you see what Catalyst intends, not what Spark did. To see the real thing, execute the DataFrame first and then explain it.

sys.path.insert(0, ".")
from c2m8_cityflow_pipeline import taxi_schema, clean, transform

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)
good, bad = clean(raw)
report = transform(good, zones)

t0 = time.time()
rows = report.collect()
print(f"executed the real transform: {len(rows):,} buckets in {time.time() - t0:.1f}s\n")
report.explain(mode="simple")

executed the real transform: 240,917 buckets in 2.7s

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=true
+- == Final Plan ==
   ResultQueryStage 2
   +- *(3) HashAggregate(keys=[PULocationID#144, Zone#133, pickup_hour#157], functions=[count(1), sum(total_amount#153)])
      +- AQEShuffleRead coalesced
         +- ShuffleQueryStage 1
            +- Exchange hashpartitioning(PULocationID#144, Zone#133, pickup_hour#157, 16), ENSURE_REQUIREMENTS, [plan_id=332]
               +- *(2) HashAggregate(keys=[PULocationID#144, Zone#133, pickup_hour#157], functions=[partial_count(1), partial_sum(total_amount#153)])
                  +- *(2) Project [PULocationID#144, total_amount#153, pickup_hour#157, Zone#133]
                     +- *(2) BroadcastHashJoin [PULocationID#144], [LocationID#135], Inner, BuildRight, false, false
                        :- *(2) Project [PULocationID#144, total_amount#153, date_trunc(hour, cast(tpep_pickup_datetime#138 as timestamp), Some(Asia/Tehran)) AS pickup_hour#157]
                        :  +- *(2) Filter (((isnotnull(tpep_pickup_datetime#138) AND (tpep_pickup_datetime#138 >= 2024-01-01 00:00:00)) AND (tpep_pickup_datetime#138 < 2024-04-01 00:00:00)) AND isnotnull(PULocationID#144))
                        :     +- *(2) ColumnarToRow
                        :        +- FileScan parquet [tpep_pickup_datetime#138,PULocationID#144,total_amount#153] Batched: true, DataFilters: [...], Format: Parquet, Location: InMemoryFileIndex(3 paths)[...], PartitionFilters: [], PushedFilters: [IsNotNull(tpep_pickup_datetime), GreaterThanOrEqual(tpep_pickup_datetime,2024-01-01T00:00), Less..., ReadSchema: struct<tpep_pickup_datetime:timestamp_ntz,PULocationID:int,total_amount:double>
                        +- BroadcastQueryStage 0
                           +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, true] as bigint)),false), [plan_id=275]
                              +- *(1) Project [cast(LocationID#131 as int) AS LocationID#135, Zone#133]
                                 +- *(1) Filter isnotnull(cast(LocationID#131 as int))
                                    +- FileScan csv [LocationID#131,Zone#133] Batched: false, ..., Location: InMemoryFileIndex(1 paths)[...], ReadSchema: struct<LocationID:string,Zone:string>
+- == Initial Plan ==
   (the same tree without the query stages -- elided here for length)

(The absolute file paths in the real output are long; they are shortened to [...] above, along with the repeated == Initial Plan == subtree, which is the identical operator tree before AQE wrapped it in query stages. Everything else is verbatim.)

Read it bottom-up, and every line is something an earlier module taught you to want.

ReadSchema: struct<tpep_pickup_datetime, PULocationID, total_amount> — three columns, not nineteen. Catalyst pushed the projection all the way into the Parquet reader, so the other sixteen columns are never decompressed or decoded. This is Module 4 Lesson 3’s column pruning, and it is why the scan is cheap despite the file being wide.

PushedFilters: [IsNotNull(tpep_pickup_datetime), GreaterThanOrEqual(...2024-01-01...), LessThan(...2024-04-01...)] — the reporting window is not applied after loading the data; it is handed to Parquet, which uses row-group statistics to skip chunks that cannot contain matching rows. The filter written in clean() as a readable Python expression became an I/O optimization without anyone asking.

BroadcastHashJoin ... BuildRight with a BroadcastQueryStage on the right and no Exchange on the left — this is the single most important line in the plan. The 265-row zone dimension is built into a hash table and shipped to every executor; the 9.5-million-row side stays exactly where it is. Module 3 Lesson 3 measured what the alternative costs: without the broadcast hint, a sort-merge join shuffles both sides across the network. Here the big side never moves at all.

Exactly one Exchange, hashpartitioning(..., 16), sitting between a partial_count/partial_sum HashAggregate and the final one. That is the whole shuffle budget of this job: one. The partial aggregate is Spark’s map-side combine — each of the ten input partitions reduces its millions of rows to its own zone-hour buckets before anything crosses the shuffle boundary, so what travels is aggregates rather than trips. Module 5 Lesson 2’s entire point, visible in two adjacent operators.

AQEShuffleRead coalesced — adaptive execution looked at the actual post-shuffle sizes and decided sixteen partitions was more than this data needed. We can measure exactly what it chose:

landed = report.select(F.spark_partition_id().alias("p")).distinct().count()
print("configured spark.sql.shuffle.partitions:", spark.conf.get("spark.sql.shuffle.partitions"))
print("partitions the report actually landed in:", landed, "  <- AQE coalesced them")

configured spark.sql.shuffle.partitions: 16
partitions the report actually landed in: 5   <- AQE coalesced them

Sixteen requested, five delivered. The 240,917 output rows are small enough that sixteen tasks would each do a trivial amount of work while paying full task-scheduling overhead, so AQE merged them into five. This is the honest reading of a setting: spark.sql.shuffle.partitions is an upper bound and a starting point, and with AQE on, Spark corrects downward using statistics you did not have when you wrote the config. It cannot correct upward, which is why leaving the default at 200 still hurts — you would get 200 tasks planned, coalesced, and a pile of scheduling you never needed.

One last thing to notice, because it looks alarming and is not: date_trunc(hour, ..., Some(Asia/Tehran)). Spark records the session time zone in the plan node. Our timestamps are TimestampNTZType — naive, no zone — so no conversion is applied and the truncation is purely arithmetic on the stored value. The zone name is metadata about the session, not a transformation of your data, and the totals prove it: they match a pandas pipeline that had no concept of a session time zone at all.

Proof 3: Guarding#

The final proof is the one that decides whether you can go to sleep. Everything so far assumed the data arriving tonight looks like the data that arrived last night. It will not, eventually. The most common way a batch feed fails is not a crash — it is a file that uploaded halfway, leaving behind perfectly valid Parquet containing a third of a month. Nothing about it is corrupt. It is simply short, and a job without input validation will process it cheerfully, overwrite the published table with a fragment of a quarter, and exit 0.

We build exactly that fault: March truncated to 100,000 rows instead of 3,582,628. Then, before running the broken feed, we take an md5 fingerprint of every published Parquet file — because the claim “the last good table survived” deserves to be checked byte by byte rather than asserted.

TRUNC = "c2m8_march_truncated.parquet"
(spark.read.schema(taxi_schema()).parquet(FILES[2])
 .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\n")

def fingerprint():
    return {f: hashlib.md5(Path(f).read_bytes()).hexdigest()
            for f in sorted(glob.glob(f"{OUT}/pickup_date=*/*.parquet"))}

before = fingerprint()
code_bad, log_bad, wall_bad = run_pipeline([FILES[0], FILES[1], TRUNC])
print("\n".join(log_bad))
print("\nEXIT CODE:", code_bad, f"  (wall clock {wall_bad}s)")

after = fingerprint()
print("\npartition files before / after the failed run:", len(before), "/", len(after))
print("every published file byte-identical (md5):", before == after)
c = spark.read.parquet(OUT).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"last good table still reads: {c['buckets']:,} buckets | "
      f"{c['trips']:,} trips | ${c['revenue']:,.2f}")
print("quarantine still holds:", spark.read.parquet(QUAR).count(), "rows")

March arrived with 100,000 rows instead of 3,582,628

18:43:26 INFO run_id=a8d9547b stage=start cores=10 shuffle_partitions=16
18:43:28 INFO run_id=a8d9547b stage=validate_input files=3
18:43:30 WARNING RULE FAILED [critical] input.volume observed=6072150 expected=>= 9,000,000
18:43:30 WARNING RULE FAILED [warn] input.distance_range observed=312722.3 expected=<= 200 miles
18:43:30 INFO RUN_SUMMARY {"run_id": "a8d9547b", "status": "aborted", "counts": {"input": 6072150}, "stage_seconds": {"validate_input": 1.9}, "seconds": 7.7, "failed_critical": ["input.volume"], "rules_passed": 3, "rules": 5}
18:43:30 INFO exit=2

EXIT CODE: 2   (wall clock 8.7s)

partition files before / after the failed run: 91 / 91
every published file byte-identical (md5): True
last good table still reads: 240,917 buckets | 9,554,757 trips | $256,692,373.14
quarantine still holds: 21 rows

Read that log by what is missing from it. There is no stage=clean_quarantine. No stage=transform. No stage=write. The job observed 6,072,150 input rows against a 9,000,000 minimum, the input.volume rule is classified critical, and the gate stopped it at stage 3 — 7.7 seconds in, having touched nothing. The summary confirms it from the machine’s side: "status": "aborted", only validate_input in stage_seconds, only 5 rules evaluated instead of 9 because the four output rules never ran (there was no output), and failed_critical: ["input.volume"] naming the culprit by name.

Then exit=2, and the calling process reads exit code 2. That integer is the entire interface between this job and whatever schedules it. An orchestrator does not read your 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 does not run on stale data. The choice of 2 over 1 is deliberate and documented in the script’s docstring: 1 means the code broke and should probably be retried; 2 means the data failed its contract and a human needs to look at the upstream export.

And the survival proof: 91 files before, 91 after, and every single md5 identical. Not “the totals still look right” — the actual bytes on disk are unchanged, file by file. The table still reads back to 240,917 buckets, 9,554,757 trips, and $256,692,373.14, and 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 has exactly one cause, and it is not sophistication. It is ordering. Move the gate below the write and this identical failure would have replaced a quarter of correct data with two-thirds of a quarter of it, then exited 2 to inform you afterward. Same rules, same severities, same code — and a disaster instead of a non-event, because four lines were in the wrong place.

for p_ in sorted(glob.glob("c2m8_*")) + ["spark-warehouse"]:
    if os.path.isdir(p_):
        shutil.rmtree(p_)
    elif os.path.isfile(p_):
        os.remove(p_)
shutil.rmtree("__pycache__", ignore_errors=True)
print("leftover c2m8_* artifacts:", sorted(glob.glob("c2m8_*")) or "none")
spark.stop()

leftover c2m8_* artifacts: none

The Production Checklist#

Eight things, in order. None of them is difficult; the value is entirely in doing all of them and in this sequence.

1. Pin the schema. Declare every field and its type. A renamed or retyped column should fail at read time, not turn into nulls that a SUM treats as zero.
2. Validate the input before you trust it. Roll the metrics into one aggregate pass so the whole suite costs one scan. Classify each rule warn or critical deliberately — a suite where everything always passes is a suite you should distrust.
3. Quarantine with a reason; never silently drop. Write the rejected rows somewhere a human can read them, with a column saying why.
4. Reconcile the arithmetic. good + quarantined == input and aggregated trips == cleaned rows are identities that only hold if your transform did what it claimed. They catch bugs input validation structurally cannot see.
5. Put the gate before the write. This is the ordering that makes every other guard worth having.
6. Overwrite partitions, never append. Dynamic partition overwrite is what makes a retry safe and a single-day backfill cheap.
7. Log one machine-readable line per run, on both paths. A failed run needs its summary more than a successful one does.
8. Exit with a number a scheduler believes. Never let a try/except swallow a data failure into a zero — a job that lies about succeeding is worse than one that crashes.

Practice Exercises#

Exercise 1 — Prove that only the touched partitions change. The idempotency proof showed totals unmoved, which is necessary but not sufficient — a job that rewrote all 91 partitions with identical content would also pass it. Fingerprint every partition file with md5, then re-run the pipeline on January and February only (leave March out), and report how many of the 91 files changed. Predict the answer before you run it, then explain the result in terms of what partitionOverwriteMode=dynamic actually replaces.

Exercise 2 — Break the transform, not the input, and watch the output rules earn their keep. Every failure in this lesson was an input problem. Introduce a bug in your own code instead: change transform() to join on the zone name rather than LocationID (add Zone to the projection in clean() and join good.Zone == zones.Zone). Run the pipeline and report which rule fires, what it observed, and whether the job aborted before or after the write. Note that all five input rules still passed.

Exercise 3 — Find the setting that changes the plan, not just the timing. Re-run the transform three ways and read .explain() after each: (a) with spark.sql.adaptive.enabled set to false, (b) with spark.sql.autoBroadcastJoinThreshold set to -1 (broadcasting disabled) while removing the F.broadcast hint, and (c) with spark.sql.shuffle.partitions at the default 200. For each, name the operator that appears or disappears and how many partitions the result lands in. Which of the three changes the shape of the plan, and which only change its size?

Summary#

You built CityFlow’s production Spark pipeline as a single 244-line file whose nine stages each trace to the module that taught them, and then proved it three ways with real subprocesses and real exit codes. It configures a local[*] session on 10 cores with 16 shuffle partitions, AQE on, and dynamic partition overwrite armed; declares a 19-field schema contract; runs 5 input rules in one aggregate pass; quarantines the 21 out-of-window strays with a reason instead of deleting them; transforms with date_trunc and a broadcast join over the 265-row zone dimension using zero UDFs; checks 4 reconciliation rules over its own output; gates on critical failures before the write; publishes 91 date partitions idempotently; and emits one JSON summary line and an exit code. Correctness: the published table read back independently gives exactly 240,917 buckets, 9,554,757 trips and $256,692,373.14, matching Course 1’s pandas-and-multiprocessing ground truth to the cent with the podium unchanged at Midtown Center 453,825, Upper East Side South 439,138, JFK 429,745, Upper East Side North 416,508, Midtown East 336,460 — and 9,554,757 + 21 = 9,554,778 reconciles. Idempotency: a second identical run in 16.9 s exited 0 and moved nothing — same totals, same 91 partitions. Guarding: handed a March export truncated to 100,000 rows, the job observed 6,072,150 against a 9-million minimum, aborted at stage 3 in 7.7 s with exit code 2, and left all 91 of 91 published files md5-identical and still reading $256,692,373.14. The finished plan showed 3 columns of 19 in ReadSchema, the reporting window in PushedFilters, a BroadcastHashJoin BuildRight with no exchange on the large side, exactly one Exchange, and AQE coalescing 16 post-shuffle partitions down to 5.

Key Concepts#

• A production job is its stages, in order — schema (M6), validate (M7), quarantine (M7), transform (M3/M4), reconcile (M7), gate (M7), write idempotently (M6), summarize (M7). The engineering is in the sequence; the gate sitting upstream of the write is what made a broken feed a non-event instead of a disaster.
• Cross-verification beats self-verification — the job checking its own read-back is a good habit, but matching a completely independent implementation at $256,692,373.14 to the cent is proof. Float summation is order-dependent, so the raw sums differ in their last bits and agree exactly once rounded.
• Idempotency is a write mode, not a hope — partitionOverwriteMode=dynamic with mode("overwrite") replaces the partitions a run produced and leaves the rest byte-identical; append would have silently doubled every total with no error anywhere.
• The plan is the tuning report — a pruned ReadSchema, pushed filters, BroadcastHashJoin with no big-side Exchange, one shuffle, and AQEShuffleRead coalesced are each a decision you can point at. A job whose plan you cannot read is a job you cannot tune.
• The exit code is the interface — a scheduler reads an integer, not a log. Separating “the data failed its contract” (2) from “the code broke” (1) is what lets an orchestrator retry one and page a human for the other.

Why This Matters#

Every job CityFlow schedules will eventually be handed data it was not designed for, and nobody will be awake when it happens. What you built here handles that without a human: it states its expectations as executable rules, checks them against what actually arrived, keeps the rows it cannot process instead of erasing the evidence, proves its own arithmetic before publishing, refuses to publish when the contract is broken, writes in a way that a retry cannot corrupt, records what it did in a form a machine can parse, and tells the scheduler the truth with an integer. The 91 untouched files after that failed run are what trust looks like when you can actually measure it — and the pattern transfers unchanged to any batch pipeline you will ever write, on any engine, over any data. The taxi trips are incidental. The ordering is not.

What You’ve Built#

Look back across the eight modules. In Module 1 you asked the honest question — why reach for a distributed engine on a laptop at all — and got an uncomfortable answer: at this scale, tuned streaming pandas beat Spark on every size tested. That measurement still stands, and it is the most useful thing in the course, because it means everything you learned afterward had to justify itself on something other than raw speed. Module 2 showed you that transformations build a plan and only actions execute it, and that the DAG of jobs, stages, and tasks is the thing you are really writing. Module 3 gave you the DataFrame vocabulary — columns, expressions, filters, joins, aggregations — and Module 4 revealed that SQL and the DataFrame API compile to identical plans, that Catalyst prunes and pushes down aggressively when you let it, and that a Python UDF is an opaque box it cannot optimize through. Module 5 made partitions, shuffles, and caching concrete, including the uncomfortable result that a naive cache was slower until the shuffle width was sized first. Module 6 turned all of it into an ETL shape: schema, read, clean, transform, write, idempotently. Module 7 made that job safe to run when nobody is watching. And this module assembled every piece into one file that publishes, refuses, and reports.

So what did Spark actually buy, given it was never the faster option here? Four things, and they are worth naming precisely because none of them is speed. Declarative code that an optimizer improves for you — you wrote a filter and a join, and Catalyst turned them into a pruned read schema, a Parquet-level pushdown, and a broadcast that kept 9.5 million rows from ever moving; you did not hand-tune any of it. A plan you can read — when this job eventually gets slow, you will not guess, you will run .explain() and find the extra Exchange or the SortMergeJoin that should have been a broadcast. A shape that survives outgrowing the machine — the pipeline you wrote runs unchanged on a cluster, where the same local[*] becomes fifty executors and the same broadcast join means the same thing; the pandas pipeline from Course 1 does not have that property at any price. And production guarantees — the idempotent partitioned write, the reconciliation identities, and the gate before the write are what let a job run at 2 a.m. and be believed in the morning. That is the trade: you gave up some speed at laptop scale and bought optimizability, observability, portability, and safety.

Where this goes next is the rest of the Data Engineering path, and both remaining courses take the job you just wrote as their starting point. Docker & Kubernetes for Data Engineering solves a problem this lesson quietly ignored: cityflow_pipeline.py runs correctly on your machine, with your JDK, your pyspark version, and your file paths, and none of that is true anywhere else. Containerizing it turns “works here” into “runs identically anywhere,” which is the precondition for it running somewhere other than a laptop at all. Then Airflow closes the last gap, and it is a gap you should be able to feel by now: right now you are the scheduler. You chose the input files, you typed the command, you read the exit code and decided what it meant. Airflow orchestrates exactly this job — a DAG that fires on a schedule without you, that knows the difference between exit 1 and exit 2 and retries one while paging for the other, that backfills a single bad day using the dynamic partition overwrite you built in stage 8, and that refuses to run downstream tasks when this one aborts. Every design decision in this capstone was made to be orchestrated: the arguments, the exit codes, the JSON summary, the partition-level idempotency. You have built the job. What is left is the system that runs it.