Lesson 2 - A Pipeline You Build by Hand

Welcome to A Pipeline You Build by Hand#

Lesson 1 defined a pipeline as tasks connected by dependencies and proved the idea with four functions in one process. Real pipelines are rarely one process — they’re usually separate scripts, written at different times, chained together by whatever runs them. This lesson builds CityFlow’s monthly report exactly that way: three real, separate .py files, run in order the way a cron entry would run them, against the actual full month of January 2024 taxi data.

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

• Write a multi-step pipeline as separate scripts that communicate purely through files on disk
• Chain scripts with shell &&, the same mechanism a cron entry uses
• Process a real 2.96-million-row dataset end to end into a genuine daily and monthly report
• Explain why “communicates through files” is both what makes this simple and what makes it fragile — the fragility Lesson 3 explores directly

CityFlow’s real January data lives on the site here:

# gate: skip
# CityFlow's real January 2024 trip data (2,964,624 rows, ~50 MB).
import pandas as pd
raw = pd.read_parquet("https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-01.parquet")

Download it once and save it next to your three scripts; every runnable block below reads the local copy, yellow_tripdata_2024-01.parquet, by name.

The Plan: Three Small Scripts, Chained#

CityFlow’s hand-built pipeline is three files, each doing one job and writing one output for the next script to read:

1. step1_extract.py — reads the raw parquet, keeps only valid January trips, writes cityflow_valid_trips.parquet.
2. step2_aggregate.py — reads that file, groups it into a daily summary, writes cityflow_daily_summary.csv.
3. step3_report.py — reads the daily summary, computes monthly totals, writes and prints cityflow_pipeline_report.txt.

All three agree on one thing: a shared working directory. Each script computes the same WORKDIR path so that whichever script runs next can find what the previous one wrote — this is the entire “dependency mechanism” a hand-built pipeline has. There’s no registry of what ran, no check that step 1 actually succeeded before step 2 starts reading its file; there’s just a filename both scripts happen to agree on.

Step 1: Extract & Validate#

import warnings
warnings.filterwarnings("ignore")
import os
import tempfile
import pandas as pd

WORKDIR = os.path.join(tempfile.gettempdir(), "cityflow_pipeline")
os.makedirs(WORKDIR, exist_ok=True)

COLUMNS = [
    "tpep_pickup_datetime",
    "passenger_count",
    "trip_distance",
    "fare_amount",
    "total_amount",
]

raw = pd.read_parquet("yellow_tripdata_2024-01.parquet", columns=COLUMNS)
total_rows = len(raw)

in_january = raw["tpep_pickup_datetime"].dt.strftime("%Y-%m") == "2024-01"
valid = (
    in_january
    & (raw["passenger_count"] > 0)
    & (raw["trip_distance"] > 0)
    & (raw["fare_amount"] > 0)
)
clean = raw.loc[valid].copy()
clean["pickup_date"] = clean["tpep_pickup_datetime"].dt.strftime("%Y-%m-%d")

clean.to_parquet(os.path.join(WORKDIR, "cityflow_valid_trips.parquet"), index=False)

print(f"rows read from parquet:      {total_rows:,}")
print(f"rows kept after validation:  {len(clean):,}")
print(f"rows dropped:                {total_rows - len(clean):,}")
print(f"wrote {os.path.join(WORKDIR, 'cityflow_valid_trips.parquet')}")

rows read from parquet:      2,964,624
rows kept after validation:  2,723,788
rows dropped:                240,836
wrote /tmp/cityflow_pipeline/cityflow_valid_trips.parquet

Nearly 3 million raw rows, and 240,836 of them (about 8%) fail CityFlow’s basic sanity checks — a zero fare, a zero-distance trip, a trip logged with no passengers. step1_extract.py is the only script that ever touches the raw parquet; everything downstream trusts that this filtering already happened, exactly the dependency relationship from Lesson 1, now expressed as “reads the file step1_extract.py wrote” instead of “receives the return value of a function call.”

Step 2: Aggregate#

import warnings
warnings.filterwarnings("ignore")
import os
import tempfile
import pandas as pd

WORKDIR = os.path.join(tempfile.gettempdir(), "cityflow_pipeline")

clean = pd.read_parquet(os.path.join(WORKDIR, "cityflow_valid_trips.parquet"))

daily = (
    clean.groupby("pickup_date")
    .agg(
        trips=("fare_amount", "size"),
        avg_fare=("fare_amount", "mean"),
        avg_distance=("trip_distance", "mean"),
        total_revenue=("total_amount", "sum"),
    )
    .round(2)
    .reset_index()
    .sort_values("pickup_date")
)

daily.to_csv(os.path.join(WORKDIR, "cityflow_daily_summary.csv"), index=False)

print(f"days summarized: {len(daily)}")
print(daily.head(3).to_string(index=False))
print(f"wrote {os.path.join(WORKDIR, 'cityflow_daily_summary.csv')}")

days summarized: 31
pickup_date  trips  avg_fare  avg_distance  total_revenue
 2024-01-01  67408     21.98          4.32     2104286.02
 2024-01-02  70489     21.37          4.22     2182589.33
 2024-01-03  77442     20.05          3.95     2266794.80
wrote /tmp/cityflow_pipeline/cityflow_daily_summary.csv

All 31 days of January land in one summary file, each row a real day’s trip count, average fare, average distance, and total revenue, computed from the validated trips step1_extract.py produced. step2_aggregate.py never opens the original 2.96-million-row parquet at all — it only knows about the file step1_extract.py promised to leave behind.

Step 3: Report#

import warnings
warnings.filterwarnings("ignore")
import os
import tempfile
import pandas as pd

WORKDIR = os.path.join(tempfile.gettempdir(), "cityflow_pipeline")

daily = pd.read_csv(os.path.join(WORKDIR, "cityflow_daily_summary.csv"))

total_trips = int(daily["trips"].sum())
total_revenue = round(float(daily["total_revenue"].sum()), 2)
busiest = daily.loc[daily["trips"].idxmax()]
quietest = daily.loc[daily["trips"].idxmin()]

report_lines = [
    "CityFlow January 2024 Pipeline Report",
    "======================================",
    f"days covered:      {len(daily)}",
    f"total valid trips: {total_trips:,}",
    f"total revenue:     ${total_revenue:,.2f}",
    f"busiest day:       {busiest['pickup_date']} ({int(busiest['trips']):,} trips)",
    f"quietest day:      {quietest['pickup_date']} ({int(quietest['trips']):,} trips)",
]
report = "\n".join(report_lines)

report_path = os.path.join(WORKDIR, "cityflow_pipeline_report.txt")
with open(report_path, "w") as f:
    f.write(report + "\n")

print(report)
print(f"\nwrote {report_path}")

CityFlow January 2024 Pipeline Report
======================================
days covered:      31
total valid trips: 2,723,788
total revenue:     $74,663,590.85
busiest day:       2024-01-27 (102,260 trips)
quietest day:      2024-01-07 (62,408 trips)

wrote /tmp/cityflow_pipeline/cityflow_pipeline_report.txt

The final number: CityFlow’s validated January trips generated $74,663,590.85 in revenue across 2,723,788 rides, with January 27 the busiest single day at 102,260 trips. That’s a real answer, computed by three small scripts, none of which needed to know anything about the others beyond a shared filename.

Running It Like Cron Would#

A cron entry doesn’t call Python functions — it runs shell commands, one after another, and the standard way to make each one depend on the last succeeding is to chain them with &&:

python3 step1_extract.py && python3 step2_aggregate.py && python3 step3_report.py

&& means “run the next command only if the previous one exited with status 0.” That’s the entire dependency mechanism in this hand-built pipeline — not a graph, not a registry, just three exit codes and a lucky ordering. It works, and you just watched it work end to end on nearly 3 million real rows. Lesson 3 is about exactly what happens the first time one of those three exit codes isn’t zero.

Key Takeaways#

• A hand-built pipeline chains separate scripts through a shared filesystem location — here, WORKDIR’s cityflow_valid_trips.parquet and cityflow_daily_summary.csv.
• Shell && is cron’s dependency mechanism: run the next command only if the previous one succeeded, checked purely by exit code.
• The whole January month — 2,964,624 raw rows, 2,723,788 valid ones — reduces to one small, real report: $74,663,590.85 in revenue, busiest day 2024-01-27 with 102,260 trips.
• Nothing here verifies that step 2’s input is actually fresh, or complete, or in the format step 1 promised — it just reads whatever file exists at that path.

Exercises#

Exercise 1 — Run all three scripts yourself, in order, exactly as the && chain shows. Confirm your step3_report.py output matches this lesson’s numbers: 2,723,788 total trips and $74,663,590.85 in revenue.

Exercise 2 — Modify step1_extract.py to also drop trips where trip_distance is greater than 100 miles (a data-entry error, not a real trip). Re-run the full chain and report the new total valid-trip count and total revenue. Are they higher or lower than this lesson’s numbers, and why?

Exercise 3 — Delete cityflow_valid_trips.parquet from WORKDIR by hand, then run only python3 step2_aggregate.py (skip step 1). What happens, and what does the error message tell you about what step 2 assumed was true before it started?