Building Your First Data Pipeline: A Walkthrough

A data pipeline moves data from where it is produced to where it is used, reshaping it on the way, and does it the same way every time. The best first pipeline is small but real: real data, real errors, and a result you can check. This walkthrough builds one in a few dozen lines of Python, using a month of New York City yellow taxi trips, and pays particular attention to the two habits that separate a pipeline from a script: rejecting bad data explicitly, and being safe to run again.

The source

The NYC Taxi and Limousine Commission publishes trip records, pickup and drop-off times and locations, distances and itemized fares, as monthly Parquet files, typically two months after the month ends (NYC TLC: trip record data). The TLC is clear that it did not create the data and makes no representations about its accuracy, which makes it an honest training ground: the data has the kinds of errors production data has.

The goal: a table of trips, fares and minutes per pickup zone per day, which a dashboard could read.

The pipeline

Each step is a function with one job, so each can be tested and changed alone:

import sqlite3
import urllib.request
from pathlib import Path
import pandas as pd

def extract(month):
    path = Path(f"yellow_{month}.parquet")
    if not path.exists():                                   # download once; reruns reuse the file
        url = f"https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_{month}.parquet"
        urllib.request.urlretrieve(url, path)
    return pd.read_parquet(path)

def transform(raw, month):
    start = pd.Timestamp(f"{month}-01")
    end = start + pd.offsets.MonthBegin(1)
    rules = {
        "pickup outside the month": ~raw["tpep_pickup_datetime"].between(start, end, inclusive="left"),
        "dropoff before pickup": raw["tpep_dropoff_datetime"] < raw["tpep_pickup_datetime"],
        "negative fare": raw["fare_amount"] < 0,
    }
    rejected = pd.Series(False, index=raw.index)
    for rule, broken in rules.items():
        print(f"  {rule:<25} {broken.sum():>7} rows")
        rejected |= broken
    trips = raw[~rejected]
    daily = (trips.assign(pickup_date=trips["tpep_pickup_datetime"].dt.strftime("%Y-%m-%d"),
                          minutes=(trips["tpep_dropoff_datetime"] - trips["tpep_pickup_datetime"]).dt.total_seconds() / 60)
                  .groupby(["pickup_date", "PULocationID"], as_index=False)
                  .agg(trips=("fare_amount", "size"), fare=("fare_amount", "sum"), minutes=("minutes", "sum")))
    return daily, len(trips)

def load(daily, month, db):
    con = sqlite3.connect(db)
    with con:                                               # one transaction: all rows or none
        con.execute("""CREATE TABLE IF NOT EXISTS daily_zone_trips (
                           pickup_date TEXT, zone_id INTEGER, trips INTEGER, fare REAL, minutes REAL,
                           PRIMARY KEY (pickup_date, zone_id))""")
        con.execute("DELETE FROM daily_zone_trips WHERE pickup_date LIKE ?", (f"{month}-%",))   # replace the month
        con.executemany("INSERT INTO daily_zone_trips VALUES (?, ?, ?, ?, ?)", daily.itertuples(index=False, name=None))
    total, days, rows = con.execute("""SELECT SUM(trips), COUNT(DISTINCT pickup_date), COUNT(*)
                                       FROM daily_zone_trips WHERE pickup_date LIKE ?""", (f"{month}-%",)).fetchone()
    con.close()
    return total, days, rows

def run(month, db="taxi.db"):
    raw = extract(month)
    print(f"extracted {len(raw)} rows")
    daily, kept = transform(raw, month)
    total, days, rows = load(daily, month, db)
    assert total == kept, "loaded trips do not match the transformed trips"
    print(f"loaded {rows} day-zone rows covering {days} days and {total} trips")

run("2024-01")
print("second run:")
run("2024-01")

extracted 2964624 rows
  pickup outside the month       18 rows
  dropoff before pickup          56 rows
  negative fare               37448 rows
loaded 6948 day-zone rows covering 31 days and 2927103 trips
second run:
extracted 2964624 rows
  pickup outside the month       18 rows
  dropoff before pickup          56 rows
  negative fare               37448 rows
loaded 6948 day-zone rows covering 31 days and 2927103 trips

Step 1: extract

The extract step fetches the source and changes nothing. Saving the raw file locally means a rerun does not download it again, and if a later step fails you can inspect exactly what arrived. In production the raw copy usually goes to object storage and is kept, so any month can be reprocessed.

Step 2: transform, with explicit rejection rules

The January 2024 file holds 2,964,624 trips, and the rules catch real problems: 18 pickups outside January (the file is organised by month, yet a few records fall outside it), 56 trips that end before they start, and 37,448 negative fares. Together they reject 37,521 rows, one fewer than the sum of the three counts because one trip broke two rules. Naming each rule and printing its count turns silent filtering into a record: if negative fares jump to 300,000 next month, the log shows it immediately.

Whether negative fares are errors or refunds is a business question, not a coding one. Here they are excluded from a revenue-style table; a finance pipeline might keep them. Write the decision down either way. Data quality testing takes these rules further.

The transform then aggregates to one row per day and pickup zone: 6,948 rows instead of nearly three million, which is what a dashboard needs.

Keep the rejects, not just the counts

Counting rejected rows is the minimum. A better pipeline also writes them, with the name of the rule each one broke, to a quarantine table or file. Then an analyst can see which trips were dropped and why, a data owner can confirm the rules, and a corrected rule can be replayed against the raw file without guessing what was lost.

Step 3: load, safely

Two details make the load safe to repeat:

• Replace, don't append. The load deletes the month's rows before inserting them, so a second run replaces the first rather than doubling it. The primary key on date and zone would also reject accidental duplicates.
• One transaction. Python's sqlite3 connection, used as a context manager, commits if the block succeeds and rolls back if it raises (Python docs: sqlite3). If the insert fails halfway, the delete is undone too, and the table keeps last run's data rather than half a month.

The second run in the output proves the point: the same 6,948 rows and 2,927,103 trips, not double. That property, that running twice gives the same result as running once, is called idempotency; idempotency in data pipelines explains why every production pipeline needs it.

Step 4: check

The final assertion compares what was loaded with what the transform produced. It is a small check, but it catches a class of silent failures, such as a load that quietly skipped rows. Real pipelines add more: row counts against the previous month, no missing days, totals within expected ranges.

What changes in production

This pipeline runs by hand for one month. The next steps are the ones every production pipeline takes:

• Parameters and backfills: run it for any month, and for a range of months, from the command line.
• Incremental loading: process only new data; incremental loads covers watermarks and partitions.
• Scheduling and retries: an orchestrator runs it when new data arrives and retries failures; see the Airflow tutorial.
• Monitoring: alerts when rejection counts or totals move outside their normal range.
• A warehouse instead of SQLite, with the same replace-by-partition pattern.

Whether to transform before loading, as here, or load raw data and transform inside the warehouse is the ETL versus ELT choice; ETL vs ELT compares them.

An end-to-end ETL pipeline is one of the projects in the Data Engineering course, alongside its data processing and ETL module.

Sources

• TLC Trip Record Data (NYC Taxi and Limousine Commission) — NYC Taxi and Limousine Commission
• sqlite3: DB-API 2.0 interface for SQLite databases (Python documentation) — Python documentation