Python Data Engineering: Comprehensive Workflow for Data Modeling, Analytics with DuckDB

import pandas as pd
import duckdb

#load data from `parquet` file
df =pd.read_parquet('https://github.com/DanielOX/Data-Engineering-Workflow-DUCKDB/raw/main/nyc_taxi_trip_data/2023/2023.parquet')

# display dataframe
display(df.head())

# meta info on dataframe - df.shape outputs (rows, columns). 
print(f'Dataset Contains: {df.shape[0]} rows,  {df.shape[1]} columns')

# Will not be using in this article, code written here just for reference only
# Continue from the below code block
def genMd5Hash(*args) -> str:
    string = ''.join([str(arg) for arg in args])

    """ Function to generate md5 when given a string """

    return hashlib.md5(string.encode())

df['trip_id'] = df.apply(genMd5Hash,axis=1)

# Just adding for sequence. the trip_id columns would be located at the end, so i will filtered out all the cols except `trip_id`

cols = list(filter(lambda x: x != 'trip_id', df.columns))

# combining array in this way ensures `trip_id` is always on the first columns, just for readability purposes

cols = ['trip_id'] + cols

df[cols].head()

# this will drop duplicates

df = df.drop_duplicates()

# It will insert `trip_id` to the first location in columns

df.insert(0,'trip_id',df.index)

datetime_dim = df[['tpep_pickup_datetime','tpep_dropoff_datetime']].reset_index(drop=True)

# PK
datetime_dim['datetime_id'] = datetime_dim.index


# pickup  datetime
datetime_dim['pick_hour'] = datetime_dim['tpep_pickup_datetime'].dt.hour
datetime_dim['pick_day'] = datetime_dim['tpep_pickup_datetime'].dt.day_name()
datetime_dim['pick_month'] = datetime_dim['tpep_pickup_datetime'].dt.month
datetime_dim['pick_year'] = datetime_dim['tpep_pickup_datetime'].dt.year
datetime_dim['pick_weekday'] = datetime_dim['tpep_pickup_datetime'].dt.weekday

# drop off datetime
datetime_dim['drop_hour'] = datetime_dim['tpep_dropoff_datetime'].dt.hour
datetime_dim['drop_day'] = datetime_dim['tpep_dropoff_datetime'].dt.day_name()
datetime_dim['drop_month'] = datetime_dim['tpep_dropoff_datetime'].dt.month
datetime_dim['drop_year'] = datetime_dim['tpep_dropoff_datetime'].dt.year
datetime_dim['drop_weekday'] = datetime_dim['tpep_dropoff_datetime'].dt.weekday

# giving proper ordering to columns for readability purpose

cols_seq = ['datetime_id', 'tpep_pickup_datetime', 'pick_hour', 'pick_day', 'pick_month', 'pick_year', 'pick_weekday',
             'tpep_dropoff_datetime', 'drop_hour', 'drop_day', 'drop_month', 'drop_year', 'drop_weekday']
datetime_dim = datetime_dim[cols_seq]

display(datetime_dim)

passenger count dim
passenger_count_dim = df[['passenger_count']].reset_index(drop=True)
passenger_count_dim['passenger_count_id'] = passenger_count_dim.index

# giving proper ordering to columns for readability purpose
cols_seq = ['passenger_count_id','passenger_count']
passenger_count_dim = passenger_count_dim[cols_seq]

display(passenger_count_dim)

trip distance dimension
trip_distance_dim = df[['trip_distance']].reset_index(drop=True)
trip_distance_dim['trip_distance_id'] = trip_distance_dim.index


# giving proper ordering to columns for readability purpose
cols_seq = ['trip_distance_id', 'trip_distance']
trip_distance_dim = trip_distance_dim

display(trip_distance_dim)

# Mapping of rate code 
rate_code_type = {
    1:"Standard rate",
    2:"JFK",
    3:"Newark",
    4:"Nassau or Westchester",
    5:"Negotiated fare",
    6:"Group ride"
}

rate_code_dim = df[['RatecodeID']].drop_duplicates().reset_index(drop=True)
rate_code_dim['rate_code_id'] = rate_code_dim.index

# map `id` to the name of rate type
rate_code_dim['rate_code_name'] = rate_code_dim['RatecodeID'].map(rate_code_type)

# giving proper ordering to columns for readability purpose

cols_seq = ['rate_code_id','RatecodeID','rate_code_name']
rate_code_dim = rate_code_dim[cols_seq]

display(rate_code_dim)

# loading lookup table
lookup_location = pd.read_csv('https://raw.githubusercontent.com/DanielOX/Data-Engineering-Workflow-DUCKDB/main/nyc_taxi_trip_data/taxi%2B_zone_lookup.csv')[['LocationID','Zone']].drop_duplicates().reset_index(drop=True)
location_dim = lookup_location.rename(columns={'LocationID':'location_id','Zone':'location'})
location_dim.head()

# preparing dimension table
pickup_location_dim = df[['PULocationID']].reset_index(drop=True)
pickup_location_dim['pickup_location_id'] = pickup_location_dim.index
cols_select = ['pickup_location_id','location']

# Join `pickup_location_dim` with `location_dim`
pickup_location_dim = pickup_location_dim.merge(location_dim, left_on='PULocationID', right_on='location_id')[cols_select]

# giving proper ordering to columns for readability purpose
cols_seq = ['pickup_location_id','location']
pickup_location_dim = pickup_location_dim[cols_seq] 

#rename `location` to `pickup_location` for good readability 
pickup_location_dim = pickup_location_dim.rename(columns={'location':'pickup_location'})


display(pickup_location_dim)

# preparing dimension table
dropoff_location_dim = df[['DOLocationID']].reset_index(drop=True)
dropoff_location_dim['dropoff_location_id'] = dropoff_location_dim.index
cols_select = ['dropoff_location_id','location']

# Join `dropoff_location_dim` with `location_dim`
dropoff_location_dim = dropoff_location_dim.merge(location_dim, left_on='DOLocationID', right_on='location_id')[cols_select]

# giving proper ordering to columns for readability purpose
cols_seq = ['dropoff_location_id','location']
dropoff_location_dim = dropoff_location_dim[cols_seq] 

#rename `location` to `pickup_location` for good readability 
dropoff_location_dim = dropoff_location_dim.rename(columns={'location':'dropff_location'})


display(dropoff_location_dim)

payment_type_name = {
    1:"Credit card",
    2:"Cash",
    3:"No charge",
    4:"Dispute",
    5:"Unknown",
    6:"Voided trip"
}
payment_type_dim = df[['payment_type']].drop_duplicates().reset_index(drop=True)
payment_type_dim['payment_type_id'] = payment_type_dim.index
payment_type_dim['payment_type_name'] = payment_type_dim['payment_type'].map(payment_type_name)

# giving proper ordering to columns for readability purpose
cols_seq = ['payment_type_id','payment_type','payment_type_name']
payment_type_dim = payment_type_dim[cols_seq]

display(payment_type_dim)

fact_table = df.merge(passenger_count_dim, left_on='trip_id', right_on='passenger_count_id', how='left') \
               .merge(trip_distance_dim, left_on='trip_id', right_on='trip_distance_id', how='left') \
               .merge(rate_code_dim, left_on='RatecodeID',right_on='RatecodeID',how='left') \
               .merge(pickup_location_dim, left_on='trip_id', right_on='pickup_location_id',how='left') \
               .merge(dropoff_location_dim, left_on='trip_id', right_on='dropoff_location_id',how='left') \
               .merge(datetime_dim, left_on='trip_id', right_on='datetime_id',how='left')\
               .merge(payment_type_dim, left_on='payment_type', right_on='payment_type',how='left') 


cols_required = ['trip_id','VendorID', 'datetime_id', 'passenger_count_id',
               'trip_distance_id', 'rate_code_id', 'store_and_fwd_flag', 'pickup_location_id', 'dropoff_location_id',
               'payment_type_id', 'fare_amount', 'extra', 'mta_tax', 'tip_amount', 'tolls_amount',
               'improvement_surcharge', 'total_amount']

fact_table = fact_table[cols_required]

display(fact_table.head())

import duckdb

def run(q: str = ""):
    """ A function takes SQL Query and executes it on dataframe"""
    out = duckdb.query(q)
    display(out)
    return out

# What is the average trip distance? 

q = """
    SELECT round(avg(trip_distance),2) as average_distance
    FROM trip_distance_dim
"""
run(q)

-- output
┌──────────────────┐
│ average_distance │
│      double      │
├──────────────────┤
│             3.85 │
└──────────────────┘

# What is the most common payment type? 

TOP_K = 3
q = f"""
    WITH PAYMENT_NAME_AGG_COUNTS AS (
         SELECT payment_type, payment_type_name , count(*) as count_
        FROM payment_type_dim 
        JOIN fact_table ON fact_table.payment_type_id = payment_type_dim.payment_type_id
        GROUP BY payment_type_dim.payment_type, payment_type_dim.payment_type_name 
    ), PAYMENT_RANK AS (
        SELECT *, ROW_NUMBER() OVER(ORDER BY count_ DESC) as rnk
        FROM PAYMENT_NAME_AGG_COUNTS
    ) SELECT payment_type_name, count_ FROM PAYMENT_RANK WHERE rnk <= {TOP_K}
"""
run(q)

-- output
┌───────────────────┬─────────┐
│ payment_type_name │ count_  │
│      varchar      │  int64  │
├───────────────────┼─────────┤
│ Credit card       │ 2411462 │
│ Cash              │  532241 │
│ NULL              │   71743 │
└───────────────────┴─────────┘

# - Average trip distance for trips with a tip_amount greater than $10, excluding trips with a payment type of 'Cash'.
q = """
    WITH TRIP_DISTANCE_BASE AS (
        SELECT round(avg(trip_distance_dim.trip_distance)) as 'average_trip_distance(miles)'
        FROM fact_table
        JOIN trip_distance_dim ON  fact_table.trip_id = trip_distance_dim.trip_distance_id
        JOIN payment_type_dim  ON  fact_table.payment_type_id = payment_type_dim.payment_type_id
        WHERE fact_table.tip_amount > 10 and lower(payment_type_dim.payment_type_name) = 'cash'
    ) select * from TRIP_DISTANCE_BASE
"""
run(q)

-- output
┌──────────────────────────────┐
│ average_trip_distance(miles) │
│            double            │
├──────────────────────────────┤
│                         15.0 │
└──────────────────────────────┘

# How many trips were made by each VendorID?

q = """
    SELECT VendorID, count(*) as total_trips
    FROM fact_table
    GROUP BY VendorID
    ORDER BY count(*) DESC
"""
run(q)

-- output
┌──────────┬─────────────┐
│ VendorID │ total_trips │
│  int64   │    int64    │
├──────────┼─────────────┤
│        2 │     2239399 │
│        1 │      827367 │
└──────────┴─────────────┘

# Which vendor has the highest average fare_amount?

q = """
    WITH VENDOR_FARE_AMOUNT AS (
        SELECT VendorID, round(avg(fare_amount),2) as avg_fare_amount
        FROM fact_table
        GROUP BY VendorID
    ), VENDOR_MAX_AMOUNT_RNK AS (
        SELECT VendorID, avg_fare_amount, row_number() OVER(ORDER BY avg_fare_amount DESC) as RNK
        FROM VENDOR_FARE_AMOUNT
    ) SELECT VendorID, avg_fare_amount FROM VENDOR_MAX_AMOUNT_RNK WHERE RNK = 1
        """
run(q)

-- output
┌──────────┬─────────────────┐
│ VendorID │ avg_fare_amount │
│  int64   │     double      │
├──────────┼─────────────────┤
│        2 │           18.71 │
└──────────┴─────────────────┘

#- What is the average number of passengers for trips that occurred on Sundays?
q = """
    SELECT round(avg(passenger_count_dim.passenger_count),2) as average_passenger_count
    FROM fact_table
    JOIN passenger_count_dim ON passenger_count_dim.passenger_count_id = fact_table.passenger_count_id
    JOIN datetime_dim ON datetime_dim.datetime_id = fact_table.datetime_id
    WHERE lower(datetime_dim.pick_day) = 'sunday' or datetime_dim.drop_day = 'sunday'
"""
print("What is the average number of passengers for trips that occurred on Sundays?")
run(q)

-- output 
┌─────────────────────────┐
│ average_passenger_count │
│         double          │
├─────────────────────────┤
│                    1.45 │
└─────────────────────────┘

#- What are the peak hours for taxi trips based on the pick-up datetime?

print("What are the peak hours for taxi trips based on the pick-up datetime?")
q = """
        WITH TRIPS_BY_HOUR_COUNT AS (
            SELECT datetime_dim.pick_hour, count(*) as total_trips
            FROM fact_table
            JOIN datetime_dim ON datetime_dim.datetime_id = fact_table.datetime_id
            GROUP BY datetime_dim.pick_hour
        ), TRIPS_BY_HOUR_RANK AS (
            SELECT  pick_hour, total_trips, DENSE_RANK() OVER(ORDER BY total_trips) as rnk
           FROM TRIPS_BY_HOUR_COUNT
        ) SELECT pick_hour,total_trips  FROM TRIPS_BY_HOUR_RANK WHERE rnk = 1
"""
run(q)

-- output
┌───────────┬─────────────┐
│ pick_hour │ total_trips │
│   int32   │    int64    │
├───────────┼─────────────┤
│         4 │       17835 │
└───────────┴─────────────┘

# Identify zones with the highest average total_amount and the lowest average trip_distance
q = """
    WITH BASE AS (
        SELECT 
            pickup_location_dim.pickup_location as zone, 
            round(avg(total_amount),2) as avg_total_amount, 
            round(avg(trip_distance_dim.trip_distance_id),2) as avg_trip_distance
        FROM fact_table 
        join trip_distance_dim ON fact_table.trip_id = trip_distance_dim.trip_distance_id
        JOIN pickup_location_dim ON pickup_location_dim.pickup_location_id = fact_table.pickup_location_id
        WHERE pickup_location_dim.pickup_location IS NOT NULL
        GROUP BY pickup_location_dim.pickup_location
    )  SELECT zone, avg_total_amount as avg_highest_trip_amount, avg_trip_distance  as avg_lowest_trip_distance 
       FROM BASE ORDER BY avg_total_amount DESC, avg_trip_distance ASC LIMIT 1
"""
run(q)

-- output
┌────────────────┬─────────────────────────┬──────────────────────────┐
│      zone      │ avg_highest_trip_amount │ avg_lowest_trip_distance │
│    varchar     │         double          │          double          │
├────────────────┼─────────────────────────┼──────────────────────────┤
│ Newark Airport │                  104.38 │               1212392.68 │
└────────────────┴─────────────────────────┴──────────────────────────┘

# Which pickup location has the highest tip_amount amount?
q = """
    WITH BASE AS(

        SELECT 
            pickup_location_dim.pickup_location as zone, 
            fact_table.tip_amount,
            ROW_NUMBER() OVER(ORDER BY fact_table.tip_amount DESC) as rnk
        FROM fact_table 
        JOIN pickup_location_dim ON pickup_location_dim.pickup_location_id = fact_table.pickup_location_id
    ) SELECT zone, tip_amount as highest_tip_amount FROM BASE WHERE rnk = 1
"""
print("Which pickup location has the highest tip_amount amount?")
run(q)

-- output
┌───────────────────────┬────────────────────┐
│         zone          │ highest_tip_amount │
│        varchar        │       double       │
├───────────────────────┼────────────────────┤
│ Upper East Side North │              380.8 │
└───────────────────────┴────────────────────┘

# How many trips were made for each rate code type?
q = """
    SELECT rate_code_dim.rate_code_name, count(*) as trip_count
    FROM fact_table
    JOIN rate_code_dim ON rate_code_dim.RatecodeID = fact_table.rate_code_id
    GROUP BY rate_code_dim.rate_code_name
    ORDER BY trip_count DESC

"""
print("How many trips were made for each rate code type?")
run(q)

-- output
┌───────────────────────┬────────────┐
│    rate_code_name     │ trip_count │
│        varchar        │   int64    │
├───────────────────────┼────────────┤
│ Standard rate         │     114239 │
│ Nassau or Westchester │      15043 │
│ Newark                │      13106 │
│ Negotiated fare       │       8958 │
│ JFK                   │       4366 │
│ Group ride            │          6 │
└───────────────────────┴────────────┘

# What is the average total_amount for each rate code type?
q = """
    SELECT rate_code_dim.rate_code_name, round(avg(fact_table.total_amount),2) as avg_total_amount
    FROM fact_table
    JOIN rate_code_dim ON rate_code_dim.RatecodeID = fact_table.rate_code_id
    GROUP BY rate_code_dim.rate_code_name
    ORDER BY avg_total_amount DESC

"""
print("What is the average total_amount for each rate code type?")
run(q)

-- output
┌───────────────────────┬──────────────────┐
│    rate_code_name     │ avg_total_amount │
│        varchar        │      double      │
├───────────────────────┼──────────────────┤
│ Group ride            │           132.78 │
│ JFK                   │           119.95 │
│ Negotiated fare       │            107.4 │
│ Standard rate         │            89.09 │
│ Nassau or Westchester │            81.39 │
│ Newark                │            37.76 │
└───────────────────────┴──────────────────┘