<s> from pyspark import StorageLevel , SparkContext <EOL> sc = SparkContext ( ) <EOL> root_dir = '<STR_LIT>' <EOL> sc . addPyFile ( root_dir + '<STR_LIT>' ) <EOL> sc . addPyFile ( root_dir + '<STR_LIT>' ) <EOL> sc . addPyFile ( root_dir + '<STR_LIT>' ) <EOL> sc . addPyFile ( root_dir + '<STR_LIT>' ) <EOL> sc . addPyFile ( root_dir + '<STR_LIT>' ) <EOL> sc . addPyFile ( root_dir + '<STR_LIT>' ) <EOL> from geo_vars import ( <EOL> tiff_to_array , <EOL> get_mask_indices <EOL> ) <EOL> from time_vars import ( <EOL> get_weather_data <EOL> ) <EOL> from rdd_utils import ( <EOL> parse_taxi_record , <EOL> get_aggregates_rdd , <EOL> combine_values , <EOL> merge_combiner , <EOL> labeled_point_to_row_col_period , <EOL> get_labeled_points <EOL> ) <EOL> from raster_utils import ( <EOL> trips_to_raster , <EOL> kernel_density , <EOL> neighbor_average <EOL> ) <EOL> from constants import ( <EOL> weekday_dict , <EOL> hours_dict <EOL> ) <EOL> weekday_map = sc . broadcast ( weekday_dict ) <EOL> hours_dict = sc . broadcast ( hours_dict ) <EOL> data_dir = '<STR_LIT>' <EOL> weather_data = sc . broadcast ( get_weather_data ( data_dir + '<STR_LIT>' ) ) <EOL> mask = sc . broadcast ( get_mask_indices ( data_dir + '<STR_LIT>' ) ) <EOL> roads = sc . broadcast ( <EOL> tiff_to_array ( data_dir + '<STR_LIT>' ) ) <EOL> roads_distance = sc . broadcast ( <EOL> tiff_to_array ( data_dir + '<STR_LIT>' ) ) <EOL> airports = sc . broadcast ( <EOL> tiff_to_array ( data_dir + '<STR_LIT>' ) ) <EOL> airports_distance = sc . broadcast ( <EOL> tiff_to_array ( data_dir + '<STR_LIT>' ) ) <EOL> transport = sc . broadcast ( <EOL> tiff_to_array ( data_dir + '<STR_LIT>' ) ) <EOL> transport_distance = sc . broadcast ( <EOL> tiff_to_array ( data_dir + '<STR_LIT>' ) ) <EOL> raw_data_url = data_dir + '<STR_LIT>' <EOL> raw_data = sc . textFile ( raw_data_url ) <EOL> trips = ( raw_data . map ( parse_taxi_record ) <EOL> . reduceByKey ( lambda a , b : a + b ) <EOL> . repartition ( <NUM_LIT> ) ) <EOL> trips . persist ( StorageLevel . MEMORY_AND_DISK ) <EOL> trips_rasters = trips_to_raster ( trips , '<STR_LIT>' ) <EOL> two_hour_agg = get_aggregates_rdd ( trips , hours_dict , <NUM_LIT:1> , <NUM_LIT:3> , <NUM_LIT:1> ) <EOL> two_hour_raster = trips_to_raster ( two_hour_agg , '<STR_LIT>' ) <EOL> two_hour_kd = two_hour_raster . mapValues ( <EOL> lambda ( label , raster ) : ( kernel_density ( '<STR_LIT>' , raster ) ) ) <EOL> two_hour_neighbor_avg = two_hour_raster . mapValues ( <EOL> lambda ( label , raster ) : ( neighbor_average ( '<STR_LIT>' , raster ) ) ) <EOL> four_hour_agg = get_aggregates_rdd ( two_hour_agg , hours_dict , <NUM_LIT:0> , <NUM_LIT:4> , <NUM_LIT:2> ) <EOL> four_hour_raster = trips_to_raster ( four_hour_agg , '<STR_LIT>' ) <EOL> four_hour_kd = four_hour_raster . mapValues ( <EOL> lambda ( label , raster ) : ( kernel_density ( '<STR_LIT>' , raster ) ) ) <EOL> four_hour_neighbor_avg = four_hour_raster . mapValues ( <EOL> lambda ( label , raster ) : ( neighbor_average ( '<STR_LIT>' , raster ) ) ) <EOL> eight_hour_agg = get_aggregates_rdd ( four_hour_agg , hours_dict , <NUM_LIT:0> , <NUM_LIT:8> , <NUM_LIT:4> ) <EOL> eight_hour_raster = trips_to_raster ( eight_hour_agg , '<STR_LIT>' ) <EOL> eight_hour_kd = eight_hour_raster . mapValues ( <EOL> lambda ( label , raster ) : ( kernel_density ( '<STR_LIT>' , raster ) ) ) <EOL> eight_hour_neighbor_avg = eight_hour_raster . mapValues ( <EOL> lambda ( label , raster ) : ( neighbor_average ( '<STR_LIT>' , raster ) ) ) <EOL> previous_day = get_aggregates_rdd ( trips , hours_dict , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1> ) <EOL> previous_day_raster = trips_to_raster ( previous_day , '<STR_LIT>' ) <EOL> previous_week = get_aggregates_rdd ( trips , hours_dict , <NUM_LIT> , <NUM_LIT> , <NUM_LIT:1> ) <EOL> previous_week_raster = trips_to_raster ( previous_week , '<STR_LIT>' ) <EOL> unioned_rdd = ( trips_rasters . union ( two_hour_raster ) <EOL> . union ( four_hour_raster ) <EOL> . union ( eight_hour_raster ) <EOL> . union ( previous_day_raster ) <EOL> . union ( previous_week_raster ) <EOL> . union ( two_hour_kd ) <EOL> . union ( two_hour_neighbor_avg ) <EOL> . union ( four_hour_kd ) <EOL> . union ( four_hour_neighbor_avg ) <EOL> . union ( eight_hour_kd ) <EOL> . union ( eight_hour_neighbor_avg ) ) <EOL> combined = unioned_rdd . aggregateByKey ( { } , combine_values , merge_combiner ) <EOL> ( training_data , test_data ) = combined . randomSplit ( [ <NUM_LIT> , <NUM_LIT> ] ) <EOL> geovars = [ roads , roads_distance , transport , transport_distance , airports , airports_distance ] <EOL> data_ordering = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> training_data_points = training_data . flatMap ( lambda ( period , data_dict ) : <EOL> get_labeled_points ( period , data_dict , data_ordering , mask , <EOL> weekday_map , geovars , weather_data ) ) <EOL> test_data_points = test_data . flatMap ( lambda ( period , data_dict ) : <EOL> get_labeled_points ( period , data_dict , data_ordering , mask , <EOL> weekday_map , geovars , weather_data ) ) <EOL> from pyspark . mllib . tree import RandomForest <EOL> model = RandomForest . trainRegressor ( training_data_points , categoricalFeaturesInfo = { } , <EOL> numTrees = <NUM_LIT:50> , featureSubsetStrategy = "<STR_LIT>" , <EOL> impurity = '<STR_LIT>' , maxDepth = <NUM_LIT:3> , maxBins = <NUM_LIT:32> ) <EOL> predictions = model . predict ( test_data_points . map ( lambda x : x . features ) ) <EOL> predicted_row_col = test_data_points . map ( labeled_point_to_row_col_period ) . zip ( predictions ) <EOL> predicted_rasters = ( trips_to_raster ( predicted_row_col , '<STR_LIT>' ) <EOL> . map ( lambda ( period , ( label , raster ) ) : ( period , raster ) ) ) <EOL> observed_test_rasters = test_data . filter ( lambda ( x , d ) : '<STR_LIT>' in d ) . mapValues ( lambda d : d [ '<STR_LIT>' ] ) <EOL> two_hour_avg = test_data . filter ( lambda ( x , d ) : '<STR_LIT>' in d ) . mapValues ( lambda d : d [ '<STR_LIT>' ] / <NUM_LIT:2> ) </s>
