<jupyter_start><jupyter_text><jupyter_code>#Initilaization section for JDK and Spark
!apt-get install openjdk-8-jdk-headless -qq > /dev/null
!wget -q http://mirror.cc.columbia.edu/pub/software/apache/spark/spark-3.0.0-preview2/spark-3.0.0-preview2-bin-hadoop2.7.tgz
!tar xf spark-3.0.0-preview2-bin-hadoop2.7.tgz
!pip install -q findspark
!rm -rf spark-3.0.0-preview2-bin-hadoop2.7.tgz
#Set up Java and Spark environmnet with spark context
import os
os.environ["JAVA_HOME"] = "/usr/lib/jvm/java-8-openjdk-amd64"
os.environ["SPARK_HOME"] = "/content/spark-3.0.0-preview2-bin-hadoop2.7"

 
#setting up Spark Context
import findspark
findspark.init("spark-3.0.0-preview2-bin-hadoop2.7")# SPARK_HOME

from pyspark.sql import SparkSession
from pyspark import SparkContext

spark = SparkSession.builder.master("local[*]").getOrCreate()
sc = SparkContext.getOrCreate()
#download data file ( small size -  Taxi data for 2013) from amazon s3 bucket  as a data set 
!wget https://s3.amazonaws.com/metcs777/taxi-data-sorted-small.csv.bz2

# Load data for  processing for all the tasks
file = "./taxi-data-sorted-small.csv.bz2"
#file = "s3://metcs777/taxi-data-sorted-small.csv.bz2"
# read input text file to RDD
lines = sc.textFile(file)
# splt  the lines of the file with "," separator 
taxis = lines.map(lambda x: x.split(','))

#map columns variables 
taxis_orig_data = taxis.map(lambda p: (p[0], p[1], p[2], p[3], p[4], p[5], p[6], p[7], p[8], p[9],\
                                  p[10], p[11], p[12], p[13], p[14], p[15], p[16]))

#Provide an action to make spark rdd execute and get the count of rows
taxis_orig_data.count()


# Check for bad data and Filter if  Medallion and hack_license( driver id) columns or any other coulmns are null
# refer to the Description of data for columns. 0 - Medallion, 1 - taxi id, 2 - pickup datetime, 4- trip in sec, 5 - trip distance, 12- surcharge
# 16 - total amount
taxis_filt_data = taxis_orig_data.filter( lambda p : False 
                                         if p[0].strip() == '' or  \
                                          p[1].strip() == '' or \
                                          p[2].strip() == '' or p[3].strip() == '' or \
                                          p[4].strip() == '' or p[5].strip() == '' or \
                                          p[6].strip() == '' or p[7].strip() == '' or \
                                          p[8].strip() == '' or p[9].strip() == '' or \
                                          p[10].strip() == '' or p[11].strip() == '' or \
                                          p[12].strip() == '' or p[13].strip() == '' or \
                                          p[14].strip() == '' or p[15].strip() == '' or \
                                          p[16].strip() == '' else  \
                                         True
                                         )

# Count of Rows
taxis_filt_data.count()
#4.1 Task1  Compute Top 10 taxis Drivers using Medallion as key, generate ( medallion, no of drivers ) as pairs
from operator import add
# Filter Duplicate records for Medallion and Driver id as data has multiple trips by the same drivers
driver_data_count = taxis_filt_data.map(lambda p: (p[0], p[1])).distinct()
driver_data_count.count()
from operator import add
# Filter Duplicate records for Medallion and Driver id as data has multiple trips by the same drivers
# Get unique Medallion and Driver ids
driver_data_count = taxis_filt_data.map(lambda p: (p[0], p[1])).distinct()

# Build a map of the Medallion with 1 which points  to a unique driver (1) and reduce by Medallion  using add
fil_driver_data_count = driver_data_count.map(lambda p: (p[0], 1)).reduceByKey(add)

#let us take 5 rows
fil_driver_data_count.take(5)
 

# Get the top 10 of the Medalion , Driver counts using top() function 
fil_driver_data_count.top(10, key = lambda p: p[1])

# Get the top 10 of the Medalion , Driver counts using takeOrdered() function
print(' 4.1 Task Top 10 Medallion Driver pairs: ')
fil_driver_data_count.takeOrdered(10, key = lambda p: -p[1])

# 4.2 Task 2 
# We would like to figure out who the top 10 best drivers are in terms of their average earned money per
# minute spent carrying a customer. The total amount field is the total money earned on a trip. In the end,
# we are interested in computing a set of (driver, money per minute) pairs.
# average earned money per minute spent carrying a customer = (total_amount/duration of trip in secs) * 60
# p[1], p[16]/p[4] * 60

# Build a map of medallion with DriverId, TripTime and Total Paid Amount as a RDD

driver_data_count = taxis_filt_data.map(lambda p: (p[1], (int(p[4]), float(p[16]))))

#Get the count of Rows with multiple Rides forthe drivers
driver_data_count.count()
#Take 5 row sof the data
driver_data_count.take(5)
# Compute the sum of the (duration of trip time in seconds)  and sum of the (total amount by the driver)
driver_data_count1 = taxis_filt_data.map(lambda p: (p[1], (int(p[4]), float(p[16])))).reduceByKey(lambda a,b: (a[0]+ b[0], a[1]+ b[1]))
# Action for RDD to execute sc
print ( 'Total 4.1 row count:', driver_data_count1.count())
driver_data_count1.take(2)
# Filter values with sum of Duration time  and sum of Total _amount as values 0
fil_driver_data_count1 =  driver_data_count1.filter(lambda p: True if (int(p[1][0]) != 0 and  float(p[1][1]) != 0 ) else False)
print ( 'Filtered Total 4.2 row count:', fil_driver_data_count1.count())
fil_driver_data_count1.take(2)

#calculate the average rate per minute by division of sums for each driver
driver_data_count2 = fil_driver_data_count1.map(lambda x: (x[0],round(float((float(x[1][1])/int(x[1][0]))*60),2 )))
driver_data_count2.take(2)
# Get the top 10 of the average rate per minute  for the drivers
print(' 4.2 Task Top 10   Driver Average Rate pairs: ')
driver_data_count2.top(10, key = lambda p: round(float(p[1]),2))
# 4.3. Task 3 (8 Points)
# We are interested to know the time of the day (pickup time of day - p[2]) that has the highest profit ratio. 
# Consider the surcharge amount in dollar for each taxi ride (without tip amount) and the distance in
# miles, and sum up the rides for each hour of the day (24 hours) – consider the pickup time for your calculation.
# Profit Ratio = (Surcharge Amount p[12] in US Dollar) / (Travel Distance p[5] in miles)
# filter data that has "0.0" dollars surcharge and 0.00 as trip distance and date has  invalid data as  zeros 
taxis_filt_data1 = taxis_filt_data.filter(lambda p: True if (p[2] != "0000-00-00 00:00:00" and float(p[12]) != 0.0  and   float(p[5]) != 0.0 ) else False)
print('4.3 count of filtered2 rows:', taxis_filt_data1.count())
# Data has multiple days and hours and Key is defined as "Date + hour [00-24]"
# Map data to get date and hour as unique  Key  surcharge, trip_distance
taxis_filt_data2 = taxis_filt_data1.map(lambda x: (x[2][0:10]+x[2][11:13], (float(x[12]), float(x[5]))) )
#print sample 5 rows of data
taxis_filt_data2.take(2)

#Filter  data if date format ( key)  is null or bad rows of data
taxis_filt_data21 = taxis_filt_data2.filter(lambda p: True if (p[0] != "0000-00-0000" or p[0] != '' or len(p[0] != 12)) else False)
print('4.3 count of filtered3 rows:' ,taxis_filt_data21.count())
#compute sum of surcharge and sum of trip_distance in miles by time of day by hour ( for all the given dates and hours combination  -  defined as key ) using reduceByKey
taxis_filt_data3 = taxis_filt_data21.reduceByKey(lambda x,y: (round(float(x[0]) + float(y[0]),2),round(float(x[1])  + float(y[1]),2)))
print('4.3 Total number of Time of day ( keys):' , taxis_filt_data3.count())
#Execute RDD with action function
taxis_filt_data3.take(2)
# compute the Profit ratio as sum(surchargeby hourof day)/sum(trip distance by hour of day)
taxis_filt_data4 = taxis_filt_data3.map(lambda x: (x[0],  round(float(float(x[1][0])/float(x[1][1])),2))) 
print('4.3 Total number of Profit Ratio Date-time:' , taxis_filt_data4.count())
taxis_filt_data4.take(3)
# Get the time of the day that has the highest profit ratio using max function
print('4.3 Time of the day that has the highest profit ratio:' )
taxis_filt_data4.max( lambda x: x[1])
 
# Get the time of the day that has the highest profit ratio using max function using top function
print('4.3 Time of the day that has the highest profit ratio:' )
taxis_filt_data4.top(1, key = lambda p: p[1])<jupyter_output>4.3 Time of the day that has the highest profit ratio:
{'fraction_non_alphanumeric': 0.12304409672830725, 'fraction_numerical': 0.03959222380275012, 'mean_word_length': 3.3244490005125575, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 5, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 25, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '380344', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 2958, 'n_tokens_neox': 2804, 'n_words': 971, 'path': 'CS777_Cos_pkothapalli_Assignmnet1.ipynb', 'repo_name': 'pvkothapalli/MET-CS777', 'size': '8436'}