notebooks/nyc-taxi-fare-gpu.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# NYC Taxi Fare Prediction using GPU<br>\n",
    "## Standard_NC24s_v3 VM on Microsoft Azure with 4*NVIDIA Tesla V100 GPUs with 16 GB memory each"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Import"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "_cell_guid": "b1076dfc-b9ad-4769-8c92-a6c4dae69d19",
    "_uuid": "8f2839f25d086af736a60e9eeb907d3b93b6e0e5",
    "papermill": {
     "duration": 7.040183,
     "end_time": "2020-12-06T22:37:15.699282",
     "exception": false,
     "start_time": "2020-12-06T22:37:08.659099",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "import pandas as pd\n",
    "import xgboost as xgb\n",
    "import dask, dask_cudf\n",
    "from dask_cuda import LocalCUDACluster\n",
    "from dask.distributed import Client\n",
    "from dask_ml.model_selection import train_test_split\n",
    "from dask.utils import parse_bytes\n",
    "from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score\n",
    "from math import cos, sin, asin, sqrt, pi\n",
    "import time\n",
    "start = time.perf_counter()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Create CUDA Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "distributed.preloading - INFO - Import preload module: dask_cuda.initialize\n",
      "distributed.preloading - INFO - Import preload module: dask_cuda.initialize\n",
      "distributed.preloading - INFO - Import preload module: dask_cuda.initialize\n",
      "distributed.preloading - INFO - Import preload module: dask_cuda.initialize\n",
      "distributed.preloading - INFO - Import preload module: dask_cuda.initialize\n",
      "distributed.preloading - INFO - Import preload module: dask_cuda.initialize\n",
      "distributed.preloading - INFO - Import preload module: dask_cuda.initialize\n",
      "distributed.preloading - INFO - Import preload module: dask_cuda.initialize\n"
     ]
    }
   ],
   "source": [
    "cluster = LocalCUDACluster(\n",
    "    rmm_pool_size=parse_bytes(\"64GB\") #I've 64GB of GPU Memory, set this according to your setup.\n",
    ")\n",
    "client = Client(cluster)\n",
    "client.restart()\n",
    "\n",
    "dask.config.set({'distributed.scheduler.work-stealing': False})\n",
    "cluster_creation_time = time.perf_counter()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Reading Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "papermill": {
     "duration": 1.240981,
     "end_time": "2020-12-06T22:37:18.792968",
     "exception": false,
     "start_time": "2020-12-06T22:37:17.551987",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "df = dask_cudf.read_csv('../data/train.csv')\n",
    "read_time = time.perf_counter()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Convert Data Types"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "def dtype_conversion(df):\n",
    "    df['key'] = df['key'].astype('datetime64[ns]')\n",
    "    df['fare_amount'] = df ['fare_amount'].astype('float32')\n",
    "    df['pickup_datetime'] = df['pickup_datetime'].astype('datetime64[ns]')\n",
    "    df['pickup_longitude'] = df ['pickup_longitude'].astype('float32')\n",
    "    df['pickup_latitude'] = df ['pickup_latitude'].astype('float32')\n",
    "    df['dropoff_longitude'] = df ['dropoff_longitude'].astype('float32')\n",
    "    df['dropoff_latitude'] = df ['dropoff_latitude'].astype('float32')\n",
    "    df['passenger_count'] = df ['passenger_count'].astype('uint8')\n",
    "    return df\n",
    "\n",
    "df = dtype_conversion(df)\n",
    "dtype_conversion_time = time.perf_counter()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Data Cleaning & Filtering, Applying NYC specific constraints"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "papermill": {
     "duration": 0.72351,
     "end_time": "2020-12-06T22:37:22.832471",
     "exception": false,
     "start_time": "2020-12-06T22:37:22.108961",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def apply_constraints(df):\n",
    "    query_frags = [\n",
    "        'fare_amount >= 2.5 and fare_amount < 500',\n",
    "        'passenger_count > 0 and passenger_count < 6',\n",
    "        'pickup_longitude > -75 and pickup_longitude < -73',\n",
    "        'dropoff_longitude > -75 and dropoff_longitude < -73',\n",
    "        'pickup_latitude > 40 and pickup_latitude < 42',\n",
    "        'dropoff_latitude > 40 and dropoff_latitude < 42'\n",
    "    ]\n",
    "    df = df.query(' and '.join(query_frags))\n",
    "    return df\n",
    "\n",
    "df = apply_constraints(df)\n",
    "apply_constraints_time = time.perf_counter()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 6. Feature Engineering"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "papermill": {
     "duration": 0.068832,
     "end_time": "2020-12-06T22:37:22.915127",
     "exception": false,
     "start_time": "2020-12-06T22:37:22.846295",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "def feature_engg(df):\n",
    "    def jfk_distance(dropoff_latitude, dropoff_longitude, jfk_distance):\n",
    "        for i, (x_1, y_1) in enumerate(zip(dropoff_latitude, dropoff_longitude)):\n",
    "            x_1 = pi/180 * x_1\n",
    "            y_1 = pi/180 * y_1\n",
    "            x_jfk = pi/180 * 40.6413\n",
    "            y_jfk = pi/180 * -73.7781\n",
    "            \n",
    "            dlon = y_jfk - y_1\n",
    "            dlat = x_jfk - x_1\n",
    "            a = sin(dlat/2)**2 + cos(x_1) * cos(x_jfk) * sin(dlon/2)**2\n",
    "            \n",
    "            c = 2 * asin(sqrt(a)) \n",
    "            r = 6371 # Radius of earth in kilometers\n",
    "            \n",
    "            jfk_distance[i] = c * r\n",
    "            \n",
    "    def lga_distance(dropoff_latitude, dropoff_longitude, lga_distance):\n",
    "        for i, (x_1, y_1) in enumerate(zip(dropoff_latitude, dropoff_longitude)):\n",
    "            x_1 = pi/180 * x_1\n",
    "            y_1 = pi/180 * y_1\n",
    "            x_lga = pi/180 * 40.7769\n",
    "            y_lga = pi/180 * -73.8740\n",
    "            \n",
    "            dlon = y_lga - y_1\n",
    "            dlat = x_lga - x_1\n",
    "            a = sin(dlat/2)**2 + cos(x_1) * cos(x_lga) * sin(dlon/2)**2\n",
    "            \n",
    "            c = 2 * asin(sqrt(a)) \n",
    "            r = 6371 # Radius of earth in kilometers\n",
    "            \n",
    "            lga_distance[i] = c * r\n",
    "            \n",
    "    def ewr_distance(dropoff_latitude, dropoff_longitude, ewr_distance):\n",
    "        for i, (x_1, y_1) in enumerate(zip(dropoff_latitude, dropoff_longitude)):\n",
    "            x_1 = pi/180 * x_1\n",
    "            y_1 = pi/180 * y_1\n",
    "            x_ewr = pi/180 * 40.6895\n",
    "            y_ewr = pi/180 * -74.1745\n",
    "            \n",
    "            dlon = y_ewr - y_1\n",
    "            dlat = x_ewr - x_1\n",
    "            a = sin(dlat/2)**2 + cos(x_1) * cos(x_ewr) * sin(dlon/2)**2\n",
    "            \n",
    "            c = 2 * asin(sqrt(a)) \n",
    "            r = 6371 # Radius of earth in kilometers\n",
    "            \n",
    "            ewr_distance[i] = c * r\n",
    "            \n",
    "    def tsq_distance(dropoff_latitude, dropoff_longitude, tsq_distance):\n",
    "        for i, (x_1, y_1) in enumerate(zip(dropoff_latitude, dropoff_longitude)):\n",
    "            x_1 = pi/180 * x_1\n",
    "            y_1 = pi/180 * y_1\n",
    "            x_tsq = pi/180 * 40.7580\n",
    "            y_tsq = pi/180 * -73.9855\n",
    "            \n",
    "            dlon = y_tsq - y_1\n",
    "            dlat = x_tsq - x_1\n",
    "            a = sin(dlat/2)**2 + cos(x_1) * cos(x_tsq) * sin(dlon/2)**2\n",
    "            \n",
    "            c = 2 * asin(sqrt(a)) \n",
    "            r = 6371 # Radius of earth in kilometers\n",
    "            \n",
    "            tsq_distance[i] = c * r\n",
    "            \n",
    "    def met_distance(dropoff_latitude, dropoff_longitude, met_distance):\n",
    "        for i, (x_1, y_1) in enumerate(zip(dropoff_latitude, dropoff_longitude)):\n",
    "            x_1 = pi/180 * x_1\n",
    "            y_1 = pi/180 * y_1\n",
    "            x_met = pi/180 * 40.7794\n",
    "            y_met = pi/180 * -73.9632\n",
    "            \n",
    "            dlon = y_met - y_1\n",
    "            dlat = x_met - x_1\n",
    "            a = sin(dlat/2)**2 + cos(x_1) * cos(x_met) * sin(dlon/2)**2\n",
    "            \n",
    "            c = 2 * asin(sqrt(a)) \n",
    "            r = 6371 # Radius of earth in kilometers\n",
    "            \n",
    "            met_distance[i] = c * r\n",
    "            \n",
    "    def wtc_distance(dropoff_latitude, dropoff_longitude, wtc_distance):\n",
    "        for i, (x_1, y_1) in enumerate(zip(dropoff_latitude, dropoff_longitude)):\n",
    "            x_1 = pi/180 * x_1\n",
    "            y_1 = pi/180 * y_1\n",
    "            x_wtc = pi/180 * 40.7126\n",
    "            y_wtc = pi/180 * -74.0099\n",
    "            \n",
    "            dlon = y_wtc - y_1\n",
    "            dlat = x_wtc - x_1\n",
    "            a = sin(dlat/2)**2 + cos(x_1) * cos(x_wtc) * sin(dlon/2)**2\n",
    "            \n",
    "            c = 2 * asin(sqrt(a)) \n",
    "            r = 6371 # Radius of earth in kilometers\n",
    "            \n",
    "            wtc_distance[i] = c * r\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",
    "        df['weekday'] = df['pickup_datetime'].dt.weekday\n",
    "        \n",
    "        df = df.apply_rows(jfk_distance, incols=['dropoff_latitude', 'dropoff_longitude'],\n",
    "                        outcols=dict(jfk_distance=np.float32), kwargs=dict())\n",
    "        \n",
    "        df = df.apply_rows(lga_distance, incols=['dropoff_latitude', 'dropoff_longitude'],\n",
    "                        outcols=dict(lga_distance=np.float32), kwargs=dict())\n",
    "            \n",
    "        df = df.apply_rows(ewr_distance, incols=['dropoff_latitude', 'dropoff_longitude'],\n",
    "                        outcols=dict(ewr_distance=np.float32), kwargs=dict())\n",
    "                \n",
    "        df = df.apply_rows(tsq_distance, incols=['dropoff_latitude', 'dropoff_longitude'],\n",
    "                        outcols=dict(tsq_distance=np.float32), kwargs=dict())\n",
    "        \n",
    "        df = df.apply_rows(met_distance, incols=['dropoff_latitude', 'dropoff_longitude'],\n",
    "                        outcols=dict(met_distance=np.float32), kwargs=dict())\n",
    "        \n",
    "        df = df.apply_rows(wtc_distance, incols=['dropoff_latitude', 'dropoff_longitude'],\n",
    "                        outcols=dict(wtc_distance=np.float32), kwargs=dict())\n",
    "        \n",
    "        df = df.drop(['pickup_datetime','key'], axis=1)\n",
    "        \n",
    "        return df\n",
    "\n",
    "    parts = [dask.delayed(add_features)(part) for part in df.to_delayed()]\n",
    "    df = dask_cudf.from_delayed(parts)\n",
    "    return df\n",
    "\n",
    "df = feature_engg(df)\n",
    "feature_engg_time = time.perf_counter()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 7. Splitting Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "papermill": {
     "duration": 0.306403,
     "end_time": "2020-12-06T22:37:45.295311",
     "exception": false,
     "start_time": "2020-12-06T22:37:44.988908",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [],
   "source": [
    "y = df['fare_amount']\n",
    "X = df.drop(['fare_amount'], axis=1)\n",
    "\n",
    "X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.1, random_state=42, shuffle=True)\n",
    "split_data_time = time.perf_counter()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 8. Training"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "papermill": {
     "duration": 998.883894,
     "end_time": "2020-12-06T22:54:24.195726",
     "exception": false,
     "start_time": "2020-12-06T22:37:45.311832",
     "status": "completed"
    },
    "tags": []
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "[07:29:23] task [xgboost.dask]:tcp://10.1.0.7:37273 got new rank 0\n",
      "[07:29:23] task [xgboost.dask]:tcp://10.1.0.7:37251 got new rank 1\n",
      "[07:29:23] task [xgboost.dask]:tcp://10.1.0.7:38403 got new rank 2\n",
      "[07:29:23] task [xgboost.dask]:tcp://10.1.0.7:40617 got new rank 3\n"
     ]
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[0]\ttrain-rmse:13.89166\tvalid-rmse:13.89163\n",
      "[100]\ttrain-rmse:4.40803\tvalid-rmse:4.41362\n",
      "[200]\ttrain-rmse:4.07664\tvalid-rmse:4.08449\n",
      "[300]\ttrain-rmse:3.96525\tvalid-rmse:3.97594\n",
      "[400]\ttrain-rmse:3.87989\tvalid-rmse:3.89296\n",
      "[500]\ttrain-rmse:3.83961\tvalid-rmse:3.85505\n",
      "[600]\ttrain-rmse:3.80805\tvalid-rmse:3.82634\n",
      "[700]\ttrain-rmse:3.78097\tvalid-rmse:3.80171\n",
      "[800]\ttrain-rmse:3.75934\tvalid-rmse:3.78234\n",
      "[900]\ttrain-rmse:3.74227\tvalid-rmse:3.76762\n",
      "[999]\ttrain-rmse:3.72744\tvalid-rmse:3.75507\n"
     ]
    }
   ],
   "source": [
    "dtrain = xgb.dask.DaskDMatrix(client, X_train, y_train)\n",
    "dvalid = xgb.dask.DaskDMatrix(client, X_test, y_test)\n",
    "watchlist = [(dtrain, 'train'), (dvalid, 'valid')]\n",
    "\n",
    "params = {\n",
    "    'min_child_weight': 1,\n",
    "    'learning_rate': 0.05,\n",
    "    'colsample_bytree': 0.7,\n",
    "    #'max_depth': 11,\n",
    "    'subsample': 0.7,\n",
    "    'booster' : 'gbtree',\n",
    "    'objective': 'reg:squarederror',\n",
    "    'tree_method':'gpu_hist',\n",
    "    'eval_metric': \"rmse\",\n",
    "    }\n",
    "\n",
    "model = xgb.dask.train(client, params, dtrain, num_boost_round=1000, evals=watchlist, early_stopping_rounds=100, verbose_eval=100)\n",
    "training_time = time.perf_counter()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 9. Testing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Root Mean Squared Error :  3.7369375\n",
      "Mean Absolute Error :  1.6827899\n",
      "R-squared Score :  0.8508566613743141\n"
     ]
    }
   ],
   "source": [
    "def eval_metrics(actual, pred):\n",
    "    rmse = np.sqrt(mean_squared_error(actual, pred))\n",
    "    mae = mean_absolute_error(actual, pred)\n",
    "    r2 = r2_score(actual, pred)\n",
    "    return (rmse, mae, r2)\n",
    "\n",
    "def test(client, model):\n",
    "    df = dask_cudf.read_csv('../data/test.csv')\n",
    "    df = dtype_conversion(df)\n",
    "    df = apply_constraints(df)\n",
    "    df = feature_engg(df)\n",
    "    actual = df['fare_amount']\n",
    "    actual = actual.compute().to_array()\n",
    "    df =  df.drop('fare_amount', axis = 1)\n",
    "    df = xgb.dask.DaskDMatrix(client, df)\n",
    "\n",
    "    pred = xgb.dask.predict(client, model, df)\n",
    "    pred = pred.compute()\n",
    "    rmse, mae, r2 = eval_metrics(actual, pred)\n",
    "    print(\"Root Mean Squared Error : \", rmse)\n",
    "    print(\"Mean Absolute Error : \", mae)\n",
    "    print(\"R-squared Score : \", r2)\n",
    "\n",
    "test(client, model)\n",
    "test_time = time.perf_counter()\n",
    "stop = time.perf_counter()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 10. Time Taken"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Creating CUDA Cluster :  7.824385081999935 s\n",
      "Reading :  1.3088564859936014 s\n",
      "Data Type Conversion :  0.5025156859774143 s\n",
      "Applying Constraints :  0.559302720008418 s\n",
      "Feature Engineering :  3.8674576099729165 s\n",
      "Splitting Data :  0.25519451999571174 s\n",
      "Training :  59.19648238609079 s\n",
      "Testing :  3.8579405489144847 s\n",
      "Total :  77.37217173795216 s\n"
     ]
    }
   ],
   "source": [
    "print(\"Creating CUDA Cluster : \", cluster_creation_time - start, \"s\")\n",
    "print(\"Reading : \", read_time - cluster_creation_time, \"s\")\n",
    "print(\"Data Type Conversion : \", dtype_conversion_time - read_time, \"s\")\n",
    "print(\"Applying Constraints : \", apply_constraints_time - dtype_conversion_time, \"s\")\n",
    "print(\"Feature Engineering : \", feature_engg_time - apply_constraints_time, \"s\")\n",
    "print(\"Splitting Data : \", split_data_time - feature_engg_time, \"s\")\n",
    "print(\"Training : \", training_time - split_data_time, \"s\")\n",
    "print(\"Testing : \", test_time - training_time, \"s\")\n",
    "print(\"Total : \", stop - start, \"s\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.8.12"
  },
  "papermill": {
   "duration": 1184.661782,
   "end_time": "2020-12-06T22:54:38.161416",
   "environment_variables": {},
   "exception": null,
   "input_path": "__notebook__.ipynb",
   "output_path": "__notebook__.ipynb",
   "parameters": {},
   "start_time": "2020-12-06T22:34:53.499634",
   "version": "2.1.0"
  },
  "toc-autonumbering": true
 },
 "nbformat": 4,
 "nbformat_minor": 4
}