text_plain_output_1.png
from dask.diagnostics import ProgressBar

import dask.dataframe as dd

import datetime

import math

dask_data = dd.read_csv('./train.csv')

dask_data.columns

dask_data.compute().shape
len(dask_data.columns)
dask_data.isnull().sum().compute()
dask_data.fare_amount.mean().compute()

missing_values = dask_data.isnull().sum().compute()
missing_values

mysize = dask_data.index.size.compute()
missing_count = missing_values / mysize * 100
missing_count

def haversine_dist(long_pickup, long_dropoff, lat_pickup, lat_dropoff):
    distance = []
    for i in range(len(long_pickup)):
        long1, long2, lat1, lat2 = map(math.radians, (long_pickup[i], long_dropoff[i], lat_pickup[i], lat_dropoff[i]))
        dlat = lat2 - lat1
        dlong = long2 - long1
        a = math.sin(dlat / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin(dlong / 2) ** 2
        distance.append(2 * math.asin(math.sqrt(a)) * 6371)
    return distance

dask_data.columns

dist_km_interim = dask_data.map_partitions(lambda df: haversine_dist(df['pickup_longitude'], df['dropoff_longitude'], df['pickup_latitude'], df['dropoff_latitude']))

dask_data_new = dask_data.assign(dist_km=dist_km_interim)

dask_data['dist_km'] = haversine_dist(dask_data['pickup_longitude'], dask_data['dropoff_longitude'], dask_data['pickup_latitude'], dask_data['dropoff_latitude'])
dask_data_test['dist_km'] = haversine_dist(dask_data_test['pickup_longitude'], dask_data_test['dropoff_longitude'], dask_data_test['pickup_latitude'], dask_data_test['dropoff_latitude'])
dask_data.head(5)