NYC_Taxi_Data.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# **NYC Taxi Data Engineering Pipeline (Yellow Taxi Trip Records)**\n",
    "\n",
    "This project outlines the steps needed to build a comprehensive data engineering pipeline using NYC taxi data from the years 2022 and 2023. This pipeline will involve extracting, transforming, and loading (ETL) data into a Snowflake database followed by creating a dashboard for visualization. The goal is to consolidate, clean, transform, and store large volumes of taxi trip data in a Snowflake database and to create a dashboard for visualizing insights from the data. The dataset comprises monthly PARQUET files containing detailed records of yellow taxi trips in New York City. The pipeline will automate the process of merging these files, cleaning and transforming the data, and loading it into a database for easy querying and analysis.\n",
    "\n",
    "## Problem Statement\n",
    "\n",
    "The primary challenge is to efficiently handle and process large volumes of taxi trip data stored in monthly PARQUET files. The project must address the following issues:\n",
    "\n",
    "- **Data Consolidation:** Merging monthly files into a single comprehensive dataset for each year.\n",
    "- **Data Cleaning:** Ensuring the data is free from errors, missing values and inconsistencies.\n",
    "- **Data Transformation:** Structuring the data to be suitable for analysis and visualisation.\n",
    "- **Database Storage:** Efficiently storing the cleaned and transformed data in a Snowflake database.\n",
    "- **Testing with Another Year (2023):** Once the pipeline is established for 2022, it can be tested with the 2023 data for validation.\n",
    "- **Visualization:** Creating a dashboard to provide insights and facilitate decision-making based on the taxi trip data.\n",
    "\n",
    "## Key Steps in the Project\n",
    "\n",
    "### Step 1: Tools and Environment Setup\n",
    "\n",
    "1. **Install Necessary Python Libraries:**\n",
    "   - Install `pandas`, `pyarrow`, `snowflake-connector-python` and `snowflake.sqlalchemy`\n",
    "   - Set up Snowflake for data storage and database management.\n",
    "   - Configure the working environment using VS Code.\n",
    "\n",
    "3. **Configure Working Environment:**\n",
    "   - Use VS Code for writing and running scripts.\n",
    "   - Set up a virtual environment for dependency management.\n",
    "\n",
    "### Step 2: Extract and Transform data from Parquet files into a Pandas DataFrame\n",
    "\n",
    "1. **Define Parquet File Paths:**\n",
    "   - Defined the paths to the Parquet files for each month of 2022\n",
    "\n",
    "2. **Read and Merge Parquet Files:**\n",
    "   - Parquet file using pyarrow and appending the resulting DataFrames to a list (data_frames)\n",
    "   - Concatenate these DataFrames into a single DataFrame (all_data)\n",
    "\n",
    "3. **Transform Data:**\n",
    "   - Dropping rows with missing data (dropna())\n",
    "   - Renaming columns (rename())\n",
    "   - Converting date/time columns to datetime objects (pd.to_datetime())\n",
    "   - Calculating trip duration in minutes (dt.total_seconds() / 60)\n",
    "\n",
    "### Step 3: Database Storage\n",
    "\n",
    "1. **Database Configuration:**\n",
    "   - Connecting Snowflake with python connector \n",
    "   - Create tables in Snowflake using query executor\n",
    "\n",
    "2. **Load Data into Snowflake Database:**\n",
    "   - Use Snowflake SQLAlchemy's ` and pandas dataframe while making sure of the data intergrity \n",
    "\n",
    "### Step 4: Testing with Another Year (2023)\n",
    "\n",
    "1. **Repeat Data Processing:**\n",
    "   - Apply the same data extraction, cleaning and transformation steps for the 2023 dataset.\n",
    "   - Load the 2023 data into Snowflake to validate the pipeline.\n",
    "\n",
    "### Step 5: Visualisation\n",
    "\n",
    "1. **Create Dashboards:**\n",
    "   - Use tools like Power BI to connect to the Snowflake database.\n",
    "   - Develop visualisations to display key metrics such as total rides per day, revenue and popular pickup and drop-off locations.\n",
    "\n",
    "By following these steps, you will build a robust and efficient data engineering pipeline for NYC Taxi data, enabling comprehensive analysis and visualisation of taxi trip records.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "All libraries installed and imported successfully!\n"
     ]
    }
   ],
   "source": [
    "# Importing libraries\n",
    "import pandas as pd\n",
    "import os\n",
    "import snowflake.connector\n",
    "import pyarrow.parquet as pq\n",
    "from snowflake.connector.pandas_tools import write_pandas\n",
    "\n",
    "print(\"All libraries installed and imported successfully!\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Extract data from Parquet files like shown in the code below \n",
    "parquet_files = [\n",
    "    r\"2022\\yellow_tripdata_2022-01.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-02.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-03.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-04.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-05.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-06.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-07.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-08.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-09.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-10.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-11.parquet\",\n",
    "    r\"2022\\yellow_tripdata_2022-12.parquet\"\n",
    "]\n",
    "\n",
    "data_frames = []\n",
    "\n",
    "for file in parquet_files:\n",
    "    df = pq.read_table(file).to_pandas()\n",
    "    data_frames.append(df)\n",
    "\n",
    "all_data = pd.concat(data_frames, ignore_index=True)\n",
    "\n",
    "# Transform the data\n",
    "all_data = all_data.dropna()  # Remove rows with missing data\n",
    "all_data = all_data.rename(columns={'tpep_pickup_datetime': 'pickup_datetime', 'tpep_dropoff_datetime': 'dropoff_datetime'})\n",
    "all_data['pickup_datetime'] = pd.to_datetime(all_data['pickup_datetime'])\n",
    "all_data['dropoff_datetime'] = pd.to_datetime(all_data['dropoff_datetime'])\n",
    "all_data['trip_duration'] = all_data['dropoff_datetime'] - all_data['pickup_datetime']\n",
    "all_data['trip_duration_minutes'] = all_data['trip_duration'].dt.total_seconds() / 60"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Snowflake connection parameters\n",
    "snowflake_account = 'pvb86519.us-east-1'\n",
    "snowflake_user = 'Elfeenah413'\n",
    "snowflake_password = '@Nafeesah413'\n",
    "snowflake_warehouse = 'NYC_DATA_PIPELINE'\n",
    "snowflake_database = 'NYC_TAXI_DB'  # Specify the desired database name\n",
    "snowflake_schema = 'PUBLIC'  # Specify the desired Schema name\n",
    "\n",
    "# Create a Snowflake connection\n",
    "conn = snowflake.connector.connect(\n",
    "    user=snowflake_user,\n",
    "    password=snowflake_password,\n",
    "    account=snowflake_account,\n",
    "    warehouse=snowflake_warehouse,\n",
    "    database=snowflake_database,\n",
    "    schema=snowflake_schema\n",
    ")\n",
    "\n",
    "# Create a cursor\n",
    "cur = conn.cursor()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<snowflake.connector.cursor.SnowflakeCursor at 0x1f1d9825520>"
      ]
     },
     "execution_count": 4,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Table creation query (adjust columns and types as per your DataFrame)\n",
    "\n",
    "# Create the database if it doesn't exist\n",
    "cur.execute(f\"CREATE DATABASE IF NOT EXISTS {snowflake_database}\")\n",
    "\n",
    "# Switch to the specified database\n",
    "cur.execute(f\"USE DATABASE {snowflake_database}\")\n",
    "\n",
    "# Create schema if it doesn't exist\n",
    "cur.execute(f\"CREATE SCHEMA IF NOT EXISTS {snowflake_schema}\")\n",
    "\n",
    "# Use the specified schema\n",
    "cur.execute(f\"USE SCHEMA {snowflake_schema}\")\n",
    "\n",
    "# Create the table if it doesn't exist\n",
    "cur.execute(\"\"\"\n",
    "    CREATE TABLE IF NOT EXISTS taxi_trips (\n",
    "        pickup_datetime TIMESTAMP,\n",
    "        dropoff_datetime TIMESTAMP,\n",
    "        passenger_count INTEGER,\n",
    "        trip_distance FLOAT,\n",
    "        store_and_fwd_flag STRING,\n",
    "        payment_type INTEGER,\n",
    "        fare_amount FLOAT,\n",
    "        extra FLOAT,\n",
    "        mta_tax FLOAT,\n",
    "        tip_amount FLOAT,\n",
    "        tolls_amount FLOAT,\n",
    "        congestion_surcharge FLOAT,\n",
    "        improvement_surcharge FLOAT,\n",
    "        airport_fee FLOAT,\n",
    "        total_amount FLOAT,\n",
    "        VendorID FLOAT,\n",
    "        RatecodeID FLOAT,\n",
    "        PULocationID FLOAT,\n",
    "        DOLocationID FLOAT,\n",
    "        trip_duration FLOAT,\n",
    "        trip_duration_minutes FLOAT\n",
    "    )\n",
    "\"\"\")\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Successfully loaded 38287795 rows into Snowflake.\n"
     ]
    }
   ],
   "source": [
    "# Clear existing data if needed\n",
    "# cur.execute(\"TRUNCATE TABLE taxi_trips\")\n",
    "\n",
    "# Begin a transaction for data loading\n",
    "cur.execute(\"BEGIN\")\n",
    "\n",
    "# Reset the index of the DataFrame\n",
    "all_data.reset_index(drop=True, inplace=True)\n",
    "\n",
    "# Load data from DataFrame to Snowflake table using write_pandas\n",
    "success, nchunks, nrows, _ = write_pandas(conn, all_data, 'TAXI_TRIPS', quote_identifiers=False)\n",
    "\n",
    "if success:\n",
    "    # Commit the transaction if the loading was successful\n",
    "    cur.execute(\"COMMIT\")\n",
    "    print(f\"Successfully loaded {nrows} rows into Snowflake.\")\n",
    "else:\n",
    "    # Rollback the transaction if the loading failed\n",
    "    cur.execute(\"ROLLBACK\")\n",
    "    print(\"Loading into Snowflake failed.\")\n",
    "\n",
    "# Close the cursor and connection\n",
    "cur.close()\n",
    "conn.close()\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **To process the data for the year 2023 and load it into Snowflake** ###"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Data loaded successfully 0 rows into TAXI_TRIPS_2023.\n"
     ]
    }
   ],
   "source": [
    "import pandas as pd\n",
    "import pyarrow.parquet as pq\n",
    "import snowflake.connector\n",
    "from snowflake.connector.pandas_tools import write_pandas\n",
    "\n",
    "# List of local paths to 2023 Parquet files\n",
    "parquet_files_2023 = [\n",
    "    r\"2023\\yellow_tripdata_2023-01.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-02.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-03.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-04.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-05.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-06.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-07.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-08.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-09.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-10.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-11.parquet\",\n",
    "    r\"2023\\yellow_tripdata_2023-12.parquet\"\n",
    "]\n",
    "\n",
    "data_frames = []\n",
    "\n",
    "for file in parquet_files_2023:\n",
    "    df = pq.read_table(file).to_pandas()\n",
    "    data_frames.append(df)\n",
    "\n",
    "all_data = pd.concat(data_frames, ignore_index=True)\n",
    "\n",
    "# Transform the data\n",
    "all_data = all_data.dropna()  # Remove rows with missing data\n",
    "all_data = all_data.rename(columns={'tpep_pickup_datetime': 'pickup_datetime', 'tpep_dropoff_datetime': 'dropoff_datetime'})\n",
    "all_data['pickup_datetime'] = pd.to_datetime(all_data['pickup_datetime'])\n",
    "all_data['dropoff_datetime'] = pd.to_datetime(all_data['dropoff_datetime'])\n",
    "all_data['trip_duration'] = all_data['dropoff_datetime'] - all_data['pickup_datetime']\n",
    "all_data['trip_duration_minutes'] = all_data['trip_duration'].dt.total_seconds() / 60\n",
    "\n",
    "# Drop duplicate 'airport_fee' column if it exists\n",
    "if 'Airport_fee' in all_data.columns:\n",
    "    all_data = all_data.drop(columns=['Airport_fee'])\n",
    "\n",
    "# Reset index to ensure it's a RangeIndex\n",
    "all_data = all_data.reset_index(drop=True)\n",
    "\n",
    "# Snowflake connection parameters\n",
    "snowflake_account = 'pvb86519.us-east-1'\n",
    "snowflake_user = 'Elfeenah413'\n",
    "snowflake_password = '@Nafeesah413'\n",
    "snowflake_warehouse = 'NYC_DATA_PIPELINE'\n",
    "snowflake_database = 'NYC_TAXI_DB'\n",
    "snowflake_schema = 'PUBLIC'\n",
    "\n",
    "# Function to create a Snowflake connection\n",
    "def get_snowflake_connection():\n",
    "    return snowflake.connector.connect(\n",
    "        user=snowflake_user,\n",
    "        password=snowflake_password,\n",
    "        account=snowflake_account,\n",
    "        warehouse=snowflake_warehouse,\n",
    "        database=snowflake_database,\n",
    "        schema=snowflake_schema\n",
    "    )\n",
    "\n",
    "# Create a Snowflake connection\n",
    "conn = get_snowflake_connection()\n",
    "cur = conn.cursor()\n",
    "\n",
    "# Create the table if it doesn't exist\n",
    "cur.execute(\"\"\"\n",
    "    CREATE TABLE IF NOT EXISTS TAXI_TRIPS_2023 (\n",
    "        pickup_datetime TIMESTAMP,\n",
    "        dropoff_datetime TIMESTAMP,\n",
    "        passenger_count INTEGER,\n",
    "        trip_distance FLOAT,\n",
    "        store_and_fwd_flag STRING,\n",
    "        payment_type INTEGER,\n",
    "        fare_amount FLOAT,\n",
    "        extra FLOAT,\n",
    "        mta_tax FLOAT,\n",
    "        tip_amount FLOAT,\n",
    "        tolls_amount FLOAT,\n",
    "        congestion_surcharge FLOAT,\n",
    "        improvement_surcharge FLOAT,\n",
    "        airport_fee FLOAT,\n",
    "        total_amount FLOAT,\n",
    "        VendorID FLOAT,\n",
    "        RatecodeID FLOAT,\n",
    "        PULocationID FLOAT,\n",
    "        DOLocationID FLOAT,\n",
    "        trip_duration FLOAT,\n",
    "        trip_duration_minutes FLOAT\n",
    "    )\n",
    "\"\"\")\n",
    "\n",
    "# Load data into Snowflake table using write_pandas\n",
    "try:\n",
    "    success, nchunks, nrows, _ = write_pandas(conn, all_data, 'TAXI_TRIPS_2023', quote_identifiers=False)\n",
    "\n",
    "    if success:\n",
    "        # Commit the transaction if the loading was successful\n",
    "        cur.execute(\"COMMIT\")\n",
    "        print(f\"Data loaded successfully {nrows} rows into TAXI_TRIPS_2023.\")\n",
    "    else:\n",
    "        # Rollback the transaction if the loading failed\n",
    "        cur.execute(\"ROLLBACK\")\n",
    "        print(\"Loading data into TAXI_TRIPS_2023 failed.\")\n",
    "\n",
    "except snowflake.connector.errors.ProgrammingError as e:\n",
    "    print(f\"Snowflake ProgrammingError: {e}\")\n",
    "\n",
    "finally:\n",
    "    # Close the cursor and connection in the finally block\n",
    "    cur.close()\n",
    "    conn.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "               PICKUP_DATETIME             DROPOFF_DATETIME  PASSENGER_COUNT  \\\n",
      "0 -22205-03-19 10:44:09.906176 -22171-02-02 03:10:49.906176                2   \n",
      "1 -22209-07-04 05:44:09.906176 -22193-06-24 13:44:09.906176                1   \n",
      "2 -22172-11-01 12:57:29.906176 -22155-11-19 09:24:09.906176                1   \n",
      "3 -22225-08-07 01:17:29.906176 -22206-09-03 15:30:49.906176                1   \n",
      "4 -22203-05-14 11:37:29.906176 -22132-09-24 06:10:49.906176                1   \n",
      "\n",
      "   TRIP_DISTANCE STORE_AND_FWD_FLAG  PAYMENT_TYPE  FARE_AMOUNT  EXTRA  \\\n",
      "0           3.80                  N             1         14.5    3.0   \n",
      "1           2.10                  N             1          8.0    0.5   \n",
      "2           0.97                  N             1          7.5    0.5   \n",
      "3           1.09                  N             2          8.0    0.5   \n",
      "4           4.30                  N             1         23.5    0.5   \n",
      "\n",
      "   MTA_TAX  TIP_AMOUNT  ...  CONGESTION_SURCHARGE  IMPROVEMENT_SURCHARGE  \\\n",
      "0      0.5        3.65  ...                   2.5                    0.3   \n",
      "1      0.5        4.00  ...                   0.0                    0.3   \n",
      "2      0.5        1.76  ...                   0.0                    0.3   \n",
      "3      0.5        0.00  ...                   2.5                    0.3   \n",
      "4      0.5        3.00  ...                   2.5                    0.3   \n",
      "\n",
      "   AIRPORT_FEE  TOTAL_AMOUNT  VENDORID  RATECODEID  PULOCATIONID  \\\n",
      "0          0.0         21.95       1.0         1.0         142.0   \n",
      "1          0.0         13.30       1.0         1.0         236.0   \n",
      "2          0.0         10.56       2.0         1.0         166.0   \n",
      "3          0.0         11.80       2.0         1.0         114.0   \n",
      "4          0.0         30.30       2.0         1.0          68.0   \n",
      "\n",
      "   DOLOCATIONID  TRIP_DURATION  TRIP_DURATION_MINUTES  \n",
      "0         236.0     1069000000                     18  \n",
      "1          42.0      504000000                      8  \n",
      "2         166.0      538000000                      9  \n",
      "3          68.0      602000000                     10  \n",
      "4         163.0     2252000000                     38  \n",
      "\n",
      "[5 rows x 21 columns]\n"
     ]
    }
   ],
   "source": [
    "# Execute your SQL query\n",
    "query = \"SELECT * FROM TAXI_TRIPS LIMIT 500\"  # Example: Fetch 100 rows\n",
    "cur.execute(query)\n",
    "\n",
    "# Fetch data into a Pandas DataFrame\n",
    "df = cur.fetch_pandas_all()\n",
    "\n",
    "# Close the connection\n",
    "cur.close()\n",
    "conn.close()\n",
    "\n",
    "print(df.head()) "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "venv",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.12.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}