Data Prep - Destinations.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Data Prep for Destination Visualization & Prediction"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "<pyspark.context.SparkContext at 0x7fe56905ded0>"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import numpy as np\n",
    "import scipy as sp\n",
    "import pandas as pd\n",
    "from pyspark.sql import SQLContext\n",
    "from pyspark.sql.types import *\n",
    "\n",
    "import geohash\n",
    "\n",
    "from datetime import *\n",
    "from dateutil.parser import parse\n",
    "\n",
    "pd.set_option('display.width', 500)\n",
    "pd.set_option('display.max_columns', 100)\n",
    "pd.set_option('display.notebook_repr_html', True)\n",
    "\n",
    "sc"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#File needed for geohash routines\n",
    "sc.addPyFile(\"geohash.py\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 1. Parse the CSV and construct RDDs"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "    2: pickup dattime\n",
    "    6: pickup long\n",
    "    7: pickup lat\n",
    "    10: dropoff long\n",
    "    11: dropoff lat\n",
    "\"\"\"\n",
    "def yCabParse(strRecord):    \n",
    "    return (parse(strRecord[2]), float(strRecord[6]), float(strRecord[7]), float(strRecord[10]), float(strRecord[11]))\n",
    "\n",
    "yCabRDD = sc.textFile(\"s3://testsetu/nyc/final/yellow/consolidated/pa*\").map(lambda line: tuple(line.split(',')))\n",
    "yCabRDD = yCabRDD.map(lambda record: yCabParse(record))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "    2: pickup dattime\n",
    "    6: pickup long\n",
    "    7: pickup lat\n",
    "    8: dropoff long\n",
    "    9: dropoff lat\n",
    "\"\"\"\n",
    "def gCabParse(strRecord):    \n",
    "    return (parse(strRecord[2]), float(strRecord[6]), float(strRecord[7]), float(strRecord[8]), float(strRecord[9]))\n",
    "\n",
    "gCabRDD = sc.textFile(\"s3://testsetu/nyc/final/green/consolidated/pa*\").map(lambda line: tuple(line.split(',')))\n",
    "gCabRDD = gCabRDD.map(lambda record: gCabParse(record))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "combinedRDD = yCabRDD.union(gCabRDD)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "    2: pickup dattime\n",
    "    6: pickup long\n",
    "    7: pickup lat\n",
    "    8: dropoff long\n",
    "    9: dropoff lat\n",
    "\"\"\"\n",
    "def prepData(record, onlyLocationAgg = False):\n",
    "    \n",
    "    geohashAccuracy = 6\n",
    "    minsPerBin = 48\n",
    "    \n",
    "    pickupDatetime = record[0] \n",
    "    pickupLong = record[1]\n",
    "    pickupLat = record[2]\n",
    "    dropOffLong = record[3]\n",
    "    dropOffLat = record[4]\n",
    "    \n",
    "    if pickupLat < 50 and pickupLat > 35 and pickupLong < -50 and pickupLong > -80:\n",
    "        pickupGeohash = geohash.encode(pickupLat,pickupLong, geohashAccuracy)\n",
    "    else:\n",
    "        return None\n",
    "    \n",
    "    if dropOffLat < 50 and dropOffLat > 35 and dropOffLong < -50 and dropOffLong > -80:\n",
    "        dropOffGeohash = geohash.encode(dropOffLat,dropOffLong, geohashAccuracy)\n",
    "    else:\n",
    "        return None\n",
    "    \n",
    "    \n",
    "    #time_cat\n",
    "    d = pickupDatetime\n",
    "    \n",
    "    totalMinsPerDay = 1440\n",
    "    totalBins = totalMinsPerDay/minsPerBin\n",
    "    \n",
    "    elapsMins = (d.hour)*60 + d.minute\n",
    "    #minsPerBin = totalMinsPerDay/totalBins\n",
    "    currentBin = elapsMins/minsPerBin\n",
    "    binnedHour = d.hour #elapsMins/60\n",
    "    binnedMin = (currentBin * minsPerBin)- (binnedHour * 60)\n",
    "    \n",
    "    binStr = \"\"\n",
    "    \n",
    "    if (binnedHour/10>0):\n",
    "        binStr = str(binnedHour)\n",
    "    else:\n",
    "        binStr = \"0\"+str(binnedHour)\n",
    "    \n",
    "    binStr = binStr + \":\"\n",
    "    \n",
    "    if (binnedMin/10>0):\n",
    "        binStr = binStr + str(binnedMin)\n",
    "    else:\n",
    "        binStr = binStr + \"0\"+str(binnedMin)\n",
    "    \n",
    "    time_num = (binnedHour*60 + binnedMin + minsPerBin / 2.0)/(60*24)  \n",
    "    \n",
    "    #day of week\n",
    "    \n",
    "    \n",
    "    #weekend\n",
    "    if d.weekday() in [5,6]:\n",
    "        weekend = 1\n",
    "    else:\n",
    "        weekend = 0\n",
    "    \n",
    "    if onlyLocationAgg is False:\n",
    "        return ((pickupGeohash, dropOffGeohash,time_num,d.weekday()),1)\n",
    "    else:\n",
    "        return ((pickupGeohash, dropOffGeohash, weekend),1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. Create aggregations across time & locations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### The output of this is used in the predictions notebook. Here we aggregate the data by pickup location, drop off location day of week & time slot."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "combinedCleanRDD = combinedRDD.map(lambda record: prepData(record)).filter(lambda a: a is not None)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "groupedRDD = combinedCleanRDD.reduceByKey(lambda a,b: a+b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "PythonRDD[29] at RDD at PythonRDD.scala:43"
      ]
     },
     "execution_count": 20,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "groupedRDD.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def toCSVLine(record):\n",
    "    data = [record[0][0], record[0][1], record[0][2], record[0][3], record[1]]\n",
    "    return ','.join(str(d) for d in data)\n",
    "\n",
    "csvRDD = groupedRDD.map(toCSVLine)\n",
    "csvRDD.repartition(1).saveAsTextFile('s3://testsetu/nyc/final/groupbydestn/singlefile')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Aggregations only by locations"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Here we aggregate all the data by only pickup & drop off locations"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "combinedCleanRDD = combinedRDD.map(lambda record: prepData(record, True)).filter(lambda a: a is not None)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "groupedRDD = combinedCleanRDD.reduceByKey(lambda a,b: a+b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 148 ms, sys: 116 ms, total: 264 ms\n",
      "Wall time: 32min 56s\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "1091619"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "groupedRDD.cache()\n",
    "%time groupedRDD.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(('dr72p9', 'dr5rec', 0), 68), (('dr5rvf', 'dr5rmn', 0), 8)]"
      ]
     },
     "execution_count": 12,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "groupedRDD.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "#((pickupGeohash, dropOffGeohash, weekend),1)\n",
    "def toCSVLineOnlyLocnAgg(record):\n",
    "    data = [record[0][0], record[0][1], record[0][2], record[1]]\n",
    "    return ','.join(str(d) for d in data)\n",
    "\n",
    "csvRDD = groupedRDD.map(toCSVLineOnlyLocnAgg)\n",
    "csvRDD.repartition(1).saveAsTextFile('s3://testsetu/nyc/final/groupbydestn_only_locn/singlefile')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[u'dr72jh,dr5rm3,1,5',\n",
       " u'dr5rgy,dr5rvr,1,1',\n",
       " u'dr5x1r,dr72jj,0,1',\n",
       " u'dr5ru2,dr72p9,1,11',\n",
       " u'dr5rs5,dr5ruc,1,35']"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "sc.textFile(\"s3://testsetu/nyc/final/groupbydestn_only_locn/singlefile/p*\").map(lambda line: line).take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# Need to fix the CSV file: lat/long needs to be included instead of geohash\n",
    "names = [\"pickup_geohash\",\"dropoff_geohash\",\"weekend\", \"count\"]\n",
    "df=pd.read_csv(\"./tmplocaldata/final/groupbydestn_only_locn/singlefile/part-00000\", header=None, names = names)\n",
    "\n",
    "def decodegeo(geo, which):\n",
    "    if len(geo) >= 6:\n",
    "        geodecoded = geohash.decode(geo)\n",
    "        return geodecoded[which]\n",
    "    else:\n",
    "        return 0\n",
    "    \n",
    "def further_data_prep(df):\n",
    "    df['pickup_lat'] = df['pickup_geohash'].apply(lambda geo: decodegeo(geo, 0))\n",
    "    df['pickup_long'] = df['pickup_geohash'].apply(lambda geo: decodegeo(geo, 1))\n",
    "    df['dropoff_lat'] = df['dropoff_geohash'].apply(lambda geo: decodegeo(geo, 0))\n",
    "    df['dropoff_long'] = df['dropoff_geohash'].apply(lambda geo: decodegeo(geo, 1))    \n",
    "    return df\n",
    "\n",
    "df = further_data_prep(df)\n",
    "df.drop('pickup_geohash', axis=1, inplace=True)\n",
    "df.drop('dropoff_geohash', axis=1, inplace=True)\n",
    "df = df[[\"pickup_lat\",\"pickup_long\", \"dropoff_lat\",\"dropoff_long\",\"weekend\",\"count\"]]\n",
    "df.to_csv(\"pickup_dropoff_aggregated.csv\", index=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Tableau Visualization from the above CSV is avialable [here] (https://public.tableau.com/profile/publish/pickup-destination-coupling/Dashboard1#!/publish-confirm)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<script type='text/javascript' src='https://public.tableau.com/javascripts/api/viz_v1.js'></script><div class='tableauPlaceholder' style='width: 804px; height: 519px;'><noscript><a href='#'><img alt='Where do people go from where? ' src='https:&#47;&#47;public.tableau.com&#47;static&#47;images&#47;pi&#47;pickup-destination-coupling&#47;Dashboard1&#47;1_rss.png' style='border: none' /></a></noscript><object class='tableauViz' width='804' height='519' style='display:none;'><param name='host_url' value='https%3A%2F%2Fpublic.tableau.com%2F' /> <param name='site_root' value='' /><param name='name' value='pickup-destination-coupling&#47;Dashboard1' /><param name='tabs' value='no' /><param name='toolbar' value='yes' /><param name='static_image' value='https:&#47;&#47;public.tableau.com&#47;static&#47;images&#47;pi&#47;pickup-destination-coupling&#47;Dashboard1&#47;1.png' /> <param name='animate_transition' value='yes' /><param name='display_static_image' value='yes' /><param name='display_spinner' value='yes' /><param name='display_overlay' value='yes' /><param name='display_count' value='yes' /><param name='showVizHome' value='no' /><param name='showTabs' value='y' /><param name='bootstrapWhenNotified' value='true' /></object></div>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "%%html\n",
    "<script type='text/javascript' src='https://public.tableau.com/javascripts/api/viz_v1.js'></script><div class='tableauPlaceholder' style='width: 804px; height: 519px;'><noscript><a href='#'><img alt='Where do people go from where? ' src='https:&#47;&#47;public.tableau.com&#47;static&#47;images&#47;pi&#47;pickup-destination-coupling&#47;Dashboard1&#47;1_rss.png' style='border: none' /></a></noscript><object class='tableauViz' width='804' height='519' style='display:none;'><param name='host_url' value='https%3A%2F%2Fpublic.tableau.com%2F' /> <param name='site_root' value='' /><param name='name' value='pickup-destination-coupling&#47;Dashboard1' /><param name='tabs' value='no' /><param name='toolbar' value='yes' /><param name='static_image' value='https:&#47;&#47;public.tableau.com&#47;static&#47;images&#47;pi&#47;pickup-destination-coupling&#47;Dashboard1&#47;1.png' /> <param name='animate_transition' value='yes' /><param name='display_static_image' value='yes' /><param name='display_spinner' value='yes' /><param name='display_overlay' value='yes' /><param name='display_count' value='yes' /><param name='showVizHome' value='no' /><param name='showTabs' value='y' /><param name='bootstrapWhenNotified' value='true' /></object></div>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Appendix"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Machine learning sandbox (for reference only - pls ignore)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "\n",
    "from pyspark.mllib.tree import RandomForest, RandomForestModel\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "sc.textFile(\"s3://testsetu/nyc/final/groupbydestn/singlefile/pa*\").map(lambda line: line).take(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "\"\"\"\n",
    "pickupGeohash, dropOffGeohash,time_num,day_of_week, count\n",
    "\"\"\"\n",
    "def groupedParse(strRecord):    \n",
    "    return (strRecord[0], strRecord[1], float(strRecord[2]), int(strRecord[3]), int(strRecord[4]))\n",
    "\n",
    "gpRDD = sc.textFile(\"s3://testsetu/nyc/final/groupbydestn/singlefile/pa*\").map(lambda line: tuple(line.split(','))).map(lambda x: groupedParse(x))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "gpRDD.cache()\n",
    "gpRDD.count()\n",
    "dropoffGeohashes = gpRDD.map(lambda x: x[1]).distinct().collect()\n",
    "dictLength = len(dropoffGeohashes)\n",
    "dropoffGeohashDict = {}\n",
    "i = 0\n",
    "for gh in dropoffGeohashes:\n",
    "    dropoffGeohashDict[gh] = i\n",
    "    i = i +1\n",
    "\n",
    "broadcastGH = sc.broadcast(dropoffGeohashDict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.mllib.regression import LabeledPoint\n",
    "import math\n",
    "#Create features as labeledpoint\n",
    "\n",
    "\"\"\"\n",
    "0:pickupGeohash\n",
    "1:dropOffGeohash\n",
    "2:time_num\n",
    "3:day_of_week\n",
    "4: count\n",
    "\"\"\"\n",
    "def extractFeaturesforML(record):\n",
    "    #np.array([1.0, 0.0, 3.0])\n",
    "    \n",
    "    count = record[4]\n",
    "    timeCos = math.cos(record[2] * 2 * math.pi)\n",
    "    timeSin = math.sin(record[2] * 2 * math.pi)\n",
    "    \n",
    "    #dayCos = math.cos(key[3] * 2 * math.pi)\n",
    "    #daySin = math.sin(key[3] * 2 * math.pi)\n",
    "    \n",
    "    pickupLatLong = geohash.decode(record[0])\n",
    "    pickupLat=pickupLatLong[0]\n",
    "    pickupLong=pickupLatLong[1]    \n",
    "    \n",
    "    features_ = np.array([record[2], record[3], count, timeCos, timeSin, pickupLat, pickupLong])\n",
    "    \n",
    "    return LabeledPoint(broadcastGH.value[record[1]], features_)\n",
    "\n",
    "gpRDD.map(lambda x: extractFeaturesforML(x)).take(2)\n",
    "featuresLP = gpRDD.map(lambda x: extractFeaturesforML(x)).cache()\n",
    "%time featuresLP.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Sandbox code: not optimal\n",
    "(trainingData, testData) = featuresLP.randomSplit([0.8, 0.2])\n",
    "\n",
    "model = RandomForest.trainClassifier(trainingData, numClasses=dictLength, categoricalFeaturesInfo={},\n",
    "                                     numTrees=10, featureSubsetStrategy=\"auto\",\n",
    "                                     impurity='gini', maxDepth=10, maxBins=32)\n",
    "\n",
    "# Evaluate model on test instances and compute test error\n",
    "predictions = model.predict(testData.map(lambda x: x.features))\n",
    "labelsAndPredictions = testData.map(lambda lp: lp.label).zip(predictions)\n",
    "testErr = labelsAndPredictions.filter(lambda (v, p): v != p).count() / float(testData.count())\n",
    "print('Test Error = ' + str(testErr))\n",
    "print('Learned classification forest model:')\n",
    "#print(model.toDebugString())\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 2",
   "language": "python",
   "name": "python2"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 2
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython2",
   "version": "2.7.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}