datasets/ETL/Taxi_ETL.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import cudf\n",
    "import numpy as np\n",
    "from numba import cuda\n",
    "import math"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "PERCENT_TRAIN = 0.8"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#Numba Kernel to calculate Haversine distance\n",
    "@cuda.jit\n",
    "def haversine_kernel(lat1, lon1, lat2, lon2, outputCol):\n",
    "    iRow = cuda.grid(1)\n",
    "    p = 0.017453292519943295 # Pi/180\n",
    "    if iRow < outputCol.size:\n",
    "        a = 0.5 - math.cos((lat2[iRow] - lat1[iRow]) * p)/2 + math.cos(lat1[iRow] * p) * \\\n",
    "            math.cos(lat2[iRow] * p) * (1 - math.cos((lon2[iRow] - lon1[iRow]) * p)) / 2                                 \n",
    "        outputCol[iRow] = 12734 * math.asin(math.sqrt(a))\n",
    "    \n",
    "def haversine_distance(gdf):\n",
    "    nRows = gdf.shape[0]\n",
    "    blockSize = 128\n",
    "    blockCount = nRows // blockSize + 1\n",
    "    lat1_arr = gdf['pickup_latitude'].to_gpu_array()\n",
    "    lon1_arr = gdf['pickup_longitude'].to_gpu_array()\n",
    "    lat2_arr = gdf['dropoff_latitude'].to_gpu_array()\n",
    "    lon2_arr = gdf['dropoff_longitude'].to_gpu_array()\n",
    "                                   \n",
    "    outputCol = cuda.device_array ( shape=(nRows), dtype=lat1_arr.dtype.name)\n",
    "    \n",
    "    haversine_kernel[(blockCount),(blockSize)](lat1_arr, lon1_arr, lat2_arr, lon2_arr, outputCol)\n",
    "    gdf.add_column(name='h_distance', data = outputCol)\n",
    "    return gdf\n",
    "\n",
    "#Numba Kernel to calculate day of the week from Date\n",
    "@cuda.jit\n",
    "def day_of_the_week_kernel(output ,year, month, day):\n",
    "    iRow = cuda.grid(1)\n",
    "    if iRow < output.size:\n",
    "        year[iRow] -= month[iRow] < 3\n",
    "        month[iRow] = (month[iRow] + 9)%12 + 1\n",
    "        output[iRow] = (year[iRow] + int(year[iRow]/4) - int(year[iRow]/100) + int(year[iRow]/400) + math.floor(2.6*month[iRow] - 0.2) + day[iRow] -1) % 7\n",
    "    \n",
    "def day_of_week(gdf):\n",
    "    nRows = gdf.shape[0]\n",
    "    blockSize = 128\n",
    "    blockCount = nRows // blockSize + 1\n",
    "    year_arr = gdf['year'].to_gpu_array()\n",
    "    month_arr = gdf['month'].to_gpu_array()\n",
    "    day_arr = gdf['day'].to_gpu_array()\n",
    "    outputCol = cuda.device_array ( shape=(nRows), dtype=day_arr.dtype.name)\n",
    "    \n",
    "    day_of_the_week_kernel[(blockCount),(blockSize)](outputCol, year_arr, month_arr, day_arr)\n",
    "    gdf.add_column(name='day_of_week', data = outputCol)\n",
    "    gdf['day_of_week'] = gdf['day_of_week'].astype('float32')\n",
    "    return gdf\n",
    "    \n",
    "import pandas as pd\n",
    "def gpu_read_csv(file_path):\n",
    "    names  = ['vendor_id','pickup_datetime','dropoff_datetime','passenger_count','trip_distance','pickup_longitude',\n",
    "              'pickup_latitude','rate_code','store_and_fwd','dropoff_longitude','dropoff_latitude','payment_type',\n",
    "              'fare_amount','surcharge','mta_tax','tip_amount','tolls_amount','total_amount']\n",
    "    \n",
    "    dtypes = ['category','date','date','int','float64','float64','float64','category','category','float64','float64',\n",
    "              'category','float64','float64','float64','float64','float64','float64']\n",
    "\n",
    "    df = cudf.read_csv(file_path, dtype=dtypes, names=names,skiprows=1)\n",
    "    return df\n",
    "\n",
    "def null_workaround(df, **kwargs):\n",
    "    for column, data_type in df.dtypes.items():\n",
    "        if str(data_type) in ['int8', 'int16', 'int32', 'int64', 'float32', 'float64']:\n",
    "            df[column] = df[column].fillna(-1)\n",
    "    return df\n",
    "\n",
    "def clean_data(df):\n",
    "    drop_list = [\n",
    "        'dropoff_datetime', 'payment_type', 'surcharge', 'mta_tax',\n",
    "        'tip_amount', 'tolls_amount', 'total_amount'\n",
    "    ]\n",
    "\n",
    "    for column in drop_list:\n",
    "        df.drop_column(column)\n",
    "        \n",
    "    df = null_workaround(df)\n",
    "        \n",
    "    df_fare = df.query('fare_amount > 0 and fare_amount < 500')\n",
    "    del(df)\n",
    "    \n",
    "    df_pass = df_fare.query('passenger_count > 0 and passenger_count < 6')\n",
    "    del(df_fare)\n",
    "    \n",
    "    df_picklong = df_pass.query('pickup_longitude > -75 and pickup_longitude < -73')\n",
    "    del(df_pass)\n",
    "    \n",
    "    df_droplong = df_picklong.query('dropoff_longitude > -75 and dropoff_longitude < -73')\n",
    "    del(df_picklong)\n",
    "    \n",
    "    df_picklat = df_droplong.query('pickup_latitude > 40 and pickup_latitude < 42')\n",
    "    del(df_droplong)\n",
    "    \n",
    "    df_droplat = df_picklat.query('dropoff_latitude > 40 and dropoff_latitude < 42')\n",
    "    del(df_picklat)\n",
    "    \n",
    "    return df_droplat\n",
    "    \n",
    "def add_features(df):\n",
    "    df['hour'] = df['pickup_datetime'].dt.hour\n",
    "    df['year'] = df['pickup_datetime'].dt.year\n",
    "    df['month'] = df['pickup_datetime'].dt.month\n",
    "    df['day'] = df['pickup_datetime'].dt.day\n",
    "    \n",
    "    df.drop_column('pickup_datetime')\n",
    "    \n",
    "    df = day_of_week(df)\n",
    "    df['is_weekend'] = (df['day_of_week']/4).floor()\n",
    "    df = haversine_distance(df)\n",
    "    return df\n",
    "    \n",
    "\n",
    "def process_data(train_path):\n",
    "    df = gpu_read_csv(train_path)\n",
    "    df = clean_data(df)\n",
    "    df = add_features(df)\n",
    "    return df"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "month = 1\n",
    "start = 2016\n",
    "end = 2016\n",
    "year = start\n",
    "\n",
    "# INPUT DIRECTORY (change this to point to where you downloaded the dataset)\n",
    "INPUT_DIRECTORY = \"/datasets/test/raw/taxi\"\n",
    "\n",
    "# OUTPUT DIRECTORY (change this to point to where you want the processed dataset output)\n",
    "OUTPUT_DIRECTORY = \"/datasets/test/taxi\"\n",
    "\n",
    "while year <= end:\n",
    "    current_part_path = INPUT_DIRECTORY + \"/yellow_tripdata_\" + str(year) + \"-\" + f\"{month:02d}\" + \".csv\"\n",
    "    \n",
    "    train_part_path_pq = OUTPUT_DIRECTORY + \"/parquet/train/yellow_tripdata_\" + str(year) + \"-\" + str(month) + \".parquet\"    \n",
    "    test_part_path_pq = OUTPUT_DIRECTORY + \"/parquet/test/yellow_tripdata_\" + str(year) + \"-\" + str(month) + \".parquet\"\n",
    "    \n",
    "    train_part_path_csv = OUTPUT_DIRECTORY + \"/csv/train/yellow_tripdata_\" + str(year) + \"-\" + str(month) + \".csv\" \n",
    "    test_part_path_csv = OUTPUT_DIRECTORY + \"/csv/test/yellow_tripdata_\" + str(year) + \"-\" + str(month) + \".csv\"\n",
    "    \n",
    "    print(current_part_path)\n",
    "    df = process_data(current_part_path)\n",
    "    month += 1\n",
    "    \n",
    "    msk = np.random.rand(len(df)) < PERCENT_TRAIN\n",
    "    \n",
    "    train = df[msk]\n",
    "    test = df[~msk]\n",
    "    \n",
    "    print(train.shape)\n",
    "    print(test.shape)\n",
    "    \n",
    "    train.to_parquet(train_part_path_pq)\n",
    "    test.to_parquet(test_part_path_pq)\n",
    "    \n",
    "    train.to_pandas().to_csv(train_part_path_csv, header=False)\n",
    "    test.to_pandas().to_csv(train_part_path_csv, header=False)\n",
    "    \n",
    "    del train\n",
    "    del test\n",
    "    del df\n",
    "    \n",
    "    if month > 12:\n",
    "        month = 1\n",
    "        year += 1\n",
    "    "
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}