Lesson 1 - What Is a Data Pipeline?

Welcome to What Is a Data Pipeline?#

Every course in this path has used the word “pipeline” loosely: CityFlow’s pipeline is faster now, CityFlow’s pipeline runs in a container. This lesson stops being loose about it. A data pipeline is not a vibe — it’s a small, specific set of ideas, and you can point at exactly where each one lives in CityFlow’s own ingest→validate→transform→load job.

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

• Define a data pipeline as a set of tasks connected by dependencies
• Explain why the order those tasks run in is not a style choice but a correctness requirement
• Define idempotency and show, with real output, why a pipeline that isn’t idempotent is dangerous to re-run
• Explain what scheduling adds on top of a script that already works correctly once

What Makes Something a “Pipeline”#

Strip away the word “pipeline” and what’s left is: a sequence of steps, each one doing a small, specific job, where a later step needs what an earlier step produced. CityFlow’s monthly reporting job has exactly four such steps:

1. Ingest — pull the raw trip records for a given day out of the month’s parquet file.
2. Validate — throw out rows that can’t be real trips (a fare of zero, no passengers, no distance).
3. Transform — turn validated rows into the numbers CityFlow actually cares about: trip count, average fare, total revenue.
4. Load — hand the finished summary to wherever it needs to go (here, just printing it, but in production a database or dashboard).

That’s it. That’s a pipeline. Nothing here requires Airflow, a scheduler, or even more than one file — a pipeline is a shape, not a piece of software. Below, each step is a plain Python function, run against a real day of CityFlow’s January 2024 taxi data:

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

TARGET_DATE = "2024-01-15"
COLUMNS = ["tpep_pickup_datetime", "passenger_count", "trip_distance", "fare_amount", "total_amount"]

def ingest(date):
    raw = pd.read_parquet("yellow_tripdata_2024-01.parquet", columns=COLUMNS)
    return raw[raw["tpep_pickup_datetime"].dt.strftime("%Y-%m-%d") == date].copy()

def validate(day):
    return day[
        (day["passenger_count"] > 0)
        & (day["trip_distance"] > 0)
        & (day["fare_amount"] > 0)
    ]

def transform(clean_day):
    return {
        "trips": len(clean_day),
        "avg_fare": round(float(clean_day["fare_amount"].mean()), 2),
        "revenue": round(float(clean_day["total_amount"].sum()), 2),
    }

def load(summary, date):
    print(f"{date}: {summary}")
    return summary

raw_day = ingest(TARGET_DATE)
clean_day = validate(raw_day)
summary = transform(clean_day)
load(summary, TARGET_DATE)

2024-01-15: {'trips': 71484, 'avg_fare': 19.64, 'revenue': 2053217.25}

Four functions, four tasks, run in a fixed order. That order is the whole game — and it’s worth seeing exactly what breaks if you get it wrong.

Dependencies: Order That Matters#

transform doesn’t just happen to run after validate — it depends on it. transform assumes every row it receives is a real trip, because that’s the entire job validate did. Break that assumption by feeding transform the raw, unvalidated day instead, skipping the dependency entirely:

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

TARGET_DATE = "2024-01-15"
COLUMNS = ["tpep_pickup_datetime", "passenger_count", "trip_distance", "fare_amount", "total_amount"]

def ingest(date):
    raw = pd.read_parquet("yellow_tripdata_2024-01.parquet", columns=COLUMNS)
    return raw[raw["tpep_pickup_datetime"].dt.strftime("%Y-%m-%d") == date].copy()

def validate(day):
    return day[
        (day["passenger_count"] > 0)
        & (day["trip_distance"] > 0)
        & (day["fare_amount"] > 0)
    ]

def transform(clean_day):
    return {
        "trips": len(clean_day),
        "avg_fare": round(float(clean_day["fare_amount"].mean()), 2),
        "revenue": round(float(clean_day["total_amount"].sum()), 2),
    }

raw_day = ingest(TARGET_DATE)
clean_day = validate(raw_day)

correct = transform(clean_day)
skipped_validate = transform(raw_day)  # dependency skipped entirely

print("with validate (correct):   ", correct)
print("without validate (skipped):", skipped_validate)
print("extra rows counted as trips:", len(raw_day) - len(clean_day))

with validate (correct):    {'trips': 71484, 'avg_fare': 19.64, 'revenue': 2053217.25}
without validate (skipped): {'trips': 77033, 'avg_fare': 19.3, 'revenue': 2165047.64}
extra rows counted as trips: 5549

Skipping the dependency didn’t crash anything — transform ran just fine on the raw data and produced a number. It’s simply the wrong number: 5,549 rows that aren’t real trips (zero fares, zero distance, no passengers) got counted anyway, inflating the trip count and quietly shifting revenue by over $100,000. This is the core reason dependencies matter: a pipeline’s correctness depends on its steps running in the right order, not just on each step being individually correct code. A dependency is a promise — “transform may only run once validate has finished” — and nothing in plain Python enforces that promise for you. You have to write the calls in the right order yourself, and remember to, every time.

Idempotency: Safe to Re-run#

A pipeline will get re-run. Maybe it crashed halfway last time, maybe someone wants to regenerate yesterday’s report, maybe a scheduler retries it automatically. Idempotency is the property that running a task again with the same input produces the same result — no double-counting, no drift, no surprises:

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

TARGET_DATE = "2024-01-15"
COLUMNS = ["tpep_pickup_datetime", "passenger_count", "trip_distance", "fare_amount", "total_amount"]

def ingest(date):
    raw = pd.read_parquet("yellow_tripdata_2024-01.parquet", columns=COLUMNS)
    return raw[raw["tpep_pickup_datetime"].dt.strftime("%Y-%m-%d") == date].copy()

def validate(day):
    return day[
        (day["passenger_count"] > 0)
        & (day["trip_distance"] > 0)
        & (day["fare_amount"] > 0)
    ]

def transform(clean_day):
    return {
        "trips": len(clean_day),
        "avg_fare": round(float(clean_day["fare_amount"].mean()), 2),
        "revenue": round(float(clean_day["total_amount"].sum()), 2),
    }

summary_first_run = transform(validate(ingest(TARGET_DATE)))
summary_second_run = transform(validate(ingest(TARGET_DATE)))

print("first run: ", summary_first_run)
print("second run:", summary_second_run)
print("identical:", summary_first_run == summary_second_run)

first run:  {'trips': 71484, 'avg_fare': 19.64, 'revenue': 2053217.25}
second run: {'trips': 71484, 'avg_fare': 19.64, 'revenue': 2053217.25}
identical: True

This chain is idempotent because every step recomputes from source instead of accumulating: ingest re-reads the same rows from the parquet file every time, and transform recomputes the summary from scratch rather than adding to a running total. Compare that to Lesson 2, where writing output with .to_csv(..., mode="a") instead of overwriting would silently double-count every re-run. Idempotency isn’t automatic — it’s a property you have to design in, by preferring “recompute and overwrite” over “accumulate and append” whenever a task might ever run twice. And in a system with schedulers and retries, a task running twice isn’t a remote edge case; it’s a Tuesday.

Scheduling: Someone Has to Kick This Off#

Notice what every example above has in common: you ran it, by executing a Python script. Real pipelines don’t work that way — CityFlow’s report needs to run every single day, on its own, without anyone remembering to type a command. Scheduling is the piece that turns “a chain of tasks that produces the right answer” into “a chain of tasks that produces the right answer automatically, on a cadence, forever.”

That sounds like a small addition. It isn’t — “run this every day at 2 AM” opens up an entire category of new questions that plain functions never had to answer: What happens if yesterday’s run is still going when today’s starts? What happens if the machine was off at 2 AM and missed a day entirely? Who finds out if it fails silently at 2:03 AM while everyone’s asleep? Cron, the traditional Unix answer to “run this on a schedule,” answers exactly one of those questions (when to start) and has no opinion on the rest. Lesson 2 builds a real pipeline you’d hand to cron; Lesson 3 is where you feel precisely which of those questions cron leaves you holding.

Key Takeaways#

• A data pipeline is a set of tasks connected by dependencies — nothing more exotic than that.
• Dependencies encode a correctness requirement, not just a convention: running transform before validate produced a real, wrong number (77,033 counted trips instead of 71,484) without ever raising an error.
• Idempotency means re-running a task with the same input reproduces the same result; this lesson’s chain is idempotent because every step recomputes from the source file instead of accumulating state.
• Scheduling is what makes a correct pipeline run itself, and it introduces new problems (overlap, missed runs, silent failures) that correctness alone doesn’t solve.

Exercises#

Exercise 1 — Pick a different date in January 2024 (e.g. "2024-01-08") and run the four-task chain from the first code block against it. Report the trip count, average fare, and revenue load prints for that day.

Exercise 2 — Using the “skipped validate” pattern from the dependencies section, compute how many rows would have been wrongly counted as trips on your Exercise 1 date if validate were skipped. Is the gap bigger or smaller than the 5,549-row gap measured for 2024-01-15, and can you explain why a busier or quieter day might have a different-sized gap?

Exercise 3 — Rewrite transform so that instead of returning a dict, it appends a row to a module-level list called history every time it’s called, and returns len(history) as part of the summary. Run the full chain twice in the same script and show that this version is not idempotent — the second run’s summary differs from the first purely because of the accumulating list, even though the input data never changed.