Lesson 5 - Guided Project: CityFlow's Daily Report as a Real DAG

Welcome to the Guided Project#

Module 1’s guided project designed CityFlow’s zone-and-fare daily report as a six-task DAG entirely on paper — a dependency table, a diagram, and a topological_order check confirming the design was acyclic and its two summary branches genuinely independent. This project builds the exact same design as a real Airflow DAG, over real data, and proves it end to end.

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

• Translate a {task: [upstream_deps]} paper design directly into TaskFlow-API tasks and dependencies
• Pass large intermediate data between tasks through a shared file path instead of XCom, and explain why
• Watch Airflow actually run two independent tasks in parallel, using timestamps as proof
• Verify a finished DAG run two independent ways: the UI (a screenshot) and the CLI (states-for-dag-run)

The Same Design, Named the Same Way#

Module 1 Lesson 5’s dependency table is reproduced here on purpose — nothing about the design changes, only its implementation:

The data is CityFlow’s real 200,000-row taxi sample (yellow_tripdata_sample.csv, the same reproducible sample Scaling Python for Data Engineering uses) plus the real 265-zone lookup table from Lesson 3, both copied into the mounted dags/data/ folder so every task can reach them at a fixed path inside the container.

One Design Decision the Paper Version Didn’t Need: Where Data Lives Between Tasks#

Module 1’s paper design passed whole pandas DataFrames between plain Python functions in one process. Airflow tasks are separate processes, and XCom — the mechanism Lesson 3’s int passed through — is backed by the metadata database and meant for small values, not a 183,633-row table. This DAG’s tasks write intermediate results to a shared path on disk and pass only the file path (a short string) through XCom, the same “communicate through an agreed-on filename” pattern Module 1 Lesson 2’s hand-built step1_extract.py / step2_aggregate.py used — except now a real orchestrator is the one enforcing that validate_trips never starts before ingest_taxi_data finishes.

# gate: skip
"""CityFlow's zone-and-fare daily report, as a real Airflow DAG - the exact
six-task design Module 1's guided project validated on paper."""
from datetime import datetime

from airflow.decorators import dag, task

WORKDIR = "/opt/airflow/dags/data/work"
RAW_CSV = "/opt/airflow/dags/data/yellow_tripdata_sample.csv"
ZONE_LOOKUP = "/opt/airflow/dags/data/taxi_zone_lookup.csv"


@dag(
    dag_id="cityflow_daily_report",
    schedule=None,
    start_date=datetime(2026, 1, 1),
    catchup=False,
    tags=["cityflow", "module-2", "guided-project"],
)
def cityflow_daily_report_dag():
    @task
    def ingest_taxi_data() -> str:
        import os
        import pandas as pd

        os.makedirs(WORKDIR, exist_ok=True)
        raw = pd.read_csv(RAW_CSV)
        out_path = f"{WORKDIR}/ingested.csv"
        raw.to_csv(out_path, index=False)
        print(f"ingested {len(raw):,} raw trips -> {out_path}")
        return out_path

    @task
    def validate_trips(ingested_path: str) -> str:
        import pandas as pd

        raw = pd.read_csv(ingested_path)
        valid = raw[
            (raw["passenger_count"] > 0)
            & (raw["trip_distance"] > 0)
            & (raw["fare_amount"] > 0)
        ]
        out_path = f"{WORKDIR}/validated.csv"
        valid.to_csv(out_path, index=False)
        print(f"kept {len(valid):,} of {len(raw):,} rows -> {out_path}")
        return out_path

    @task
    def compute_fare_summary(validated_path: str) -> list:
        import pandas as pd

        valid = pd.read_csv(validated_path)
        bins = [0, 10, 25, 50, float("inf")]
        labels = ["0-10", "10-25", "25-50", "50+"]
        valid["fare_tier"] = pd.cut(valid["fare_amount"], bins=bins, labels=labels)
        summary = (
            valid.groupby("fare_tier", observed=True)
            .agg(trips=("fare_amount", "size"), revenue=("total_amount", "sum"))
            .round(2)
            .reset_index()
        )
        print(summary.to_string(index=False))
        return summary.to_dict(orient="records")

    @task
    def compute_zone_summary(validated_path: str) -> list:
        import pandas as pd

        valid = pd.read_csv(validated_path)
        zones = pd.read_csv(ZONE_LOOKUP)
        merged = valid.merge(
            zones, left_on="PULocationID", right_on="LocationID", how="left"
        )
        summary = (
            merged.groupby("Zone")
            .agg(trips=("fare_amount", "size"), revenue=("total_amount", "sum"))
            .round(2)
            .sort_values("trips", ascending=False)
            .head(5)
            .reset_index()
        )
        print(summary.to_string(index=False))
        return summary.to_dict(orient="records")

    @task
    def merge_summaries(fare_summary: list, zone_summary: list) -> dict:
        return {"fare_summary": fare_summary, "top_zones": zone_summary}

    @task
    def load_report(merged: dict) -> None:
        import json

        out_path = f"{WORKDIR}/cityflow_daily_report.json"
        with open(out_path, "w") as f:
            json.dump(merged, f, indent=2)
        print(f"wrote {out_path}")
        print(json.dumps(merged, indent=2))

    ingested = ingest_taxi_data()
    validated = validate_trips(ingested)
    fare_summary = compute_fare_summary(validated)
    zone_summary = compute_zone_summary(validated)
    merged = merge_summaries(fare_summary, zone_summary)
    load_report(merged)


cityflow_daily_report_dag()

Note that compute_fare_summary and compute_zone_summary are both called with validated as their only argument — neither refers to the other’s output. That’s the entire mechanism that lets Airflow schedule them in parallel: nothing in the code forces sequential order between them, exactly Module 1’s design decision, now expressed in code instead of a dependency table.

Triggering and Verifying the Real Run#

Same pattern as Lesson 3 — unpause, trigger with a unique run ID, poll to a terminal state — now against a DAG with six tasks and a real branch:

import json
import subprocess
import time

dag_id = "cityflow_daily_report"

# The scheduler needs a few seconds to register a brand-new DAG file in the
# metadata database before it can be triggered; this loop is a no-op if the
# DAG (as here) has already been picked up.
deadline = time.time() + 60
registered = False
while time.time() < deadline:
    check = 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(check.stdout)):
        registered = True
        break
    time.sleep(3)
assert registered, f"{dag_id} was never registered 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"lesson5_verify_{int(time.time())}"
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 {run_id!r}")

deadline = time.time() + 120
state = "queued"
while state not in ("success", "failed") and time.time() < deadline:
    time.sleep(4)
    result = 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(result.stdout)
    state = next(r["state"] for r in runs if r["run_id"] == run_id)

print(f"run {run_id!r} reached state: {state}")
assert state == "success", f"expected success, got {state}"

states = subprocess.run(
    ["docker", "exec", "airflow-local-airflow-scheduler-1",
     "airflow", "tasks", "states-for-dag-run", dag_id, run_id],
    capture_output=True, text=True, check=True,
)
print(states.stdout)

triggered 'cityflow_daily_report' as run 'lesson5_verify_1784740182'
run 'lesson5_verify_1784740182' reached state: success
dag_id                | execution_date            | task_id              | state   | start_date                       | end_date
======================+===========================+======================+=========+==================================+=================================
cityflow_daily_report | 2026-07-22T17:09:44+00:00 | compute_zone_summary | success | 2026-07-22T17:09:49.108456+00:00 | 2026-07-22T17:09:49.811869+00:00
cityflow_daily_report | 2026-07-22T17:09:44+00:00 | ingest_taxi_data     | success | 2026-07-22T17:09:45.518895+00:00 | 2026-07-22T17:09:47.035934+00:00
cityflow_daily_report | 2026-07-22T17:09:44+00:00 | validate_trips       | success | 2026-07-22T17:09:47.512785+00:00 | 2026-07-22T17:09:48.881423+00:00
cityflow_daily_report | 2026-07-22T17:09:44+00:00 | compute_fare_summary | success | 2026-07-22T17:09:49.095653+00:00 | 2026-07-22T17:09:49.804608+00:00
cityflow_daily_report | 2026-07-22T17:09:44+00:00 | merge_summaries      | success | 2026-07-22T17:09:50.703169+00:00 | 2026-07-22T17:09:50.811393+00:00
cityflow_daily_report | 2026-07-22T17:09:44+00:00 | load_report          | success | 2026-07-22T17:09:51.808477+00:00 | 2026-07-22T17:09:51.913788+00:00

Six tasks, all success, the whole run finished in about six seconds. Look closely at the start times: compute_fare_summary started at 17:09:49.095653 and compute_zone_summary started at 17:09:49.108456 — thirteen milliseconds apart, both only after validate_trips ended at 17:09:48.881423. That’s Airflow’s LocalExecutor genuinely running two independent tasks in parallel, not one after the other — the exact thing a &&-chained script from Module 1 Lesson 2 could never do, because && only knows how to run things one at a time.

The real numbers the run produced, straight from cityflow_daily_report.json:

{
  "fare_summary": [
    {"fare_tier": "0-10",   "trips": 66255, "revenue": 946899.47},
    {"fare_tier": "10-25",  "trips": 84658, "revenue": 1961156.0},
    {"fare_tier": "25-50",  "trips": 21011, "revenue": 1027284.71},
    {"fare_tier": "50+",    "trips": 11709, "revenue": 1090474.76}
  ],
  "top_zones": [
    {"Zone": "JFK Airport",                   "trips": 9242, "revenue": 750284.76},
    {"Zone": "Midtown Center",                 "trips": 9171, "revenue": 220885.62},
    {"Zone": "Upper East Side South",          "trips": 9031, "revenue": 178791.94},
    {"Zone": "Upper East Side North",          "trips": 8615, "revenue": 173937.92},
    {"Zone": "Penn Station/Madison Sq West",   "trips": 6890, "revenue": 166637.74}
  ]
}

Of 200,000 raw sampled trips, validate_trips kept 183,633 (dropping 16,367 with a zero fare, zero distance, or no passengers — the same validity rule Module 1 used throughout). The busiest pickup zone by trip count is JFK Airport, and it’s also the single highest-revenue zone by a wide margin — a $10 airport-run fare tier away from Midtown Center’s trip count being nearly identical but its revenue less than a third as large, exactly the kind of fact a zone-only or fare-only report would have missed on its own.

Verified Two Independent Ways#

The UI: a real branch, visibly#

The CLI: every task’s real final state#

The states-for-dag-run output above is the second, independent verification — not a paraphrase of the screenshot, but the same underlying metadata database queried a completely different way. Both agree: six tasks, success, the branch shape matches Module 1’s dependency table exactly.

What This Project Actually Proved#

Module 1’s guided project proved a six-task, branching DAG design was acyclic, fully resolved, and correctly parallel — using a topological_order function, on paper, with no orchestrator anywhere. This project took that exact design and made it real: the same task names, the same dependency shape, running over 200,000 real taxi trips instead of no data at all, triggered with one CLI command and verified two independent ways. The parallel timestamps for compute_fare_summary and compute_zone_summary are the proof that “these two tasks are independent” stopped being a design claim on paper and became something a real scheduler actually acted on.

Key Takeaways#

• A paper {task: [upstream_deps]} design translates directly into TaskFlow-API code: the same task names, the same dependency shape, no redesign needed.
• Large intermediate data belongs in a shared file location referenced by a short string passed through XCom, not in XCom directly — the same “agree on a filename” pattern Module 1’s hand-built scripts used, now enforced by a real scheduler instead of a lucky && chain.
• Two tasks with no edge between them (here, compute_fare_summary and compute_zone_summary) really do run in parallel under LocalExecutor — millisecond-apart start times are the real evidence, not an assumption.
• A finished DAG run is worth confirming two independent ways — a UI screenshot and a CLI state query — because they read the same metadata database through entirely different code paths, and agreement between them is real evidence, not a coincidence to hope for.

Exercises#

Exercise 1 — Open the Graph view for cityflow_daily_report yourself and click the merge_summaries box. Confirm its Details tab lists both compute_fare_summary and compute_zone_summary as upstream dependencies — the UI’s own account of the same edge Module 1’s dependency table specified.

Exercise 2 — Add a seventh task, notify_team, that depends on load_report and prints a one-line “CityFlow daily report ready” message — the same task Module 1 Lesson 5’s Exercise 2 asked you to design on paper. Trigger the updated DAG and confirm all seven tasks reach success.

Exercise 3 — Using docker exec airflow-local-airflow-scheduler-1 airflow tasks states-for-dag-run cityflow_daily_report <run_id>, compute by hand (from the real start_date/end_date columns) how much wall-clock time the parallel branch saved compared to if compute_fare_summary and compute_zone_summary had run strictly one after the other. Is the saving large in this DAG, and would you expect it to matter more on a bigger dataset?