2. DataPrepAWSSpark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# NYC Taxi Dataset Project - Data Prep AWS Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "## Overall Steps\n",
    "\n",
    "**Step 0:** Prerequisites\n",
    "\n",
    "**Step 1:** Start Spark Cluster\n",
    "\n",
    "**Step 2:** Upload this notebook and packages\n",
    "\n",
    "**Step 3:** Clean Data and generate features\n",
    "\n",
    "**Step 4:** Save RDD to file on s3\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "Note: Step 1 is based on the CS109 [instructions](https://piazza.com/class/icf0cypdc3243c?cid=1369). However there are modifications for optimizing performance for this project\n",
    "\n",
    "### Step 0: Prerequisites\n",
    "\n",
    "1. You need the files CS109.pem and credentials.csv.If you had followed the cs109 instructions (for lab8 or HW5) you will already have these files.\n",
    "\n",
    "2. You will need a directory containing the following files:\n",
    "    \n",
    "    a) CS109.pem\n",
    "    \n",
    "    b) credentials.csv\n",
    "    \n",
    "    c) Setup Project.ipynb\n",
    "    \n",
    "    d) myConfig.json\n",
    "    \n",
    "    e) DataPrepAWSSpark.ipynb (this notebook)\n",
    "    \n",
    "    f) geohash.py\n",
    "3. You must have completed the instructions in Setup Project.ipynb\n",
    "\n",
    "#### Note: The notebook was updated so the datafiles must be downloaded again in order to be properly preprocessed."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "### Step 1: Start Spark cluster and sanity check\n",
    "\n",
    "#### Step 1a) Start your Spark cluster as described in Step 1 from Setup Project (unless your spark cluster is already running)\n",
    "\n"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "export CLUSTER_ID=`aws emr create-cluster --name \"CS109 Proj Spark cluster\" \\\n",
    "--release-label emr-4.2.0 --applications Name=Spark Name=Ganglia --ec2-attributes KeyName=CS109 \\\n",
    "--instance-type c3.2xlarge --instance-count 5 --configurations file://myConfig.json --use-default-roles \\\n",
    "--bootstrap-actions Path=s3://cs109-2015/install-anaconda-emr,Name=Install_Anaconda \\\n",
    "--query 'ClusterId' --output text` && echo $CLUSTER_ID"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Step 1c) Wait for the cluster to be ready: AWS web console has to show \"WAITING\""
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "aws emr describe-cluster --cluster-id $CLUSTER_ID --query 'Cluster.Status.State' --output text"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Step 1d)  Get the cluster master's IP:"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "export DNS_NAME=`aws emr describe-cluster --cluster-id $CLUSTER_ID \\\n",
    "--query 'Cluster.MasterPublicDnsName' --output text` && echo $DNS_NAME"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Step 1e) Run the script to configure Spark "
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "ssh -o ServerAliveInterval=10 -i CS109.pem hadoop@$DNS_NAME 'sh configure-spark.sh'"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Step 1f) Create an SSH tunel to the AWS box and connect to the cluster. This command assumes your SSH key is on the same directory you are invoking the SSH command from. At the end of this you will be in a terminal session on the cluster's master node."
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "ssh -o ServerAliveInterval=10 -i CS109.pem hadoop@$DNS_NAME -L 8989:localhost:8888"
   ]
  },
  {
   "cell_type": "raw",
   "metadata": {},
   "source": [
    "pyspark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 2: Upload this notebook and geohash.py\n",
    "\n",
    "#### Upload this Jupyter Notebook and geohash.py using the console from http://localhost:8989\n",
    "\n",
    "Notes: \n",
    "1. All the steps in Step 2 are to be executed from the Jupyter Notebook iteself\n",
    "2. We will frequently be loading data form the s3 bucket you created in Step 3 of Setup Project (I will use the bucket name: \"sdaultontestbucket\", but replace this with your own\n",
    "3. All the steps in Step 2 are to be executed from the Jupyter Notebook iteself"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Define the name of the AWS S3-bucket where your data is stored and where you want the prepared data to be transferred to. You need to have read/write permission to this bucket."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "bucket = \"cs109-nyc-taxi\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Sanity check: make sure Spark cluster is working"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "10"
      ]
     },
     "execution_count": 2,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import sys\n",
    "rdd = sc.parallelize(xrange(10),10)\n",
    "aa = rdd.map(lambda x: sys.version)\n",
    "aa.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Make sure geohash.py was copied properly"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "'txheec'"
      ]
     },
     "execution_count": 3,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import geohash\n",
    "geohash.encode(40,74,6)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "sc.addPyFile(\"geohash.py\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 3: Clean Data and Calculate Features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Read yellow cab data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "y_rdd = sc.textFile(\"s3://\" + bucket + \"/nyc/yellow*.csv\")\n",
    "y_rdd = y_rdd.map(lambda line: tuple(line.split(',')))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "g_rdd = sc.textFile(\"s3://\" + bucket + \"/nycg/green*.csv\")\n",
    "g_rdd = g_rdd.map(lambda line: tuple(line.split(',')))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Data preparation specification\n",
    "Given, a certain granularity in location (geohash length g), granularity in time (bins per day b) and a chosen wideness (w) of the neighbourhood we want to look at, the aggregated data in the end should have the following columns\n",
    "\n",
    "#### \"geohash\"\n",
    "Geohash with length g (categorical feature). This column will not actually be used in the prediction. It is just an id and can be used when calculating the distances between the geohashes.\n",
    "#### \"time_cat\" \n",
    "Time of the day as a categorical feature. If $b = 24$ (one bin for every hour), then \"time_cat\" for a pickup at 14:20:00 should be the string \"14:00\". If $b = 96$ (one bin for every quarter of an hour), then \"time_cat\" for a pickup at 14:20:00 should be the string `'14:15'`.\n",
    "#### \"time_num\" \n",
    "Time of the day as a (binned!) floating point number between 0 and 1, where the center of the bin is converted to a floating point number between 0 and 1. So if $b = 24$, then \"time_num\" for a pickup at 14:20:00 should be $14.5\\,/\\,24 =  0.6042$. If $b = 96$, it should translate to $14.375\\,/\\,24 = 0.5990$.\n",
    "#### \"time_cos\" \n",
    "The binned \"time_num\" variable converted to a cosine version so that time nicely 'loops' rather than going saw-like when it traverses midnight. See the figure below. This transformation doesn't have any magic powers, but it can make it easier for a model to find the right patterns. \"time_cos\" = $\\cos(\\textrm{time_num} \\cdot 2\\pi)$. So for 24 bins, 14:20:00 would translate to $\\cos(0.6042 \\cdot 2\\pi) = -0.7932$.\n",
    "<img src=\"figures/cyclic-numeric-feature-transformation.png\">\n",
    "#### \"time_sin\" \n",
    "Same thing as 4) but then with sine. So, \"time_sin\" = $\\sin(\\textrm{time_num} \\cdot 2 \\pi)$. For 24 bins per day, 14:20:00 would translate to $\\sin(0.6042 \\cdot 2 \\pi) = -0.6089$.\n",
    "#### \"day_cat\" \n",
    "Day of the week as a categorical feature: \"Monday\", \"Tuesday\", etc.\n",
    "#### \"day_num\" \n",
    "Day of the week as  a numerical feature going from 0 (Monday morning, start of the week) to 1 (Sunday night), European style. With 24 bins, Tuesday afternoon 14:20:00 would translate to $(1 + \\frac{14.5}{24})\\,/\\,7 = 0.2292$.\n",
    "#### \"day_cos\" \n",
    "Binned \"day_num\" variable converted to a cosine version. \"day_cos\" = $\\cos(\\textrm{day_num} \\cdot 2\\pi)$\n",
    "#### \"day_sin\" \n",
    "Binned \"day_num\"variable converted to a sine version. \"day_sin\" = $\\sin(\\textrm{day_num} \\cdot 2\\pi)$\n",
    "#### \"weekend\" \n",
    "0 if weekday, 1 if weekend (Saturday/Sunday)\n",
    "#### Location features \n",
    "Latitude and longitude of the center of the geohash the record was bucketed in.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Helper functions for cleaning and feature extraction/generation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Needed libraries\n",
    "import time\n",
    "from datetime import date\n",
    "import math\n",
    "\n",
    "def date_extractor(date_str,b,minutes_per_bin):\n",
    "    # Takes a datetime object as a parameter\n",
    "    # and extracts and returns a tuple of the form: (as per the data specification)\n",
    "    # (time_cat, time_num, time_cos, time_sin, day_cat, day_num, day_cos, day_sin, weekend)\n",
    "    # Split date string into list of date, time\n",
    "    \n",
    "    d = date_str.split()\n",
    "    \n",
    "    #safety check\n",
    "    if len(d) != 2:\n",
    "        return tuple([None,])\n",
    "    \n",
    "    # TIME (eg. for 16:56:20 and 15 mins per bin)\n",
    "    #list of hour,min,sec (e.g. [16,56,20])\n",
    "    time_list = [int(t) for t in d[1].split(':')]\n",
    "    \n",
    "    #safety check\n",
    "    if len(time_list) != 3:\n",
    "        return tuple([None,])\n",
    "    \n",
    "    # calculate number of minute into the day (eg. 1016)\n",
    "    num_minutes = time_list[0] * 60 + time_list[1]\n",
    "    \n",
    "    # Time of the start of the bin\n",
    "    time_bin = num_minutes / minutes_per_bin     # eg. 1005\n",
    "    hour_bin = num_minutes / 60                  # eg. 16\n",
    "    min_bin = (time_bin * minutes_per_bin) % 60  # eg. 45\n",
    "    \n",
    "    #get time_cat\n",
    "    hour_str = str(hour_bin) if hour_bin / 10 > 0 else \"0\" + str(hour_bin)  # eg. \"16\"\n",
    "    min_str = str(min_bin) if min_bin / 10 > 0 else \"0\" + str(min_bin)      # eg. \"45\"\n",
    "    time_cat = hour_str + \":\" + min_str                                     # eg. \"16:45\"\n",
    "    \n",
    "    # Get a floating point representation of the center of the time bin\n",
    "    time_num = (hour_bin*60 + min_bin + minutes_per_bin / 2.0)/(60*24)      # eg. 0.7065972222222222\n",
    "    \n",
    "    time_cos = math.cos(time_num * 2 * math.pi)\n",
    "    time_sin = math.sin(time_num * 2 * math.pi)\n",
    "    \n",
    "    # DATE\n",
    "    # Parse year, month, day\n",
    "    date_list = d[0].split('-')\n",
    "    d_obj = date(int(date_list[0]),int(date_list[1]),int(date_list[2]))\n",
    "    day_to_str = {0: \"Monday\",\n",
    "                  1: \"Tuesday\",\n",
    "                  2: \"Wednesday\",\n",
    "                  3: \"Thursday\",\n",
    "                  4: \"Friday\",\n",
    "                  5: \"Saturday\",\n",
    "                  6: \"Sunday\"}\n",
    "    day_of_week = d_obj.weekday()\n",
    "    day_cat = day_to_str[day_of_week]\n",
    "    day_num = (day_of_week + time_num)/7.0\n",
    "    day_cos = math.cos(day_num * 2 * math.pi)\n",
    "    day_sin = math.sin(day_num * 2 * math.pi)\n",
    "    \n",
    "    year = d_obj.year\n",
    "    month = d_obj.month\n",
    "    day = d_obj.day\n",
    "    \n",
    "    weekend = 0\n",
    "    #check if it is the weekend\n",
    "    if day_of_week in [5,6]:\n",
    "        weekend = 1\n",
    "       \n",
    "    return (year, month, day, time_cat, time_num, time_cos, time_sin, day_cat, day_num, day_cos, day_sin, weekend)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "def data_cleaner(zipped_row):\n",
    "    # takes a tuple (row,g,b,minutes_per_bin) as a parameter and returns a tuple of the form:\n",
    "    # (time_cat, time_num, time_cos, time_sin, day_cat, day_num, day_cos, day_sin, weekend,geohash)\n",
    "    row = zipped_row[0]\n",
    "    g = zipped_row[1]\n",
    "    b = zipped_row[2]\n",
    "    minutes_per_bin = zipped_row[3]\n",
    "    # The indices of pickup datetime, longitude, and latitude respectively\n",
    "    indices = (1, 6, 5)\n",
    "    \n",
    "    #safety check: make sure row has enough features\n",
    "    if len(row) < 7:\n",
    "        return None\n",
    "    \n",
    "    #extract day of the week and hour\n",
    "    date_str = row[indices[0]]\n",
    "    clean_date = date_extractor(date_str,b,minutes_per_bin)\n",
    "    #get geo hash\n",
    "\n",
    "    latitude = float(row[indices[1]])\n",
    "    longitude = float(row[indices[2]])\n",
    "    location = None\n",
    "    #safety check: make sure latitude and longitude are valid\n",
    "    if latitude < 41.1 and latitude > 40.5 and longitude < -73.6 and longitude > -74.1:\n",
    "        location = geohash.encode(latitude,longitude, g)\n",
    "    else:\n",
    "        return None\n",
    "\n",
    "    return tuple(list(clean_date)+[location])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Specify Parameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "g = 7 #geohash length\n",
    "b = 48 # number of time bins per day\n",
    "# Note: b must evenly divide 60\n",
    "minutes_per_bin = int((24 / float(b)) * 60)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Clean data create and create features as specified above"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "gclean_rdd = g_rdd.map(lambda row: (row, g, b, minutes_per_bin))\\ # Add the parameters to each record\n",
    "                  .map(data_cleaner)\\                # Clean the data and add all the necessary features\n",
    "                  .filter(lambda row: row != None)\\  # Filter out any records that were invalid and returned None\n",
    "                  .map(lambda row: (row,1))\\         # Add a 1 to be able to count\n",
    "                  .reduceByKey(lambda a,b: a + b)\\   # Group by each geohash/time combination and count the number of pickups\n",
    "                  .map(lambda row: (row,'g'))        # Add the color of the cab"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "yclean_rdd = y_rdd.map(lambda row: (row, g, b, minutes_per_bin))\\ # Add the parameters to each record\n",
    "                  .map(data_cleaner)\\                # Clean the data and add all the necessary features\n",
    "                  .filter(lambda row: row != None)\\  # Filter out any records that were invalid and returned None\n",
    "                  .map(lambda row: (row,1))\\         # Add a 1 to be able to count\n",
    "                  .reduceByKey(lambda a,b: a + b)\\   # Group by each geohash/time combination and count the number of pickups\n",
    "                  .map(lambda row: (row,'y'))        # Add the color of the cab"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "#### Combine rows from both rdds"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In this step we combine all the aggregated data from the yellow and the green taxis. We also add a random number which will be used to sample the data before saving it to S3. This is necessary because the aggreggated data - especially of the training set (72 million records) - is still too large to be able to process on a laptop."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import numpy as np\n",
    "combined_rdd = yclean_rdd.union(gclean_rdd)   # Create a combined dataset of yellow and green taxis\n",
    "#get rid of g, y letters and reduce\n",
    "final_rdd = combined_rdd.map(lambda row: row[0])\\  # Get rid of the taxi color\n",
    "                        .reduceByKey(lambda a,b: a + b)\\  # Combine the counts\n",
    "                        .map(lambda (a,b): (a,b,np.random.random()))  # Add a random number"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Step 4: Save RDD to file on s3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "# From the training set (roughly 72 million records) we will take 10% totalling about 7 million\n",
    "train_fraction = 0.1\n",
    "\n",
    "# From the validation set (roughly 6 million records) we will take 25% totalling about 1.5 million\n",
    "valid_fraction = 0.25"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Underneath we create the training, validation and test sets. We do this by filtering on the dates and by sampling down to manageable number of records. We then repartition the data as a single file and save it to AWS S3.\n",
    "* **Training set**: January 2013 to February 2015\n",
    "* **Validation set**: March 2015 to April 2015\n",
    "* **Test set**: May 2015 to June 2015"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "trainset = final_rdd.filter(lambda (a,b,c): ((a[0] < 2015) | (a[1] <= 2)) & (c <= train_fraction))\n",
    "trainset.repartition(1).saveAsTextFile(\"s3n://\" + bucket + \"/trainset7\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "validset = final_rdd.filter(lambda (a,b,c): (a[0] == 2015) & ((a[1] == 3) | (a[1] == 4)) & (c <= valid_fraction))\n",
    "validset.repartition(1).saveAsTextFile(\"s3n://\" + bucket + \"/validset7\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "testset  = final_rdd.filter(lambda (a,b,c): (a[0] == 2015) & ((a[1] == 5) | (a[1] == 6))) # \n",
    "testset.repartition(1).saveAsTextFile(\"s3n://\" + bucket + \"/testset7\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now you can download the files as RDD's to your computer. With the ConvertRDDtoCSV notebook, the files can be converted to csv, so they can be loaded in the machine learning algorithms."
   ]
  }
 ],
 "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
}