DATATWEETS
Courses
DATATWEETS

Lesson 3 - Validate and Transform Tasks

Welcome to Validate and Transform Tasks

Lesson 1 argued that validate and transform belong in separate tasks so a bad batch stops before wasted transform work runs. This lesson makes both tasks real. validate_trips doesn’t just check len(df) > 0 — it runs five genuine data-quality checks against every real ingested row and writes what it found somewhere a human (or another task) can actually inspect. transform_daily_borough_summary only ever sees the rows that passed, and does real, meaningful aggregation work on them.

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

  • Write real null, range, and ordering checks against a real DataFrame, and combine them into one overall pass/fail decision
  • Explain why validation results belong in a file, not just a print() statement
  • Join and aggregate real data into a genuinely different, smaller, more useful shape

Five Real Checks, Not One len() > 0

validate_trips reads Lesson 2’s real raw_trips.csv and checks it five different ways, each one a real condition a real downstream consumer would care about:

CheckWhat it catches
null_required_fieldsMissing passenger_count, trip_distance, fare_amount, PULocationID, or DOLocationID
trip_distance_out_of_rangeNegative or implausibly long (> 100 miles) trips
fare_amount_out_of_rangeNegative or implausibly high (> 500) fares
passenger_count_out_of_rangeFewer than 0 or more than 6 passengers
pickup_after_dropoffA pickup timestamp later than its own dropoff timestamp

A row fails validation if it fails any of these — the checks aren’t mutually exclusive, so the task reports each one’s own count as well as the combined total.

# gate: skip
"""CityFlow's real validate + transform tasks (Module 5 Lesson 3).

Two real, separate tasks -- not one script doing both jobs. validate_trips
runs genuine data-quality checks (schema/null checks, range checks on
distance/fare/passenger_count, a pickup-before-dropoff time check) against a
real day of CityFlow trips and writes its findings to an inspectable JSON
report plus a quarantine CSV of the rows that failed, instead of just
printing a count and discarding it. transform_daily_borough_summary only
ever sees the rows validate_trips passed, and does real aggregation work:
joining pickup zone to borough and computing a real per-day/per-borough
summary.

ingest_taxi_trips is unchanged from Lesson 2 -- reused here, not
reimplemented, so this DAG has real input to validate against."""
from datetime import datetime

from airflow.decorators import dag, task

PARQUET_PATH = "/opt/airflow/dags/data/yellow_tripdata_2024-01.parquet"
ZONE_LOOKUP_PATH = "/opt/airflow/dags/data/taxi_zone_lookup.csv"
WORK_DIR = "/opt/airflow/dags/data/work/etl_validate_transform_demo"
RAW_TRIPS_PATH = f"{WORK_DIR}/raw_trips.csv"
CLEAN_TRIPS_PATH = f"{WORK_DIR}/clean_trips.csv"
QUARANTINE_PATH = f"{WORK_DIR}/quarantine_trips.csv"
VALIDATION_REPORT_PATH = f"{WORK_DIR}/validation_report.json"
TRANSFORMED_PATH = f"{WORK_DIR}/daily_borough_summary.csv"
INGEST_START = "2024-01-01"
INGEST_END_EXCLUSIVE = "2024-01-02"
KEEP_COLUMNS = [
    "tpep_pickup_datetime", "tpep_dropoff_datetime", "passenger_count",
    "trip_distance", "PULocationID", "DOLocationID", "payment_type",
    "fare_amount", "tip_amount", "total_amount",
]
MIN_PASS_RATE = 0.5  # sanity floor -- real data comfortably clears this


@dag(
    dag_id="cityflow_etl_validate_transform_demo",
    schedule=None,
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["cityflow", "module-5", "validate-transform"],
)
def cityflow_etl_validate_transform_demo_dag():
    @task
    def ingest_taxi_trips() -> int:
        import os

        import pandas as pd

        raw = pd.read_parquet(PARQUET_PATH, columns=["tpep_pickup_datetime"] + [
            c for c in KEEP_COLUMNS if c != "tpep_pickup_datetime"
        ])
        raw["tpep_pickup_datetime"] = pd.to_datetime(raw["tpep_pickup_datetime"])
        raw["tpep_dropoff_datetime"] = pd.to_datetime(raw["tpep_dropoff_datetime"])
        start = pd.Timestamp(INGEST_START)
        end = pd.Timestamp(INGEST_END_EXCLUSIVE)
        sliced = raw[(raw["tpep_pickup_datetime"] >= start) & (raw["tpep_pickup_datetime"] < end)]
        sliced = sliced[KEEP_COLUMNS]
        os.makedirs(WORK_DIR, exist_ok=True)
        sliced.to_csv(RAW_TRIPS_PATH, index=False)
        print(f"ingested {len(sliced)} real CityFlow trips -> {RAW_TRIPS_PATH}")
        return len(sliced)

    @task
    def validate_trips(ingested_rows: int) -> dict:
        """Real data-quality checks against every real ingested row.
        Writes a real JSON report and a real quarantine CSV -- nothing here
        is just printed and thrown away."""
        import json

        import pandas as pd

        trips = pd.read_csv(RAW_TRIPS_PATH, parse_dates=["tpep_pickup_datetime", "tpep_dropoff_datetime"])
        assert len(trips) == ingested_rows, "raw_trips.csv row count doesn't match what ingest reported"

        required = ["passenger_count", "trip_distance", "fare_amount", "PULocationID", "DOLocationID"]
        null_mask = trips[required].isna().any(axis=1)
        dist_mask = (trips["trip_distance"] < 0) | (trips["trip_distance"] > 100)
        fare_mask = (trips["fare_amount"] < 0) | (trips["fare_amount"] > 500)
        pax_mask = (trips["passenger_count"] < 0) | (trips["passenger_count"] > 6)
        time_mask = trips["tpep_pickup_datetime"] > trips["tpep_dropoff_datetime"]

        checks = {
            "null_required_fields": int(null_mask.sum()),
            "trip_distance_out_of_range": int(dist_mask.sum()),
            "fare_amount_out_of_range": int(fare_mask.sum()),
            "passenger_count_out_of_range": int(pax_mask.sum()),
            "pickup_after_dropoff": int(time_mask.sum()),
        }
        any_fail = null_mask | dist_mask | fare_mask | pax_mask | time_mask
        rows_failed = int(any_fail.sum())
        rows_passed = int((~any_fail).sum())
        pass_rate = rows_passed / len(trips)

        report = {
            "total_rows": len(trips), "checks": checks,
            "rows_failed_any_check": rows_failed, "rows_passed": rows_passed,
            "pass_rate": round(pass_rate, 4),
        }
        with open(VALIDATION_REPORT_PATH, "w") as f:
            json.dump(report, f, indent=2)
        print(f"validation report -> {VALIDATION_REPORT_PATH}: {report}")

        trips[any_fail].to_csv(QUARANTINE_PATH, index=False)
        trips[~any_fail].to_csv(CLEAN_TRIPS_PATH, index=False)
        print(f"{rows_failed} rows quarantined -> {QUARANTINE_PATH}")
        print(f"{rows_passed} clean rows -> {CLEAN_TRIPS_PATH}")

        assert pass_rate >= MIN_PASS_RATE, (
            f"pass_rate {pass_rate:.4f} is below the {MIN_PASS_RATE} floor -- "
            f"stopping here on purpose so transform never runs on data this bad"
        )
        return report

    @task
    def transform_daily_borough_summary(validation_report: dict) -> dict:
        """Real, meaningful work on the validated rows only: join each
        trip's pickup zone to its borough, derive trip duration, and
        aggregate to a real per-day/per-borough summary -- not a
        pass-through copy."""
        import pandas as pd

        clean = pd.read_csv(CLEAN_TRIPS_PATH, parse_dates=["tpep_pickup_datetime", "tpep_dropoff_datetime"])
        assert len(clean) == validation_report["rows_passed"], "clean_trips.csv doesn't match validate_trips's count"

        zones = pd.read_csv(ZONE_LOOKUP_PATH)[["LocationID", "Borough"]].rename(columns={"LocationID": "PULocationID"})
        clean = clean.merge(zones, on="PULocationID", how="left")
        clean["Borough"] = clean["Borough"].fillna("Unknown")

        clean["trip_duration_minutes"] = (
            clean["tpep_dropoff_datetime"] - clean["tpep_pickup_datetime"]
        ).dt.total_seconds() / 60.0
        clean["pickup_date"] = clean["tpep_pickup_datetime"].dt.date

        agg = clean.groupby(["pickup_date", "Borough"]).agg(
            trip_count=("fare_amount", "size"),
            avg_fare_amount=("fare_amount", "mean"),
            avg_trip_distance=("trip_distance", "mean"),
            avg_trip_duration_minutes=("trip_duration_minutes", "mean"),
            total_revenue=("total_amount", "sum"),
        ).reset_index()
        for col in ("avg_fare_amount", "avg_trip_distance", "avg_trip_duration_minutes", "total_revenue"):
            agg[col] = agg[col].round(2)

        agg.to_csv(TRANSFORMED_PATH, index=False)
        total_trip_count = int(agg["trip_count"].sum())
        assert total_trip_count == validation_report["rows_passed"], (
            f"transformed total_trip_count {total_trip_count} != "
            f"validate_trips's rows_passed {validation_report['rows_passed']}"
        )
        print(f"{len(agg)} real day/borough summary rows -> {TRANSFORMED_PATH}, total_trip_count={total_trip_count}")
        return {"summary_rows": len(agg), "total_trip_count": total_trip_count}

    transform_daily_borough_summary(validate_trips(ingest_taxi_trips()))


cityflow_etl_validate_transform_demo_dag()

Two details worth calling out. First, validate_trips writes three real artifacts, not one: the JSON report (the numbers), a quarantine CSV (the actual bad rows, for someone to go look at), and a clean CSV (what transform is allowed to see). Second, transform_daily_borough_summary doesn’t just trust that clean_trips.csv has the right number of rows — it asserts len(clean) == validation_report["rows_passed"], a real invariant check that the file transform is reading actually is the file validate said it wrote, passed forward through XCom rather than assumed.

Triggered for Real: 11,723 Quarantined, 69,290 Clean

# gate: skip
import json
import subprocess
import time

dag_id = "cityflow_etl_validate_transform_demo"

deadline = time.time() + 320
found = False
while time.time() < deadline:
    result = subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "list", "--output", "json"],
        capture_output=True, text=True, check=True,
    )
    if any(d["dag_id"] == dag_id for d in json.loads(result.stdout)):
        found = True
        break
    time.sleep(5)
assert found, f"{dag_id} was never picked up by the scheduler"

subprocess.run(["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "unpause", dag_id],
                capture_output=True, text=True, check=True)
run_id = f"lesson3_m5_verify_{int(time.time())}"
trigger = subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "trigger", dag_id, "--run-id", run_id],
    capture_output=True, text=True,
)
if trigger.returncode != 0:
    subprocess.run(["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "reserialize"],
                    capture_output=True, text=True, check=True)
    subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "trigger", dag_id, "--run-id", run_id],
        capture_output=True, text=True, check=True,
    )
print(f"triggered {dag_id!r} as {run_id!r}")

deadline = time.time() + 90
state = "queued"
while state not in ("success", "failed") and time.time() < deadline:
    time.sleep(3)
    r = subprocess.run(
        ["docker", "exec", "airflow-local-airflow-scheduler-1", "airflow", "dags", "list-runs", "-d", dag_id, "--output", "json"],
        capture_output=True, text=True, check=True,
    )
    runs = json.loads(r.stdout)
    state = next(x["state"] for x in runs if x["run_id"] == run_id)
print(f"run {run_id!r} reached state: {state}")
assert state == "success", f"expected success, got {state}"

# Independent check: read the real report and real files the tasks wrote.
with open(".airflow-local/dags/data/work/etl_validate_transform_demo/validation_report.json") as f:
    report = json.load(f)
print("validation_report.json:", json.dumps(report, indent=2))

import pandas as pd

q = pd.read_csv(".airflow-local/dags/data/work/etl_validate_transform_demo/quarantine_trips.csv")
c = pd.read_csv(".airflow-local/dags/data/work/etl_validate_transform_demo/clean_trips.csv")
agg = pd.read_csv(".airflow-local/dags/data/work/etl_validate_transform_demo/daily_borough_summary.csv")
print(f"direct file check: {len(q)} quarantined, {len(c)} clean, {len(agg)} summary rows")
assert len(q) == report["rows_failed_any_check"]
assert len(c) == report["rows_passed"]
assert agg["trip_count"].sum() == report["rows_passed"]
triggered 'cityflow_etl_validate_transform_demo' as 'lesson3_m5_verify_1784914302'
run 'lesson3_m5_verify_1784914302' reached state: success
validation_report.json: {
  "total_rows": 81013,
  "checks": {
    "null_required_fields": 10518,
    "trip_distance_out_of_range": 1,
    "fare_amount_out_of_range": 1307,
    "passenger_count_out_of_range": 4,
    "pickup_after_dropoff": 2
  },
  "rows_failed_any_check": 11723,
  "rows_passed": 69290,
  "pass_rate": 0.8553
}
direct file check: 11723 quarantined, 69290 clean, 7 summary rows

The real breakdown is genuinely informative: null_required_fields (10,518 rows, almost entirely missing passenger_count) dwarfs every other check, and fare_amount_out_of_range (1,307 rows — real negative fares from refunds, and a handful of implausibly high ones) is the second-biggest real issue. trip_distance_out_of_range, passenger_count_out_of_range, and pickup_after_dropoff each catch only a handful of rows — real, but rare. An 85.53% pass rate on real, unfiltered NYC taxi data is a believable number precisely because it’s real: production data is never 100% clean, and a validation report that always shows zero failures is a report nobody would trust.

The real aggregation transform produced from the 69,290 clean rows:

  pickup_date        Borough  trip_count  avg_fare_amount  avg_trip_distance  avg_trip_duration_minutes  total_revenue
0  2024-01-01          Bronx          76            26.79               4.51                      18.78        2317.19
1  2024-01-01       Brooklyn         646            27.45               4.41                      19.67       22562.33
2  2024-01-01            EWR          20            78.48               0.86                       0.61        2015.52
3  2024-01-01      Manhattan       59000            17.06               2.87                      14.87     1467296.17
4  2024-01-01         Queens        9174            53.79              13.04                      27.40      661604.49
5  2024-01-01  Staten Island           3            96.40               5.87                      12.74         342.94
6  2024-01-01        Unknown         371            40.06               3.44                      19.39       18225.87

Seven real rows — one per borough that had trips pickup on 2024-01-01 — down from 81,013 raw rows and 69,290 clean ones. trip_count sums to exactly 69,290, matching validate_trips’s own rows_passed, because transform_daily_borough_summary asserted that invariant before ever writing the file.

Key Takeaways

  • Real data-quality checks report which check failed and how many rows, not just a single pass/fail bit — validation_report.json’s checks breakdown is what makes 11,723 failures actionable instead of just alarming.
  • A quarantine file (the actual bad rows) is as important as the report (the counts) — someone eventually has to go look at why 10,518 rows were missing passenger_count, and the report alone can’t answer that.
  • transform_daily_borough_summary never trusts validate_trips’s numbers blindly — it re-asserts len(clean) == rows_passed and total_trip_count == rows_passed as real, executable invariants, not comments.
  • An 85.53% pass rate on real data is more trustworthy evidence the checks are real than a 100% pass rate would be.

Exercises

Exercise 1 — Add a sixth check, total_amount_less_than_fare_amount (flagging any row where total_amount < fare_amount, which shouldn’t be possible since total_amount includes fare plus tips/tolls/fees), and re-trigger. How many real rows does it catch?

Exercise 2 — Open quarantine_trips.csv after a real run and compute, using pandas, what fraction of the quarantined rows failed only the null_required_fields check (as opposed to failing that one plus another). Is it closer to 100% or notably lower?

Exercise 3 — Change MIN_PASS_RATE to 0.9 (above the real 0.8553 this lesson’s data produces) and re-trigger. Confirm validate_trips now genuinely fails, and that transform_daily_borough_summary shows upstream_failed — the same mechanism Lesson 1 proved, now enforced by a real number instead of a forced flag.

Previous lesson
Lesson 2 - The Ingest Task
Next lesson
Lesson 4 - The Load Task
Sponsor

Keep DATATWEETS free. Help fund practical data, AI, and engineering lessons for learners worldwide.

Buy Me a Coffee at ko-fi.com