<commit_before>##Imports
import pyspark
import rtree
import geopandas as gpd
import shapely.geometry as geom
import sys

### yellow_tripdata_2011-05.csv

### 0  vendor_id,pickup_datetime,dropoff_datetime,passenger_count,
### 4  trip_distance,pickup_longitude,pickup_latitude,rate_code,
### 8  store_and_fwd_flag,dropoff_longitude,dropoff_latitude,
### 11 payment_type,fare_amount,surcharge,mta_tax,tip_amount,
### 16 tolls_amount,total_amount

import operator
import heapq

def tripMapper(records):
    import rtree
    import geopandas as gpd
    import shapely.geometry as geom
    #first create index
    index = rtree.Rtree()
    hoods = gpd.read_file('neighborhoods.geojson')
    for idx,geometry in enumerate(hoods.geometry):
        index.insert(idx,geometry.bounds)
    #for each record yield ((nei_idx,boro_idx), 1)
    neiborhood_name = "Undefined"
    boro_name = "Undefined"
    for record in records:
        list_record = record.split(',')
        if  list_record[0] != 'vendor_id' and list_record[0]!='':
            try:
               point_origin      = geom.Point(float(list_record[5]),float(list_record[6]))
               point_destination = geom.Point(float(list_record[9]),float(list_record[10]))
            except:
               continue

            #get origin neighborhood
            matches = list(index.intersection((point_origin.x,point_origin.y)))
            for ind in matches:
                if any(map(lambda x: x.contains(point_origin), hoods.geometry[ind])):
                    neiborhood_name = hoods.neighborhood[ind]
            #get destination borough
            matches = list(index.intersection((point_destination.x,point_destination.y)))
            for ind in matches:
                if any(map(lambda x: x.contains(point_destination), hoods.geometry[ind])):
                    boro_name = hoods.borough[ind]
            yield ((neiborhood_name,boro_name), 1)


def compute_top3((boro,hoodcounts)):
    yield (boro, heapq.nlargest(3, hoodcounts, lambda x: x[1]))

if __name__=='__main__':
    if len(sys.argv)<3:
        print "Usage: <input files> <output path>"
        sys.exit(-1)

    sc = pyspark.SparkContext()

    trips = sc.textFile(','.join(sys.argv[1:-1]))

    #output = trips \
     #   .mapPartitions(tripMapper).reduceByKey(operator.add).map(lambda x: (x[0][1], (x[0][0], x[1]))) \
      #  .groupByKey().map(compute_top3)
    output = sc.parallelize(trips.mapPartitions(tripMapper).reduceByKey(operator.add).\
                map(lambda x: (x[0][1], (x[0][0], x[1]))).take(100))

    output.saveAsTextFile(sys.argv[-1])

    #groupbyKey does not work, i emailed prof.<commit_msg>optimization 1 uses reduceByKey instead of groupByKey().map(...)<commit_after>##Imports
import pyspark
import rtree
import geopandas as gpd
import shapely.geometry as geom
import sys

### yellow_tripdata_2011-05.csv

### 0  vendor_id,pickup_datetime,dropoff_datetime,passenger_count,
### 4  trip_distance,pickup_longitude,pickup_latitude,rate_code,
### 8  store_and_fwd_flag,dropoff_longitude,dropoff_latitude,
### 11 payment_type,fare_amount,surcharge,mta_tax,tip_amount,
### 16 tolls_amount,total_amount

import operator
import heapq

def tripMapper(records):
    import rtree
    import geopandas as gpd
    import shapely.geometry as geom
    #first create index
    index = rtree.Rtree()
    hoods = gpd.read_file('neighborhoods.geojson')
    for idx,geometry in enumerate(hoods.geometry):
        index.insert(idx,geometry.bounds)
    #for each record yield ((nei_idx,boro_idx), 1)
    neiborhood_name = "Undefined"
    boro_name = "Undefined"
    for record in records:
        list_record = record.split(',')
        if  list_record[0] != 'vendor_id' and list_record[0]!='':
            try:
               point_origin      = geom.Point(float(list_record[5]),float(list_record[6]))
               point_destination = geom.Point(float(list_record[9]),float(list_record[10]))
            except:
               continue

            #get origin neighborhood
            matches = list(index.intersection((point_origin.x,point_origin.y)))
            for ind in matches:
                if any(map(lambda x: x.contains(point_origin), hoods.geometry[ind])):
                    neiborhood_name = hoods.neighborhood[ind]
            #get destination borough
            matches = list(index.intersection((point_destination.x,point_destination.y)))
            for ind in matches:
                if any(map(lambda x: x.contains(point_destination), hoods.geometry[ind])):
                    boro_name = hoods.borough[ind]
            yield ((neiborhood_name,boro_name), 1)


def compute_top3((boro,hoodcounts)):
    return (boro, heapq.nlargest(3, hoodcounts, lambda x: x[1]))


def top3Reducer(lst1,lst2):
    return heapq.nlargest(3,lst1+lst2,lambda x: x[1])



if __name__=='__main__':
    if len(sys.argv)<3:
        print "Usage: <input files> <output path>"
        sys.exit(-1)

    sc = pyspark.SparkContext()

    trips = sc.textFile(','.join(sys.argv[1:-1]))

    #output = trips \
     #   .mapPartitions(tripMapper).reduceByKey(operator.add).map(lambda x: (x[0][1], (x[0][0], x[1]))) \
      #  .groupByKey().map(compute_top3)
    output = sc.parallelize(trips.mapPartitions(tripMapper).reduceByKey(operator.add).\
                map(lambda x: (x[0][1], [(x[0][0], x[1])])).reduceByKey(top3Reducer).collect())

    output.saveAsTextFile(sys.argv[-1])

    #groupbyKey does not work, i emailed prof.
{'fraction_non_alphanumeric': 0.1284537968892955, 'fraction_numerical': 0.0172003659652333, 'mean_word_length': 3.2143407864302236, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1146346', 'n_tokens_mistral': 1879, 'n_tokens_neox': 1779, 'n_words': 339}