A small, real pipeline for monthly taxi files

4 min read

Why it matters

"Design and build ETL pipelines" appeared in 96% of the Data Engineer job descriptions we read. Interviewers ask you to walk through a pipeline you built: sources, transformations, where it ran, how you knew it worked, and what happened when it failed. This lesson gives you one to talk about, on real public data.

The concept

A pipeline has stages, each with one job:

Extract raw data exactly as the source gives it, and keep a copy (the raw layer).
Validate early, before bad data spreads.
Transform into clean, typed, modelled tables.
Load into the destination in a layout that suits queries (for files: Parquet partitioned by date).
Log what happened: rows in, rows out, rows rejected, duration.

Two properties matter more than speed:

Idempotent: running the same month twice gives the same result, not duplicates.
Observable: when it fails, the log says where and why.

Worked example: NYC TLC yellow taxi months

pipeline.py

"""Load NYC TLC yellow taxi months into partitioned, cleaned Parquet.

    python pipeline.py 2025-01 2025-02
"""
import logging
import sys
import time
from pathlib import Path

import duckdb

SOURCE = "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_{month}.parquet"
RAW = Path("lake/raw/yellow")
CLEAN = Path("lake/clean/yellow")
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("taxi")


def extract(con: duckdb.DuckDBPyConnection, month: str) -> Path:
    target = RAW / f"{month}.parquet"
    if not target.exists():                       # the raw layer is written once, never edited
        RAW.mkdir(parents=True, exist_ok=True)
        con.sql(f"copy (select * from '{SOURCE.format(month=month)}') to '{target}' (format parquet)")
    return target


def validate(con: duckdb.DuckDBPyConnection, raw: Path, month: str) -> None:
    rows, outside = con.sql(f"""
        select count(*),
               count(*) filter (where strftime(tpep_pickup_datetime, '%Y-%m') <> '{month}')
        from '{raw}'""").fetchone()
    if rows < 1_000_000:
        raise ValueError(f"{month}: only {rows} rows, expected millions")
    if outside / rows > 0.001:
        raise ValueError(f"{month}: {outside} pickups outside the month")
    log.info("%s validated: %s rows, %s outside the month", month, rows, outside)


def transform_and_load(con: duckdb.DuckDBPyConnection, raw: Path, month: str) -> int:
    out = CLEAN / f"month={month}"
    out.mkdir(parents=True, exist_ok=True)
    target = out / "trips.parquet"                # overwritten on every run: idempotent
    con.sql(f"""
        copy (
          select tpep_pickup_datetime  as pickup_at,
                 tpep_dropoff_datetime as dropoff_at,
                 PULocationID as pickup_zone, DOLocationID as dropoff_zone,
                 trip_distance, fare_amount, tip_amount, payment_type
          from '{raw}'
          where strftime(tpep_pickup_datetime, '%Y-%m') = '{month}'
            and tpep_dropoff_datetime >= tpep_pickup_datetime
            and fare_amount >= 0
        ) to '{target}' (format parquet)""")
    return con.sql(f"select count(*) from '{target}'").fetchone()[0]


def run(month: str) -> None:
    started = time.time()
    con = duckdb.connect()
    raw = extract(con, month)
    validate(con, raw, month)
    loaded = transform_and_load(con, raw, month)
    log.info("%s loaded %s rows in %.1fs", month, loaded, time.time() - started)


if __name__ == "__main__":
    for month in sys.argv[1:]:
        run(month)

Run it for January 2025. The log reports 3,475,226 rows with 22 pickups outside the month, and about 3.33 million clean rows loaded; the rest are refunds with negative fares and impossible timestamps. Run it again: the raw file is reused and the clean file is overwritten, so the result is identical. That is idempotency in its simplest form.

Query the output as one table across months, with the partition as a column:

query_the_lake.sql

select month, count(*) as trips, round(median(fare_amount), 2) as median_fare
from read_parquet('lake/clean/yellow/*/trips.parquet', hive_partitioning = true)
group by month order by month;

Common mistakes

Transforming during extract. If a rule is wrong, you cannot rebuild without the raw data. Land raw first.
Appending on rerun. An insert without deleting the partition first doubles the data on every retry.
Validating at the end. By then bad data is in every downstream table. Validate right after extract.
Silent row loss. Log rows in and out of every stage, so a filter that suddenly drops 40% is visible.

Interview angle

"Walk me through a pipeline you built." Structure: source and volume, the raw, clean and modelled layers, validation rules, how reruns are safe, how it is scheduled, and one failure you handled. This pipeline covers every point.

Practice task

Add the zone lookup as a dimension and write a fct_trips table with borough names.
Schedule the script with cron or GitHub Actions for the 1st of each month (the TLC publishes with a delay; handle "file not found yet").
Add a reject table: rows that failed a rule, with the rule name.

#Why it matters

#The concept

#Worked example: NYC TLC yellow taxi months

#Common mistakes

#Interview angle

#Practice task

#Further reading