{ "cells": [ { "cell_type": "markdown", "id": "8a04c046-b7da-44d1-847c-fc9e127c07b2", "metadata": {}, "source": [ "# NYC Taxi dataset anaylsis notebook\n", "\n", "The objective is to perform **exploratory data analysis** on the `NYC Yellow Taxi` dataset to uncover patterns in taxi usage, create visualizations to interpret these patterns, and build a machine learning model to predict trip duration based on features like trip distance and pickup location.\n" ] }, { "cell_type": "markdown", "id": "f9cabeaf-1884-427d-856f-efe60bbfb502", "metadata": { "tags": [] }, "source": [ "## Step 1: Set Up the Environment\n", "\n", "We start by initializing a PySpark session and connecting to the Connectors to access our datasets. Modify the variables at the top to have the correct dataset and project ID.\n", "\n", "- **Note:** Replace `PROJECT_ID` with your actual ID from the URL (https://eu.dataplatform.ovh.net/dpe/#/{PROJECT_ID}/notebooks). Use default_dataset or your custom dataset name." ] }, { "cell_type": "code", "execution_count": null, "id": "7d9e5070-ed5a-49b4-8bc9-d738a051da75", "metadata": {}, "outputs": [], "source": [ "import logging\n", "from forepaas.dwh import connect\n", "from pyspark.sql import SparkSession\n", "from pyspark.sql.functions import col, hour, dayofweek, unix_timestamp, avg, count, sum, when\n", "import matplotlib.pyplot as plt\n", "import seaborn as sns\n", "import pandas as pd\n", " \n", "# Set up variables (dataset, dataplant_url and YEAR, MONTH of the yellow taxi file\n", "DATASET = \"default_dataset\"\n", "PROJECT_ID = \"PROJECT_ID\"\n", "YEAR = \"2025\"\n", "MONTH = \"01\"\n", " \n", "# Set up logging for debugging\n", "logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')\n", " \n", "# Initialize SparkSession\n", "try:\n", " spark = SparkSession.builder.appName(\"NYC_Yellow_Taxi_Analysis\").getOrCreate()\n", " logging.info(f\"Spark Version: {spark.version}\")\n", "except Exception as e:\n", " logging.error(f\"Failed to initialize SparkSession: {e}\")\n", " raise\n", " \n", "# Connect to Lakehouse - default_dataset (default)\n", "try:\n", " cn_prim = connect(f\"dwh/{DATASET}/\")\n", " logging.info(f\"Connected to Lakehouse - Dataset: {DATASET}\")\n", "except Exception as e:\n", " logging.error(f\"Failed to connect to Lakehouse - Dataset: {DATASET} | {e}\")\n", " raise" ] }, { "cell_type": "markdown", "id": "2351ebdd-91ae-4c21-999d-d32f38aa0b0c", "metadata": { "tags": [] }, "source": [ "### Explanation:\n", "\n", "- **Why:** The `SparkSession` is the entry point for PySpark, allowing us to work with DataFrames and perform distributed computations. The `connect` function from `forepaas.dwh` links to the OVHcloud Connectors, which organizes datasets in the Lakehouse.\n", "- **What:** We set up logging to track progress and catch errors. The `appName` helps identify the Spark job in the cluster.\n", "- **Output:** Confirms the Spark version and successful connection to the Lakehouse dataset - default_dataset or other dataset if you create/use another one." ] }, { "cell_type": "markdown", "id": "c726f2c6-7009-4d46-b3d4-49408eee1488", "metadata": {}, "source": [ "## Step 2: Load and Inspect Data\n", "\n", "We load the `Yellow Taxi dataset` and the `Taxi Zone Lookup` table from the default_dataset and inspect their schemas." ] }, { "cell_type": "code", "execution_count": null, "id": "6cfd979f-315a-4b2f-b0c1-fb36b13703ab", "metadata": {}, "outputs": [], "source": [ "# Load datasets (Replace PROJECT_ID by yours, same for default_dataset if you create another one\n", "yellow_df = cn_prim.query(f\"SELECT * FROM db_{PROJECT_ID}_{DATASET}.{DATASET}.yellow_tripdata_{YEAR}_{MONTH}\")\n", "taxi_zones_df = cn_prim.query(f\"SELECT LocationID, Borough, Zone FROM db_{PROJECT_ID}_{DATASET}.{DATASET}.taxi_zone_lookup\")\n", " \n", "# Cache DataFrames for performance\n", "yellow_df.cache()\n", "taxi_zones_df.cache()\n", " \n", "# Verify data loading\n", "logging.info(f\"Yellow Taxi Records: {yellow_df.count()}\")\n", "logging.info(f\"Taxi Zones Records: {taxi_zones_df.count()}\")\n", " \n", "# Inspect schemas\n", "print(\"Yellow Taxi Schema:\")\n", "yellow_df.printSchema()\n", "print(\"Taxi Zones Schema:\")\n", "taxi_zones_df.printSchema()" ] }, { "cell_type": "markdown", "id": "e5d42ed9-60d2-43b8-8cef-70a7d55edab4", "metadata": {}, "source": [ "### Explanation:\n", "\n", "- **Why:** The Connectors stores datasets in a structured format (Parquet for Yellow Taxi, CSV for Taxi Zones). Caching improves performance for repeated operations on large datasets.\n", "- **What:** We load the datasets using SQL queries via `cn_prim.query`. The `count()` method verifies the number of records, and `printSchema()` shows the structure of the data.\n", "- **Output:**\n", " - **Yellow Taxi Records:** ~3,475,226 rows.\n", " - **Taxi Zones Records:** 265 rows.\n", " - **Yellow Taxi Schema:** Includes fields like vendorid, tpep_pickup_datetime, trip_distance, fare_amount, pulocationid, and dolocationid.\n", " - **Taxi Zones Schema:** Includes LocationID, Borough, and Zone." ] }, { "cell_type": "markdown", "id": "2d8a5f07-bfe9-4db8-b2f6-83e588d853ce", "metadata": {}, "source": [ "## Step 3: Clean the Data\n", "We clean the `Yellow Taxi dataset` to remove invalid or incomplete records." ] }, { "cell_type": "code", "execution_count": null, "id": "fa269d63-f9d7-4937-aa0a-a7cc519df6eb", "metadata": {}, "outputs": [], "source": [ "# Clean Yellow Taxi DataFrame\n", "yellow_df_clean = yellow_df.filter(\n", " (col(\"tpep_pickup_datetime\").isNotNull()) &\n", " (col(\"tpep_dropoff_datetime\").isNotNull()) &\n", " (col(\"passenger_count\").isNotNull()) &\n", " (col(\"passenger_count\") > 0) &\n", " (col(\"trip_distance\") > 0) &\n", " (col(\"fare_amount\") > 0)\n", ")\n", " \n", "# Verify cleaned data\n", "logging.info(f\"Cleaned Yellow Taxi Records: {yellow_df_clean.count()}\")" ] }, { "cell_type": "markdown", "id": "afa315a0-e0f3-4d73-b3ad-88c616f3cb2b", "metadata": {}, "source": [ "### Explanation:\n", "\n", "- **Why:** Cleaning removes records with missing or unrealistic values (zero passengers or negative fares) to ensure accurate analysis. Filtering out invalid trips (< 60 seconds, < 0.1 miles)\n", "- **What:** We use filter with conditions to keep only valid records. The `col` function helps reference columns in PySpark.\n", "- **Output:** ~2,816,835 records, indicating ~19% of records were removed due to invalid data." ] }, { "cell_type": "markdown", "id": "3d9e0aa6-0970-4a1e-b308-ba717cc8dcf4", "metadata": {}, "source": [ "## Step 4: Join with Taxi Zones\n", "We join the `Yellow Taxi dataset` with the `Taxi Zone Lookup` table to add geographical context." ] }, { "cell_type": "code", "execution_count": null, "id": "aebf0d5d-5bd2-42aa-ac97-a670727ca2ee", "metadata": {}, "outputs": [], "source": [ "# Join with taxi zones for pickup and dropoff locations\n", "yellow_df_clean = yellow_df_clean.join(taxi_zones_df, yellow_df_clean.pulocationid == taxi_zones_df.LocationID, \"left\") \\\n", " .withColumnRenamed(\"Borough\", \"pickup_borough\") \\\n", " .withColumnRenamed(\"Zone\", \"pickup_zone\") \\\n", " .drop(\"LocationID\")\n", " \n", "yellow_df_clean = yellow_df_clean.join(taxi_zones_df, yellow_df_clean.dolocationid == taxi_zones_df.LocationID, \"left\") \\\n", " .withColumnRenamed(\"Borough\", \"dropoff_borough\") \\\n", " .withColumnRenamed(\"Zone\", \"dropoff_zone\") \\\n", " .drop(\"LocationID\")\n", " \n", "# Filter out invalid zones and boroughs\n", "yellow_df_clean = yellow_df_clean.filter(\n", " (col(\"pickup_zone\") != \"Unknown\") & (col(\"dropoff_zone\") != \"Unknown\") &\n", " (col(\"pickup_borough\") != \"Unknown\") & (col(\"pickup_borough\") != \"N/A\") & (col(\"pickup_borough\").isNotNull()) &\n", " (col(\"dropoff_borough\") != \"Unknown\") & (col(\"dropoff_borough\") != \"N/A\") & (col(\"dropoff_borough\").isNotNull())\n", ")\n", " \n", "# Verify cleaned and joined data\n", "logging.info(f\"Cleaned Yellow Taxi Records after borough filtering: {yellow_df_clean.count()}\")\n", "yellow_df_clean.select(\"pulocationid\", \"pickup_borough\", \"pickup_zone\", \"dolocationid\", \"dropoff_borough\", \"dropoff_zone\").show(5)" ] }, { "cell_type": "markdown", "id": "75c9149e-6cc6-48bd-a7dd-25bd4167bdcc", "metadata": {}, "source": [ "### Explanation:\n", "\n", "- **Why:** Joining with taxi_zones_df maps `pulocationid` and `dolocationid` to human-readable boroughs and zones (Manhattan, Upper East Side).\n", "\n", "- **What:** We perform two left joins to add pickup and dropoff locations, rename columns for clarity, and filter out records with \"Unknown\", \"N/A\", or null boroughs to ensure data quality.\n", "\n", "- **Output:** Displays a sample of joined data:\n", "\n", "| pulocationid | pickup_borough | pickup_zone | dolocationid | dropoff_borough | dropoff_zone |\n", "|--------------|----------------|-----------------------|---------------|------------------|-----------------------|\n", "| 237.0 | Manhattan | Upper East Side South | 140.0 | Manhattan | Lenox Hill East |\n", "| 239.0 | Manhattan | Upper West Side South | 142.0 | Manhattan | Lincoln Square East |\n", "| 140.0 | Manhattan | Lenox Hill East | 236.0 | Manhattan | Upper East Side North |\n", "| 68.0 | Manhattan | East Chelsea | 107.0 | Manhattan | Gramercy |\n", "| 246.0 | Manhattan | West Chelsea/Hudson Yards | 48.0 | Manhattan | Clinton East |\n", "\n", "only showing top 5 rows" ] }, { "cell_type": "markdown", "id": "9bf50471-63a3-4195-82b6-de9ba32e84ad", "metadata": {}, "source": [ "## Step 5: Feature Engineering and Enhanced Data Cleaning\n", "We add derived features and apply additional cleaning to prepare the data for machine learning by removing invalid trips and capping outliers." ] }, { "cell_type": "code", "execution_count": null, "id": "2d330640-b936-4ae9-9bab-7f6ab31412ed", "metadata": {}, "outputs": [], "source": [ "# Add derived features\n", "yellow_df_clean = yellow_df_clean.withColumn(\"pickup_hour\", hour(\"tpep_pickup_datetime\")) \\\n", " .withColumn(\"day_of_week\", dayofweek(\"tpep_pickup_datetime\")) \\\n", " .withColumn(\"trip_duration\", unix_timestamp(\"tpep_dropoff_datetime\") - unix_timestamp(\"tpep_pickup_datetime\")) \\\n", " .withColumn(\"total_revenue\", col(\"fare_amount\") + col(\"tip_amount\") + col(\"congestion_surcharge\") + col(\"airport_fee\"))\n", " \n", "# Inspect trip_duration before additional cleaning\n", "yellow_df_clean.select(\"trip_duration\").summary(\"count\", \"min\", \"max\", \"mean\").show()\n", " \n", "# Check for outliers\n", "logging.info(f\"Trips with duration > 1 hour: {yellow_df_clean.filter(col('trip_duration') > 3600).count()}\")\n", "logging.info(f\"Trips with distance > 50 miles: {yellow_df_clean.filter(col('trip_distance') > 50).count()}\")\n", " \n", "# Filter out invalid trips and cap outliers\n", "yellow_df_clean = yellow_df_clean.filter(\n", " (col(\"trip_duration\") >= 60) & # Minimum 1-minute trip duration\n", " (col(\"trip_distance\") >= 0.1) & # Minimum 0.1 miles\n", " (col(\"trip_distance\").isNotNull()) & # Remove null distances\n", " (col(\"pickup_hour\").isNotNull()) & # Remove null hours\n", " (col(\"tpep_pickup_datetime\").isNotNull()) & # Remove null pickup times\n", " (col(\"tpep_dropoff_datetime\").isNotNull()) & # Remove null dropoff times\n", " (col(\"passenger_count\").isNotNull()) # Ensure passenger_count is not null\n", ")\n", " \n", "# Cap outliers\n", "yellow_df_clean = yellow_df_clean.withColumn(\"trip_duration\", when(col(\"trip_duration\") > 3600, 3600).otherwise(col(\"trip_duration\"))) \\\n", " .withColumn(\"trip_distance\", when(col(\"trip_distance\") > 50, 50).otherwise(col(\"trip_distance\")))\n", " \n", "# Verify cleaned data after filtering and capping\n", "logging.info(f\"Yellow Taxi Records after duration filtering and outlier capping: {yellow_df_clean.count()}\")\n", "yellow_df_clean.select(\"trip_duration\").summary(\"count\", \"min\", \"max\", \"mean\").show()" ] }, { "cell_type": "markdown", "id": "4595ebd9-6dcc-425f-8f85-78d0a3ec076e", "metadata": {}, "source": [ "### Explanation:\n", "\n", "- **Why:**\n", "\n", " - **Data Quality for ML:** Machine learning models require clean, consistent data to produce reliable predictions. Negative or extremely high trip_duration values (-3362 or 337,579 seconds) and unrealistic trip_distance values (>50 miles) indicate data errors or outliers that can skew model training. Filtering short trips (<1 minute) and capping outliers (durations >1 hour, distances >50 miles) ensures the data is realistic for NYC taxi trips.\n", "\n", " - **Feature Engineering:** Derived features like `pickup_hour`, `day_of_week`, `trip_duration`, and `total_revenue` enable temporal and financial analysis and serve as input features for ML models.\n", "\n", "- **What:**\n", "\n", " - We inspect `trip_duration` to identify outliers (28,432 trips >1 hour, 43 trips >50 miles).\n", "\n", " - We add features using `withColumn` for `pickup_hour`, `day_of_week`, `trip_duration`, and `total_revenue`.\n", "\n", " - We filter out trips with durations <60 seconds or distances <0.1 miles, and ensure non-null values.\n", "\n", " - We cap `trip_duration` at 3600 seconds (1 hour) and `trip_distance` at 50 miles to handle outliers.\n", "\n", "- **Output:**\n", "\n", " - Initial trip_duration summary: ~2,792,399 records, min -3362, max 337,579, mean ~901.62 seconds.\n", "\n", " - After filtering and capping: ~2,781,003 records, min 60, max 3600, mean ~867.17 seconds.\n", "\n", " - This confirms the removal of ~1.2% of records and a more realistic dataset for ML." ] }, { "cell_type": "markdown", "id": "6a665e8f-9731-4f69-a2f4-8a02f14ee71a", "metadata": {}, "source": [ "## Step 6: Exploratory Data Analysis (EDA)\n", "We perform EDA to uncover patterns in taxi usage." ] }, { "cell_type": "markdown", "id": "d22371cc-0e7d-4055-9f9b-4c385c6c16b7", "metadata": {}, "source": [ "### Analysis 1: Trips by Hour" ] }, { "cell_type": "code", "execution_count": null, "id": "8cd6197e-bea1-43fd-82d1-0c094a24fd3f", "metadata": {}, "outputs": [], "source": [ "# Group by pickup_hour and count trips\n", "yellow_hourly_trips = yellow_df_clean.groupBy(\"pickup_hour\") \\\n", " .agg(count(\"*\").alias(\"num_trips\")) \\\n", " .orderBy(\"pickup_hour\")\n", "yellow_hourly_trips_pd = yellow_hourly_trips.toPandas()\n", " \n", "# Visualize\n", "plt.figure(figsize=(10, 6))\n", "plt.bar(yellow_hourly_trips_pd[\"pickup_hour\"], yellow_hourly_trips_pd[\"num_trips\"], color=\"skyblue\")\n", "plt.xlabel(\"Hour of Day\")\n", "plt.ylabel(\"Number of Trips\")\n", "plt.title(\"Yellow Taxi Trips per Hour (January 2025)\")\n", "plt.grid(axis='y', linestyle='--', alpha=0.7)\n", "plt.show()" ] }, { "cell_type": "markdown", "id": "0c2241b7-38cc-4e6b-a560-d3b147ce9524", "metadata": {}, "source": [ "#### Explanation:\n", "\n", "- **Why:** This analysis shows how taxi demand varies by hour, helping identify peak travel times (rush hours).\n", "- **What:** We group by `pickup_hour`, `count trips`, and convert to a Pandas DataFrame for visualization with Matplotlib.\n", "- **Result:** The bar plot likely shows peaks during morning (7–9 AM) and evening (5–7 PM) rush hours, reflecting commuting patterns in NYC. Nighttime hours (2–4 AM) have fewer trips due to lower demand." ] }, { "cell_type": "markdown", "id": "c8f4cb65-18dc-416f-832f-b5a165f8c57d", "metadata": {}, "source": [ "### Analysis 2: Average Trip Duration by Day of Week" ] }, { "cell_type": "code", "execution_count": null, "id": "2086ce9e-45b7-4323-82e9-ca8b18b2f069", "metadata": {}, "outputs": [], "source": [ "# Group by day_of_week and calculate average trip duration\n", "yellow_duration_by_day = yellow_df_clean.groupBy(\"day_of_week\") \\\n", " .agg(avg(\"trip_duration\").alias(\"avg_duration_seconds\")) \\\n", " .orderBy(\"day_of_week\")\n", "yellow_duration_by_day_pd = yellow_duration_by_day.toPandas()\n", "yellow_duration_by_day_pd[\"avg_duration_minutes\"] = yellow_duration_by_day_pd[\"avg_duration_seconds\"] / 60\n", " \n", "# Visualize\n", "plt.figure(figsize=(10, 6))\n", "plt.plot(yellow_duration_by_day_pd[\"day_of_week\"], yellow_duration_by_day_pd[\"avg_duration_minutes\"], marker='o', color='coral')\n", "plt.xlabel(\"Day of Week (1=Sunday, 7=Saturday)\")\n", "plt.ylabel(\"Average Trip Duration (Minutes)\")\n", "plt.title(\"Average Yellow Taxi Trip Duration by Day of Week (January 2025)\")\n", "plt.grid(True)\n", "plt.show()" ] }, { "cell_type": "markdown", "id": "8411b717-136d-4bb1-96fa-f2f681fbc4a2", "metadata": {}, "source": [ "#### Explanation:\n", "\n", "- **Why:** Trip duration varies by day due to differences in traffic or trip purpose.\n", "- **What:** We calculate the average `trip_duration` per day of the week and convert seconds to minutes for readability.\n", "- **Result:** The line plot may show longer trips on weekdays and less traffic the weekends." ] }, { "cell_type": "markdown", "id": "3e9e6f96-8fba-4997-9a3c-723895850b59", "metadata": {}, "source": [ "### Analysis 3: Revenue by Borough" ] }, { "cell_type": "code", "execution_count": null, "id": "cc49b90b-66bc-497c-8f3b-b2da92307e5b", "metadata": {}, "outputs": [], "source": [ "# Group by pickup_borough and calculate average revenue\n", "yellow_revenue_by_borough = yellow_df_clean.groupBy(\"pickup_borough\") \\\n", " .agg(avg(\"total_revenue\").alias(\"avg_revenue\"), count(\"*\").alias(\"num_trips\")) \\\n", " .orderBy(\"avg_revenue\", ascending=False)\n", "yellow_revenue_by_borough_pd = yellow_revenue_by_borough.toPandas()\n", " \n", "# Visualize\n", "plt.figure(figsize=(10, 6))\n", "sns.barplot(data=yellow_revenue_by_borough_pd, x=\"avg_revenue\", y=\"pickup_borough\", hue=\"pickup_borough\", palette=\"Blues_d\")\n", "plt.xlabel(\"Average Revenue ($)\")\n", "plt.ylabel(\"Pickup Borough\")\n", "plt.title(\"Average Yellow Taxi Revenue by Pickup Borough (January 2025)\")\n", "plt.show()" ] }, { "cell_type": "markdown", "id": "382bff5e-bc13-4067-8a42-ce2e0278178d", "metadata": {}, "source": [ "#### Explanation:\n", "\n", "- **Why:** Revenue analysis helps identify which boroughs generate the most income, useful for taxi operators or urban planners.\n", "- **What:** We calculate the average `total_revenue` per borough and visualize it with Seaborn for a cleaner presentation.\n", "- **Result:** Boroughs like Queens and EWR (Newark Airport) likely show higher average revenue due to longer trips (airport rides). In analysis 5, we will check what is the boroughs with the most trips." ] }, { "cell_type": "markdown", "id": "dae3648f-ce45-4803-96e2-8c1cbdc50e93", "metadata": {}, "source": [ "### Analysis 4: Tipping Behavior for Credit Card Payments" ] }, { "cell_type": "code", "execution_count": null, "id": "b7244017-bbe7-4879-b383-df03453d8109", "metadata": {}, "outputs": [], "source": [ "# Filter for credit card payments (payment_type = 1)\n", "yellow_credit_tips = yellow_df_clean.filter(col(\"payment_type\") == 1) \\\n", " .groupBy(\"pickup_borough\") \\\n", " .agg(avg(\"tip_amount\").alias(\"avg_tip\"), count(\"*\").alias(\"num_tipped_trips\")) \\\n", " .orderBy(\"avg_tip\", ascending=False)\n", "yellow_credit_tips_pd = yellow_credit_tips.toPandas()\n", " \n", "# Visualize\n", "plt.figure(figsize=(10, 6))\n", "sns.barplot(data=yellow_credit_tips_pd, x=\"avg_tip\", y=\"pickup_borough\", hue=\"pickup_borough\", palette=\"Greens_d\")\n", "plt.xlabel(\"Average Tip ($)\")\n", "plt.ylabel(\"Pickup Borough\")\n", "plt.title(\"Average Tip Amount for Credit Card Payments by Pickup Borough (January 2025)\")\n", "plt.show()" ] }, { "cell_type": "markdown", "id": "4d3bd398-80f7-4554-889a-67fa45948bd1", "metadata": {}, "source": [ "#### Explanation:\n", "\n", "- **Why:** Tipping behavior varies by borough and is only recorded for credit card payments (payment_type = 1).\n", "- **What:** We filter for credit card payments, calculate average tips per borough, and visualize the results.\n", "- **Result:** EWR and Queens likely show higher tips due to longer trips (airport rides). Bronx lower tips reflect shorter urban trips where tipping is less common." ] }, { "cell_type": "markdown", "id": "349a766a-65ea-465a-949c-2d94c58a48bf", "metadata": {}, "source": [ "### Analysis 5: Identifying the Zone with the Most Trips" ] }, { "cell_type": "code", "execution_count": null, "id": "d4fed1ad-058d-4287-8c12-d8a0729f30a5", "metadata": {}, "outputs": [], "source": [ "# Analysis 5: Trips by Pickup Zone\n", "yellow_zone_trips = yellow_df_clean.groupBy(\"pickup_zone\") \\\n", " .agg(count(\"*\").alias(\"num_trips\")) \\\n", " .orderBy(\"num_trips\", ascending=False)\n", " \n", "# Limit to top 10 zones for visualization\n", "yellow_zone_trips_top10 = yellow_zone_trips.limit(10)\n", "yellow_zone_trips_top10_pd = yellow_zone_trips_top10.toPandas()\n", " \n", "# Visualize\n", "plt.figure(figsize=(12, 6))\n", "sns.barplot(data=yellow_zone_trips_top10_pd, x=\"num_trips\", y=\"pickup_zone\", hue=\"pickup_zone\", palette=\"Purples_d\")\n", "plt.xlabel(\"Number of Trips\")\n", "plt.ylabel(\"Pickup Zone\")\n", "plt.title(\"Top 10 Yellow Taxi Pickup Zones by Number of Trips (January 2025)\")\n", "plt.show()\n", " \n", "# Log the top zone\n", "top_zone = yellow_zone_trips_top10_pd.iloc[0][\"pickup_zone\"]\n", "top_zone_trips = yellow_zone_trips_top10_pd.iloc[0][\"num_trips\"]\n", "logging.info(f\"Zone with the most trips: {top_zone} with {top_zone_trips} trips\")" ] }, { "cell_type": "markdown", "id": "f992eb9c-e013-4605-aac8-a08b6ba13a38", "metadata": {}, "source": [ "#### Explanation\n", "- **Why:** Grouping by pickup_zone and counting trips reveals which specific areas (Upper East Side, Midtown) have the highest taxi demand. Limiting to the top 10 zones keeps the visualization manageable, as there are 265 zones in the taxi_zone_lookup table.\n", "- **What:**\n", " - The `groupBy(\"pickup_zone\")` operation aggregates trips by pickup zone.\n", " - The `count(\"*\").alias(\"num_trips\")` counts the number of trips per zone.\n", " - The `orderBy(\"num_trips\", ascending=False)` sorts zones by trip count in descending order.\n", " - The `limit(10)` selects the top 10 zones to avoid cluttering the visualization.\n", " - The data is converted to a Pandas DataFrame `(toPandas())` for visualization with Seaborn’s barplot, which is ideal for categorical data.\n", " - The `Purples_d` palette provides a visually appealing gradient.\n", " - Logging the top zone provides a clear summary of the result.\n", "- **Result:** \n", " - **Expected Top Zones:** Based on historical NYC taxi data, zones like Upper East Side, Midtown, Times Square often have the most trips due to their status as residential, business, or tourist hubs. For example:\n", " - **Upper East Side:** A dense residential area with high-income residents who frequently use taxis.\n", " - **Midtown:** Commercial and tourist areas with heavy foot traffic.\n", " - **Airport Zones (JFK or LaGuardia):** May appear if airport trips are common, though they typically have fewer trips but higher revenue." ] }, { "cell_type": "markdown", "id": "ff01d6f1-d686-41ee-8e8d-73ca3aa301a9", "metadata": {}, "source": [ "### Step 7: Machine Learning Model to Predict Trip Duration\n", "In this step, we build and compare three machine learning models - Linear Regression, Random Forest Regressor, and Gradient Boosting Trees (GBT) - to predict the trip_duration of Yellow Taxi trips using features like `trip_distance`, `pickup_hour`, `day_of_week`, and `pickup_borough`. This analysis is valuable for taxi operators to optimize scheduling, estimate fares, or improve operational efficiency. We preprocess the data, train the models, evaluate their performance using Root Mean Squared Error (RMSE) and R-squared metrics, and interpret feature importance to understand which factors most influence trip duration.\n", "\n", "**Note:** This step may take a few minutes to execute, depending on the number of Data Processing Units (DPUs) allocated to your Jupyter notebook." ] }, { "cell_type": "code", "execution_count": null, "id": "c85982cc-37c7-4274-ad88-ddc410531430", "metadata": {}, "outputs": [], "source": [ "from pyspark.ml.feature import StringIndexer, VectorAssembler, StandardScaler\n", "from pyspark.ml.regression import LinearRegression, RandomForestRegressor, GBTRegressor\n", "from pyspark.ml.evaluation import RegressionEvaluator\n", "import logging\n", " \n", "# Step 7.1: Prepare Features\n", "# Drop pickup_borough_index if it already exists to avoid conflicts\n", "if \"pickup_borough_index\" in yellow_df_clean.columns:\n", " yellow_df_clean = yellow_df_clean.drop(\"pickup_borough_index\")\n", " logging.info(\"Dropped existing pickup_borough_index column to avoid conflict.\")\n", " \n", "# Encode pickup_borough as a numerical feature\n", "indexer = StringIndexer(inputCol=\"pickup_borough\", outputCol=\"pickup_borough_index\")\n", "yellow_df_clean = indexer.fit(yellow_df_clean).transform(yellow_df_clean)\n", " \n", "# Assemble initial feature vector (excluding pickup_borough_index for scaling)\n", "assembler_initial = VectorAssembler(inputCols=[\"trip_distance\", \"pickup_hour\", \"day_of_week\"], outputCol=\"features_initial\")\n", "data_initial = assembler_initial.transform(yellow_df_clean)\n", " \n", "# Standardize features\n", "scaler = StandardScaler(inputCol=\"features_initial\", outputCol=\"features_scaled\", withStd=True, withMean=True)\n", "scaler_model = scaler.fit(data_initial)\n", "data_scaled = scaler_model.transform(data_initial).drop(\"features_initial\")\n", " \n", "# Combine scaled features with pickup_borough_index\n", "assembler_final = VectorAssembler(inputCols=[\"features_scaled\", \"pickup_borough_index\"], outputCol=\"features\")\n", "data = assembler_final.transform(data_scaled).select(\"features\", \"trip_duration\")\n", " \n", "# Step 7.2: Split Data into Training and Test Sets\n", "train_data, test_data = data.randomSplit([0.8, 0.2], seed=42)\n", "logging.info(f\"Training data records: {train_data.count()}\")\n", "logging.info(f\"Test data records: {test_data.count()}\")\n", " \n", "# Step 7.3: Train and Compare Models\n", "# Linear Regression\n", "lr = LinearRegression(featuresCol=\"features\", labelCol=\"trip_duration\")\n", "lr_model = lr.fit(train_data)\n", "lr_predictions = lr_model.transform(test_data)\n", " \n", "# Random Forest Regressor\n", "rf = RandomForestRegressor(featuresCol=\"features\", labelCol=\"trip_duration\", numTrees=100, seed=42)\n", "rf_model = rf.fit(train_data)\n", "rf_predictions = rf_model.transform(test_data)\n", " \n", "# Gradient Boosting Trees\n", "gbt = GBTRegressor(featuresCol=\"features\", labelCol=\"trip_duration\", maxIter=50, seed=42)\n", "gbt_model = gbt.fit(train_data)\n", "gbt_predictions = gbt_model.transform(test_data)\n", " \n", "# Step 7.4: Evaluate Models\n", "evaluator = RegressionEvaluator(labelCol=\"trip_duration\", predictionCol=\"prediction\", metricName=\"rmse\")\n", "evaluator_r2 = RegressionEvaluator(labelCol=\"trip_duration\", predictionCol=\"prediction\", metricName=\"r2\")\n", " \n", "# Evaluate Linear Regression\n", "lr_rmse = evaluator.evaluate(lr_predictions)\n", "lr_r2 = evaluator_r2.evaluate(lr_predictions)\n", "logging.info(f\"Linear Regression - Root Mean Squared Error (RMSE): {lr_rmse}\")\n", "logging.info(f\"Linear Regression - R-squared: {lr_r2}\")\n", " \n", "# Evaluate Random Forest\n", "rf_rmse = evaluator.evaluate(rf_predictions)\n", "rf_r2 = evaluator_r2.evaluate(rf_predictions)\n", "logging.info(f\"RandomForest - Root Mean Squared Error (RMSE): {rf_rmse}\")\n", "logging.info(f\"RandomForest - R-squared: {rf_r2}\")\n", " \n", "# Evaluate Gradient Boosting\n", "gbt_rmse = evaluator.evaluate(gbt_predictions)\n", "gbt_r2 = evaluator_r2.evaluate(gbt_predictions)\n", "logging.info(f\"GBT - Root Mean Squared Error (RMSE): {gbt_rmse}\")\n", "logging.info(f\"GBT - R-squared: {gbt_r2}\")\n", " \n", "# Show sample predictions for all models\n", "logging.info(\"Linear Regression Predictions:\")\n", "lr_predictions.select(\"features\", \"trip_duration\", \"prediction\").show(5)\n", "logging.info(\"RandomForest Predictions:\")\n", "rf_predictions.select(\"features\", \"trip_duration\", \"prediction\").show(5)\n", "logging.info(\"GBT Predictions:\")\n", "gbt_predictions.select(\"features\", \"trip_duration\", \"prediction\").show(5)\n", " \n", "# Check feature importance for RandomForest\n", "logging.info(f\"RandomForest Feature Importances: {rf_model.featureImportances}\")" ] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (ipykernel)", "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.11.6" } }, "nbformat": 4, "nbformat_minor": 5 }