Skip to content

03-nyc-taxi-pipeline.md

Latest commit

 

History

History
308 lines (248 loc) · 7.08 KB

03-nyc-taxi-pipeline.md

File metadata and controls

308 lines (248 loc) · 7.08 KB
Raw

5.3 - Building an End-to-End Pipeline with NYC Taxi Data

Architecture

Three-layered pipeline using DuckDB as a locally hosted database:

  1. Ingestion layer: extract data and store in raw format
  2. Staging layer: pre-process, clean, transform, join with lookup tables
  3. Reports layer: aggregate data and run calculations

All assets have dependencies that create the data lineage Bruin uses for orchestration.

Project setup

Initialize from the zoomcamp template:

bruin init zoomcamp my-taxi-pipeline
cd my-taxi-pipeline

Project structure:

zoomcamp/
├── .bruin.yml
├── README.md
└── pipeline/
    ├── pipeline.yml
    └── assets/
        ├── ingestion/
        │   ├── trips.py
        │   ├── requirements.txt
        │   ├── payment_lookup.asset.yml
        │   └── payment_lookup.csv
        ├── staging/
        │   └── trips.sql
        └── reports/
            └── trips_report.sql

.bruin.yml

default_environment: default

environments:
  default:
    connections:
      duckdb:
        - name: duckdb-default
          path: duckdb.db

pipeline.yml

name: nyc_taxi
schedule: daily
start_date: "2022-01-01"
default_connections:
  duckdb: duckdb-default
variables:
  taxi_types:
    type: array
    items:
      type: string
    default: ["yellow"]
  • start_date: when running a full refresh, process data starting from this date
  • Custom variables: taxi_types lets you control which taxi types to ingest (yellow, green, or both)
  • Variables can be overridden at runtime with --var

Ingestion layer

Python asset: trips.py

The Python asset connects to the NYC taxi API and extracts data.

"""@bruin
name: ingestion.trips
type: python
image: python:3.11

materialization:
  type: table
  strategy: append

columns:
  - name: pickup_datetime
    type: timestamp
    description: "When the meter was engaged"
  - name: dropoff_datetime
    type: timestamp
    description: "When the meter was disengaged"
@bruin"""

import os
import json
import pandas as pd

def materialize():
    start_date = os.environ["BRUIN_START_DATE"]
    end_date = os.environ["BRUIN_END_DATE"]
    taxi_types = json.loads(os.environ["BRUIN_VARS"]).get("taxi_types", ["yellow"])

    # Generate list of months between start and end dates
    # Fetch parquet files from:
    # https://d37ci6vzurychx.cloudfront.net/trip-data/{taxi_type}_tripdata_{year}-{month}.parquet

    return final_dataframe
  • materialize() returns a DataFrame; Bruin handles inserting it into the destination
  • append strategy: each run inserts data without touching existing rows
  • Uses BRUIN_START_DATE / BRUIN_END_DATE environment variables for the time window
  • Uses BRUIN_VARS to read the taxi_types pipeline variable

Seed file: payment_lookup.asset.yml

Seed files ingest data from local CSV files into the database.

name: ingestion.payment_lookup
type: duckdb.seed
parameters:
  path: payment_lookup.csv
columns:
  - name: payment_type_id
    type: integer
    description: "Numeric code for payment type"
    primary_key: true
    checks:
      - name: not_null
      - name: unique
  - name: payment_type_name
    type: string
    description: "Human-readable payment type"
    checks:
      - name: not_null

payment_lookup.csv:

payment_type_id,payment_type_name
0,flex_fare
1,credit_card
2,cash
3,no_charge
4,dispute
5,unknown
6,voided_trip

Quality checks (not_null, unique) run automatically after the asset finishes.

requirements.txt

pandas
requests
pyarrow
python-dateutil

Bruin handles the environment and installs dependencies locally within the pipeline.

Staging layer

SQL asset: staging/trips.sql

/* @bruin
name: staging.trips
type: duckdb.sql

depends:
  - ingestion.trips
  - ingestion.payment_lookup

materialization:
  type: table
  strategy: time_interval
  incremental_key: pickup_datetime
  time_granularity: timestamp

columns:
  - name: pickup_datetime
    type: timestamp
    primary_key: true
    checks:
      - name: not_null

custom_checks:
  - name: row_count_greater_than_zero
    query: |
      SELECT CASE WHEN COUNT(*) > 0 THEN 1 ELSE 0 END
      FROM staging.trips
    value: 1
@bruin */

SELECT
    t.pickup_datetime,
    t.dropoff_datetime,
    t.pickup_location_id,
    t.dropoff_location_id,
    t.fare_amount,
    t.taxi_type,
    p.payment_type_name
FROM ingestion.trips t
LEFT JOIN ingestion.payment_lookup p
    ON t.payment_type = p.payment_type_id
WHERE t.pickup_datetime >= '{{ start_datetime }}'
  AND t.pickup_datetime < '{{ end_datetime }}'
QUALIFY ROW_NUMBER() OVER (
    PARTITION BY t.pickup_datetime, t.dropoff_datetime,
                 t.pickup_location_id, t.dropoff_location_id, t.fare_amount
    ORDER BY t.pickup_datetime
) = 1
  • time_interval strategy: deletes rows in the time window, then inserts the query result
  • The WHERE clause must filter to the same time window to avoid duplicates
  • QUALIFY ROW_NUMBER() deduplicates using a composite key
  • Dependencies on both ingestion.trips and ingestion.payment_lookup ensure this runs after ingestion

Reports layer

SQL asset: reports/trips_report.sql

/* @bruin
name: reports.trips_report
type: duckdb.sql

depends:
  - staging.trips

materialization:
  type: table
  strategy: time_interval
  incremental_key: trip_date
  time_granularity: date

columns:
  - name: trip_date
    type: date
    primary_key: true
  - name: taxi_type
    type: string
    primary_key: true
  - name: payment_type
    type: string
    primary_key: true
  - name: trip_count
    type: bigint
    checks:
      - name: non_negative
@bruin */

SELECT
    CAST(pickup_datetime AS DATE) AS trip_date,
    taxi_type,
    payment_type_name AS payment_type,
    COUNT(*) AS trip_count,
    SUM(fare_amount) AS total_fare,
    AVG(fare_amount) AS avg_fare
FROM staging.trips
WHERE pickup_datetime >= '{{ start_datetime }}'
  AND pickup_datetime < '{{ end_datetime }}'
GROUP BY 1, 2, 3

Running the full pipeline

# Validate structure and definitions
bruin validate ./pipeline/pipeline.yml

# Run with a small date range for testing
bruin run ./pipeline/pipeline.yml --start-date 2022-01-01 --end-date 2022-02-01

# Full refresh
bruin run ./pipeline/pipeline.yml --full-refresh

# Query results
bruin query --connection duckdb-default --query "SELECT COUNT(*) FROM ingestion.trips"

Open the pipeline YAML file in the Bruin panel and view the lineage tab to see all assets and their dependencies. Execution order:

  1. Ingestion assets run first (trips + lookup, in parallel)
  2. Staging asset runs after both ingestion assets complete
  3. Report asset runs after staging completes

Materialization strategies summary

Strategy Behavior
table Drop and recreate the table each time
append Insert new data without touching existing rows
merge Upsert based on key columns
time_interval Delete rows in date range, then re-insert
delete+insert Delete matching rows, then insert
create+replace Create or replace the table