MapReduce Optional Assignment: NYC Yellow Taxi Data Analysis

Assignment Segments

1. Introduction
2. Dataset Description
3. Tasks
   • Data Ingestion
   • MapReduce Programming

---

Introduction

Build an end-to-end big data pipeline on AWS EMR using:

• Hadoop Framework
• Amazon RDS (MySQL/PostgreSQL)
• Apache Sqoop
• Apache HBase
• MRJob for MapReduce

Ingest and analyze NYC TLC Yellow Taxi trip data for 2017 (Jan–Jun) and answer key analytical questions.

---

Dataset Description

CSV files (each several GB) for Jan–Jun 2017:

• yellow_tripdata_2017-01.csv
• yellow_tripdata_2017-02.csv
• yellow_tripdata_2017-03.csv
• yellow_tripdata_2017-04.csv
• yellow_tripdata_2017-05.csv
• yellow_tripdata_2017-06.csv

Data Dictionary
Refer to the NYC TLC Yellow Taxi data dictionary for detailed field definitions.

Note: For initial testing using 2 month's CSV to develop, debug, and validate your ingestion and MapReduce code before running on all files as the files are very large.

---

Tasks

1. Data Ingestion

Task 1: RDS Setup & CSV Load

1. Launch EMR Cluster (single-node m4.xlarge with Hadoop & HBase).
2. Create RDS MySQL instance, open port 3306 for EMR master.
3. SSH into EMR master, download CSVs:

   wget https://nyc-tlc-upgrad.s3.amazonaws.com/yellow_tripdata_2017-01.csv
   wget https://nyc-tlc-upgrad.s3.amazonaws.com/yellow_tripdata_2017-02.csv

4. Connect to RDS:

   mysql -h <RDS_ENDPOINT> -u admin -p

5. Create taxi table with schema matching the data dictionary:

   CREATE TABLE taxi (
     VendorID INT,
     tpep_pickup_datetime DATETIME,
     tpep_dropoff_datetime DATETIME,
     passenger_count INT,
     trip_distance DOUBLE,
     RatecodeID INT,
     store_and_fwd_flag CHAR(1),
     PULocationID INT,
     DOLocationID INT,
     payment_type INT,
     fare_amount DOUBLE,
     extra DOUBLE,
     mta_tax DOUBLE,
     tip_amount DOUBLE,
     tolls_amount DOUBLE,
     improvement_surcharge DOUBLE,
     total_amount DOUBLE,
     congestion_surcharge DOUBLE,
     airport_fee DOUBLE,
     cbd_congestion_fee DOUBLE
   );

6. Load data:

   LOAD DATA LOCAL INFILE '/home/hadoop/yellow_tripdata_2017-01.csv'
     INTO TABLE taxi
     FIELDS TERMINATED BY ','
     IGNORE 1 LINES;
   LOAD DATA LOCAL INFILE '/home/hadoop/yellow_tripdata_2017-02.csv'
     INTO TABLE taxi
     FIELDS TERMINATED BY ','
     IGNORE 1 LINES;

Task 2: RDS → HBase via Sqoop

1. Create HBase table:

   hbase shell
   create 'hbtaxi','td'
   quit

2. Install MySQL connector on EMR master:

   wget https://de-mysql-connector.s3.amazonaws.com/mysql-connector-java-8.0.25.tar.gz
   tar -xzf mysql-connector-java-8.0.25.tar.gz
   sudo cp mysql-connector-java-8.0.25/mysql-connector-java-8.0.25.jar /usr/lib/sqoop/lib/

3. Import from RDS:

   sqoop import      --connect jdbc:mysql://<RDS_ENDPOINT>:3306/yellow      --table taxi      --hbase-table hbtaxi      --column-family td      --hbase-row-key VendorID,tpep_pickup_datetime,tpep_dropoff_datetime      --split-by VendorID      --username admin --password <password>      --num-mappers 5      --hbase-bulkload

4. Verify:

   hbase shell
   count 'hbtaxi'

Task 3: Bulk Import Additional Months

python3 batch_ingest.py   --csv-dir /home/hadoop/yellow   --table hbtaxi --column-family td --hbase-host localhost

---

2. MapReduce Programming

Before running jobs:

sudo pip3 install mrjob
hadoop fs -mkdir /user/hadoop/yellow
hadoop fs -put yellow_tripdata_2017-01.csv /user/hadoop/yellow
# Repeat for other CSVs

Script	Description
mrtask_a.py	Vendor with most trips & total revenue
mrtask_b.py	Pickup location with highest total revenue
mrtask_c.py	Payment type counts, sorted descending
mrtask_d.py	Average trip duration per pickup location
mrtask_e.py	Average tip/revenue ratio per pickup location (sorted)
mrtask_f.py	Avg trip revenue by month, hour (day vs night), and day of week (wknd vs wd)

Example:

python3 vendor_revenue.py   --input hdfs:///user/hadoop/yellow/*.csv   --output-dir results/vendor_revenue