DataPrepLocalTest.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "%matplotlib inline\n",
    "import numpy as np\n",
    "import scipy as sp\n",
    "import matplotlib as mpl\n",
    "import matplotlib.cm as cm\n",
    "import matplotlib.pyplot as plt\n",
    "import pandas as pd\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",
    "import seaborn as sns\n",
    "sns.set_style(\"whitegrid\")\n",
    "sns.set_context(\"poster\")\n",
    "import geohash"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "import os\n",
    "os.environ['PYSPARK_PYTHON'] = '/anaconda/bin/python'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "/usr/local/opt/apache-spark/libexec\n"
     ]
    }
   ],
   "source": [
    "import findspark\n",
    "findspark.init()\n",
    "print findspark.find()\n",
    "# Depending on your setup you might have to change this line of code\n",
    "#findspark makes sure I dont need the below on homebrew.\n",
    "#os.environ['SPARK_HOME']=\"/usr/local/Cellar/apache-spark/1.5.1/libexec/\"\n",
    "#the below actually broke my spark, so I removed it. \n",
    "#Depending on how you started the notebook, you might need it.\n",
    "#os.environ['PYSPARK_SUBMIT_ARGS']=\"--master local pyspark --executor-memory 4g\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import pyspark\n",
    "conf = (pyspark.SparkConf()\n",
    "    .setMaster('local')\n",
    "    .setAppName('pyspark'))\n",
    "sc = pyspark.SparkContext(conf=conf)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql import SQLContext\n",
    "sqlsc=SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'2',\n",
       "  u'2013-12-01 00:00:00',\n",
       "  u'2013-12-01 20:44:23',\n",
       "  u'N',\n",
       "  u'1',\n",
       "  u'0',\n",
       "  u'0',\n",
       "  u'-73.957260131835937',\n",
       "  u'40.742355346679687',\n",
       "  u'1',\n",
       "  u'4.00',\n",
       "  u'13',\n",
       "  u'0.5',\n",
       "  u'0.5',\n",
       "  u'0',\n",
       "  u'0',\n",
       "  u'',\n",
       "  u'14',\n",
       "  u'2',\n",
       "  u'',\n",
       "  u'',\n",
       "  u''),\n",
       " (u'2',\n",
       "  u'2013-12-01 00:00:00',\n",
       "  u'2013-12-01 02:53:23',\n",
       "  u'N',\n",
       "  u'1',\n",
       "  u'0',\n",
       "  u'0',\n",
       "  u'0',\n",
       "  u'0',\n",
       "  u'1',\n",
       "  u'3.17',\n",
       "  u'12',\n",
       "  u'0.5',\n",
       "  u'0.5',\n",
       "  u'0',\n",
       "  u'0',\n",
       "  u'',\n",
       "  u'13',\n",
       "  u'2',\n",
       "  u'',\n",
       "  u'',\n",
       "  u'')]"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "g_rdd = sc.textFile('data/green_tripdata_2013-12.csv')\n",
    "g_rdd = g_rdd.map(lambda line: tuple(line.split(',')))\n",
    "g_rdd.take(2)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(u'CMT',\n",
       "  u'2013-01-01 15:11:48',\n",
       "  u'2013-01-01 15:18:10',\n",
       "  u'4',\n",
       "  u'1',\n",
       "  u'-73.978165000000004',\n",
       "  u'40.757976999999997',\n",
       "  u'1',\n",
       "  u'N',\n",
       "  u'-73.989840000000001',\n",
       "  u'40.751173000000001',\n",
       "  u'CSH',\n",
       "  u'6.5',\n",
       "  u'0',\n",
       "  u'0.5',\n",
       "  u'0',\n",
       "  u'0',\n",
       "  u'7'),\n",
       " (u'CMT',\n",
       "  u'2013-01-06 00:18:35',\n",
       "  u'2013-01-06 00:22:54',\n",
       "  u'1',\n",
       "  u'1.5',\n",
       "  u'-74.006680000000003',\n",
       "  u'40.731780999999998',\n",
       "  u'1',\n",
       "  u'N',\n",
       "  u'-73.994499000000005',\n",
       "  u'40.750658999999999',\n",
       "  u'CSH',\n",
       "  u'6',\n",
       "  u'0.5',\n",
       "  u'0.5',\n",
       "  u'0',\n",
       "  u'0',\n",
       "  u'7')]"
      ]
     },
     "execution_count": 8,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "y_rdd = sc.textFile('data/yellow_tripdata_2013-01.csv')\n",
    "y_rdd = y_rdd.map(lambda line: tuple(line.split(',')))\n",
    "y_rdd.take(2)"
   ]
  },
  {
   "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. I'm not sure if it adds much predictive power, but only one way to find out. \"time_cos\" = cos(time_num * 2 * Pi). So for 24 bins, 14:20:00 would translate to cos(0.6042 * 2 * Pi) = -0.7932.\n",
    "#### \"time_sin\" \n",
    "Same thing as 4) but then with sine. So, \"time_sin\" = sin(time_num * 2 * Pi). For 24 bins per day, 14:20:00 would translate to sin(0.6042 * 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). Yeah, it's European to start the week on Monday. Whatever, haha :P. Anyway, with 24 bins, Tuesday afternoon 14:20:00 would translate to (1 + 14.5/24)/7 = 0.2292.\n",
    "#### \"day_cos\" \n",
    "Binned \"day_num\" variable converted to a cosine version. \"day_cos\" = cos(day_num * 2 * Pi)\n",
    "#### \"day_sin\" \n",
    "Binned \"day_num\"variable converted to a sine version. \"day_sin\" = sin(day_num * 2 * Pi)\n",
    "#### \"weekend\" \n",
    "0 if weekday, 1 if weekend (Saturday/Sunday)\n",
    "#### \"holiday\" (not included yet)\n",
    "0 if normal day, 1 if holiday. Not sure if it's easy to find all the holidays in NYC from 2013 - 2015…\n",
    "to xxxxx) \n",
    "#### Location features \n",
    "For each binned location, there should be a feature. The name of each feature should be the geohash. To calculate the value of the feature, you need the distance between the location of the row and the location of the column. The value should be calculated as 1/(distance_between_locations_in_km + 1)^w (i.e. inverse distance weighting). The larger w gets, the quicker everything drops to 0. So, for w = 4, the value for the location itself will be 1/(0 + 1)^4 = 1. For a location 2 km away, it will be 1/(2 + 1)^4 = 1/3^4 = 0.012. So information about pickup rates from locations 2 km away are pretty much not taken into account at all when predicting for a certain location. If w = 2, the value for a location 2 km away is a bit larger: 0.11. If w = 1, it is even larger: 0.33. So w is a parameter with which we can determine how 'smooth' are predictions are going to be. The larger we take w, the more we look at the neighbourhoods, the smoother the predictions will be. What the optimal w is, I don't know. I think we should try a few different options and see which works best on the validation set.\n",
    "\n",
    "In the end, this should result in a number of records equal to 7 * b * x where x is the number of locations. For b = 24 and a geohash length g of 7, this should come down to about 24 * 7 * 10000 = roughly 1 to 2 million records.\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import time\n",
    "from datetime import date\n",
    "import math\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)\n",
    "\n",
    "import geohash\n",
    "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": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "# Clean Data\n",
    "# Create Data as Specified\n",
    "# Parameters\n",
    "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)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "gclean_rdd = g_rdd.map(lambda row: (row, g, b, minutes_per_bin))\\\n",
    "                .map(data_cleaner)\\\n",
    "                .map(lambda row: (row,1))\\\n",
    "                .reduceByKey(lambda a,b: a + b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "412299"
      ]
     },
     "execution_count": 13,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "gclean_rdd.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[((2013,\n",
       "   12,\n",
       "   4,\n",
       "   '23:00',\n",
       "   0.96875,\n",
       "   0.9807852804032303,\n",
       "   -0.19509032201612872,\n",
       "   'Wednesday',\n",
       "   0.42410714285714285,\n",
       "   -0.8884456359788722,\n",
       "   0.45898186446753797,\n",
       "   0,\n",
       "   'dr5rknq'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   25,\n",
       "   '18:00',\n",
       "   0.7604166666666666,\n",
       "   0.06540312923014237,\n",
       "   -0.9978589232386036,\n",
       "   'Wednesday',\n",
       "   0.3943452380952381,\n",
       "   -0.7876268391214754,\n",
       "   0.6161525479096176,\n",
       "   0,\n",
       "   'dr72jpt'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   19,\n",
       "   '00:00',\n",
       "   0.010416666666666666,\n",
       "   0.9978589232386035,\n",
       "   0.06540312923014306,\n",
       "   'Thursday',\n",
       "   0.4300595238095238,\n",
       "   -0.9049862302634799,\n",
       "   0.4254408572686641,\n",
       "   0,\n",
       "   'dr72w3b'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   29,\n",
       "   '16:30',\n",
       "   0.6979166666666666,\n",
       "   -0.3214394653031618,\n",
       "   -0.9469301294951056,\n",
       "   'Sunday',\n",
       "   0.9568452380952381,\n",
       "   0.963463687815439,\n",
       "   -0.2678389857005029,\n",
       "   1,\n",
       "   'dr5rt54'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   29,\n",
       "   '17:00',\n",
       "   0.71875,\n",
       "   -0.19509032201612866,\n",
       "   -0.9807852804032303,\n",
       "   'Sunday',\n",
       "   0.9598214285714286,\n",
       "   0.9683035221222615,\n",
       "   -0.24977647816722684,\n",
       "   1,\n",
       "   'dr5rmjx'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   31,\n",
       "   '23:30',\n",
       "   0.9895833333333334,\n",
       "   0.9978589232386035,\n",
       "   -0.0654031292301428,\n",
       "   'Tuesday',\n",
       "   0.2842261904761905,\n",
       "   -0.21339578555795022,\n",
       "   0.9769658329266716,\n",
       "   0,\n",
       "   'dr5rktc'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   24,\n",
       "   '21:30',\n",
       "   0.90625,\n",
       "   0.8314696123025452,\n",
       "   -0.5555702330196022,\n",
       "   'Tuesday',\n",
       "   0.27232142857142855,\n",
       "   -0.1397903395354992,\n",
       "   0.9901811253364456,\n",
       "   0,\n",
       "   'dr5rkpr'),\n",
       "  2),\n",
       " ((2013,\n",
       "   12,\n",
       "   4,\n",
       "   '12:30',\n",
       "   0.53125,\n",
       "   -0.9807852804032304,\n",
       "   -0.19509032201612836,\n",
       "   'Wednesday',\n",
       "   0.36160714285714285,\n",
       "   -0.6451719835420874,\n",
       "   0.7640373758216077,\n",
       "   0,\n",
       "   'dr5rvxp'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   31,\n",
       "   '23:00',\n",
       "   0.96875,\n",
       "   0.9807852804032303,\n",
       "   -0.19509032201612872,\n",
       "   'Tuesday',\n",
       "   0.28125,\n",
       "   -0.1950903220161282,\n",
       "   0.9807852804032304,\n",
       "   0,\n",
       "   'dr5rtmy'),\n",
       "  2),\n",
       " ((2013,\n",
       "   12,\n",
       "   28,\n",
       "   '13:30',\n",
       "   0.5729166666666666,\n",
       "   -0.8968727415326884,\n",
       "   -0.4422886902190011,\n",
       "   'Saturday',\n",
       "   0.7961309523809524,\n",
       "   0.28580783578619784,\n",
       "   -0.9582869512850573,\n",
       "   1,\n",
       "   'dr5rkqc'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   3,\n",
       "   '00:30',\n",
       "   0.03125,\n",
       "   0.9807852804032304,\n",
       "   0.19509032201612825,\n",
       "   'Tuesday',\n",
       "   0.14732142857142858,\n",
       "   0.6013170912984058,\n",
       "   0.7990104853582489,\n",
       "   0,\n",
       "   'dr5rv7h'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   30,\n",
       "   '12:00',\n",
       "   0.5104166666666666,\n",
       "   -0.9978589232386035,\n",
       "   -0.06540312923014287,\n",
       "   'Monday',\n",
       "   0.07291666666666666,\n",
       "   0.8968727415326884,\n",
       "   0.4422886902190012,\n",
       "   0,\n",
       "   'dr72rp7'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   5,\n",
       "   '15:00',\n",
       "   0.6354166666666666,\n",
       "   -0.6593458151000691,\n",
       "   -0.7518398074789773,\n",
       "   'Thursday',\n",
       "   0.519345238095238,\n",
       "   -0.9926219238595809,\n",
       "   -0.12125063411506089,\n",
       "   0,\n",
       "   'dr72mt2'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   28,\n",
       "   '11:00',\n",
       "   0.46875,\n",
       "   -0.9807852804032304,\n",
       "   0.19509032201612858,\n",
       "   'Saturday',\n",
       "   0.78125,\n",
       "   0.19509032201612828,\n",
       "   -0.9807852804032304,\n",
       "   1,\n",
       "   'dr5rs3d'),\n",
       "  2),\n",
       " ((2013,\n",
       "   12,\n",
       "   2,\n",
       "   '07:00',\n",
       "   0.3020833333333333,\n",
       "   -0.32143946530316136,\n",
       "   0.9469301294951057,\n",
       "   'Monday',\n",
       "   0.043154761904761904,\n",
       "   0.963463687815439,\n",
       "   0.2678389857005031,\n",
       "   0,\n",
       "   'dr72jkq'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   25,\n",
       "   '19:00',\n",
       "   0.8020833333333334,\n",
       "   0.3214394653031615,\n",
       "   -0.9469301294951057,\n",
       "   'Wednesday',\n",
       "   0.40029761904761907,\n",
       "   -0.8101147350670905,\n",
       "   0.586271367224409,\n",
       "   0,\n",
       "   'dr5rth2'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   6,\n",
       "   '20:00',\n",
       "   0.84375,\n",
       "   0.5555702330196018,\n",
       "   -0.8314696123025456,\n",
       "   'Friday',\n",
       "   0.6919642857142857,\n",
       "   -0.35662153266231295,\n",
       "   -0.9342489402945999,\n",
       "   0,\n",
       "   'dr72m3d'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   15,\n",
       "   '19:00',\n",
       "   0.8020833333333334,\n",
       "   0.3214394653031615,\n",
       "   -0.9469301294951057,\n",
       "   'Sunday',\n",
       "   0.9717261904761905,\n",
       "   0.9842617686733542,\n",
       "   -0.17671663964664053,\n",
       "   1,\n",
       "   'dr72jm9'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   29,\n",
       "   '06:30',\n",
       "   0.28125,\n",
       "   -0.1950903220161282,\n",
       "   0.9807852804032304,\n",
       "   'Sunday',\n",
       "   0.8973214285714286,\n",
       "   0.7990104853582488,\n",
       "   -0.6013170912984062,\n",
       "   1,\n",
       "   'dr72jjt'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   30,\n",
       "   '20:00',\n",
       "   0.84375,\n",
       "   0.5555702330196018,\n",
       "   -0.8314696123025456,\n",
       "   'Monday',\n",
       "   0.12053571428571429,\n",
       "   0.7266603220340271,\n",
       "   0.6869969260349016,\n",
       "   0,\n",
       "   'dr5ry9p'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   17,\n",
       "   '14:00',\n",
       "   0.59375,\n",
       "   -0.8314696123025455,\n",
       "   -0.555570233019602,\n",
       "   'Tuesday',\n",
       "   0.22767857142857142,\n",
       "   0.13979033953549952,\n",
       "   0.9901811253364455,\n",
       "   0,\n",
       "   'dr5rkn0'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   7,\n",
       "   '00:30',\n",
       "   0.03125,\n",
       "   0.9807852804032304,\n",
       "   0.19509032201612825,\n",
       "   'Saturday',\n",
       "   0.71875,\n",
       "   -0.19509032201612866,\n",
       "   -0.9807852804032303,\n",
       "   1,\n",
       "   'dr5rkrs'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   7,\n",
       "   '18:30',\n",
       "   0.78125,\n",
       "   0.19509032201612828,\n",
       "   -0.9807852804032304,\n",
       "   'Saturday',\n",
       "   0.8258928571428571,\n",
       "   0.4589818644675373,\n",
       "   -0.8884456359788725,\n",
       "   1,\n",
       "   'dr72wbf'),\n",
       "  2),\n",
       " ((2013,\n",
       "   12,\n",
       "   24,\n",
       "   '21:00',\n",
       "   0.8854166666666666,\n",
       "   0.7518398074789769,\n",
       "   -0.6593458151000694,\n",
       "   'Tuesday',\n",
       "   0.2693452380952381,\n",
       "   -0.1212506341150614,\n",
       "   0.9926219238595808,\n",
       "   0,\n",
       "   'dr72j3z'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   24,\n",
       "   '22:30',\n",
       "   0.9479166666666666,\n",
       "   0.9469301294951056,\n",
       "   -0.32143946530316186,\n",
       "   'Tuesday',\n",
       "   0.2782738095238095,\n",
       "   -0.17671663964663958,\n",
       "   0.9842617686733544,\n",
       "   0,\n",
       "   'dr72m7r'),\n",
       "  3),\n",
       " ((2013,\n",
       "   12,\n",
       "   23,\n",
       "   '14:00',\n",
       "   0.59375,\n",
       "   -0.8314696123025455,\n",
       "   -0.555570233019602,\n",
       "   'Monday',\n",
       "   0.08482142857142858,\n",
       "   0.861312628232409,\n",
       "   0.5080753452465294,\n",
       "   0,\n",
       "   'dr5rt5r'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   20,\n",
       "   '04:00',\n",
       "   0.17708333333333334,\n",
       "   0.44228869021900125,\n",
       "   0.8968727415326884,\n",
       "   'Friday',\n",
       "   0.5967261904761905,\n",
       "   -0.8209357053342324,\n",
       "   -0.5710206368489552,\n",
       "   0,\n",
       "   'dr72j3x'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   24,\n",
       "   '20:30',\n",
       "   0.8645833333333334,\n",
       "   0.6593458151000691,\n",
       "   -0.7518398074789773,\n",
       "   'Tuesday',\n",
       "   0.2663690476190476,\n",
       "   -0.10266852999422842,\n",
       "   0.994715624160405,\n",
       "   0,\n",
       "   'dr72qqt'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   18,\n",
       "   '11:00',\n",
       "   0.46875,\n",
       "   -0.9807852804032304,\n",
       "   0.19509032201612858,\n",
       "   'Wednesday',\n",
       "   0.35267857142857145,\n",
       "   -0.6013170912984058,\n",
       "   0.7990104853582489,\n",
       "   0,\n",
       "   'dr72hgv'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   10,\n",
       "   '21:30',\n",
       "   0.90625,\n",
       "   0.8314696123025452,\n",
       "   -0.5555702330196022,\n",
       "   'Tuesday',\n",
       "   0.27232142857142855,\n",
       "   -0.1397903395354992,\n",
       "   0.9901811253364456,\n",
       "   0,\n",
       "   'dr5rvvk'),\n",
       "  3),\n",
       " ((2013,\n",
       "   12,\n",
       "   29,\n",
       "   '22:30',\n",
       "   0.9479166666666666,\n",
       "   0.9469301294951056,\n",
       "   -0.32143946530316186,\n",
       "   'Sunday',\n",
       "   0.9925595238095238,\n",
       "   0.9989074228728941,\n",
       "   -0.046732863484202206,\n",
       "   1,\n",
       "   'dr72qbw'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   31,\n",
       "   '20:00',\n",
       "   0.84375,\n",
       "   0.5555702330196018,\n",
       "   -0.8314696123025456,\n",
       "   'Tuesday',\n",
       "   0.26339285714285715,\n",
       "   -0.08405052492924749,\n",
       "   0.9964614941176192,\n",
       "   0,\n",
       "   'dr72np2'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   12,\n",
       "   '22:00',\n",
       "   0.9270833333333334,\n",
       "   0.8968727415326883,\n",
       "   -0.4422886902190014,\n",
       "   'Thursday',\n",
       "   0.5610119047619048,\n",
       "   -0.9274171786569749,\n",
       "   -0.37402857742682805,\n",
       "   0,\n",
       "   'dr5rkuj'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   13,\n",
       "   '23:30',\n",
       "   0.9895833333333334,\n",
       "   0.9978589232386035,\n",
       "   -0.0654031292301428,\n",
       "   'Friday',\n",
       "   0.712797619047619,\n",
       "   -0.23162662925103128,\n",
       "   -0.9728047618210991,\n",
       "   0,\n",
       "   'dr5rth5'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   6,\n",
       "   '07:30',\n",
       "   0.3229166666666667,\n",
       "   -0.44228869021900113,\n",
       "   0.8968727415326884,\n",
       "   'Friday',\n",
       "   0.6175595238095238,\n",
       "   -0.7393793371649462,\n",
       "   -0.6732890878170572,\n",
       "   0,\n",
       "   'dr5rvc4'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   12,\n",
       "   '10:00',\n",
       "   0.4270833333333333,\n",
       "   -0.8968727415326883,\n",
       "   0.4422886902190013,\n",
       "   'Thursday',\n",
       "   0.48958333333333337,\n",
       "   -0.9978589232386035,\n",
       "   0.06540312923014312,\n",
       "   0,\n",
       "   'dr72jn1'),\n",
       "  3),\n",
       " ((2013,\n",
       "   12,\n",
       "   26,\n",
       "   '10:30',\n",
       "   0.4479166666666667,\n",
       "   -0.9469301294951056,\n",
       "   0.32143946530316175,\n",
       "   'Thursday',\n",
       "   0.4925595238095238,\n",
       "   -0.9989074228728941,\n",
       "   0.04673286348420253,\n",
       "   0,\n",
       "   'dr5rkr9'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   27,\n",
       "   '18:00',\n",
       "   0.7604166666666666,\n",
       "   0.06540312923014237,\n",
       "   -0.9978589232386036,\n",
       "   'Friday',\n",
       "   0.6800595238095238,\n",
       "   -0.42544085726866415,\n",
       "   -0.9049862302634798,\n",
       "   0,\n",
       "   'dr72m6y'),\n",
       "  2),\n",
       " ((2013,\n",
       "   12,\n",
       "   30,\n",
       "   '18:30',\n",
       "   0.78125,\n",
       "   0.19509032201612828,\n",
       "   -0.9807852804032304,\n",
       "   'Monday',\n",
       "   0.11160714285714286,\n",
       "   0.7640373758216075,\n",
       "   0.6451719835420876,\n",
       "   0,\n",
       "   'dr5rkqw'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   25,\n",
       "   '14:30',\n",
       "   0.6145833333333334,\n",
       "   -0.7518398074789774,\n",
       "   -0.6593458151000688,\n",
       "   'Wednesday',\n",
       "   0.37351190476190477,\n",
       "   -0.7004645364217583,\n",
       "   0.7136872096482122,\n",
       "   0,\n",
       "   'dr72jt2'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   25,\n",
       "   '20:00',\n",
       "   0.84375,\n",
       "   0.5555702330196018,\n",
       "   -0.8314696123025456,\n",
       "   'Wednesday',\n",
       "   0.40625,\n",
       "   -0.8314696123025453,\n",
       "   0.5555702330196022,\n",
       "   0,\n",
       "   'dr5ryg6'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   28,\n",
       "   '19:00',\n",
       "   0.8020833333333334,\n",
       "   0.3214394653031615,\n",
       "   -0.9469301294951057,\n",
       "   'Saturday',\n",
       "   0.8288690476190476,\n",
       "   0.4755145427755747,\n",
       "   -0.8797078603769186,\n",
       "   1,\n",
       "   'dr72jh4'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   13,\n",
       "   '00:30',\n",
       "   0.03125,\n",
       "   0.9807852804032304,\n",
       "   0.19509032201612825,\n",
       "   'Friday',\n",
       "   0.5758928571428571,\n",
       "   -0.8884456359788725,\n",
       "   -0.45898186446753736,\n",
       "   0,\n",
       "   'dr5rt5h'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   27,\n",
       "   '21:30',\n",
       "   0.90625,\n",
       "   0.8314696123025452,\n",
       "   -0.5555702330196022,\n",
       "   'Friday',\n",
       "   0.7008928571428571,\n",
       "   -0.30367674510961506,\n",
       "   -0.9527751227228962,\n",
       "   0,\n",
       "   'dr5rksc'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   24,\n",
       "   '04:00',\n",
       "   0.17708333333333334,\n",
       "   0.44228869021900125,\n",
       "   0.8968727415326884,\n",
       "   'Tuesday',\n",
       "   0.1681547619047619,\n",
       "   0.4918809440263435,\n",
       "   0.8706624701362481,\n",
       "   0,\n",
       "   'dr72myu'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   21,\n",
       "   '22:30',\n",
       "   0.9479166666666666,\n",
       "   0.9469301294951056,\n",
       "   -0.32143946530316186,\n",
       "   'Saturday',\n",
       "   0.849702380952381,\n",
       "   0.5862713672244091,\n",
       "   -0.8101147350670904,\n",
       "   1,\n",
       "   'dr5rknn'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   10,\n",
       "   '18:30',\n",
       "   0.78125,\n",
       "   0.19509032201612828,\n",
       "   -0.9807852804032304,\n",
       "   'Tuesday',\n",
       "   0.2544642857142857,\n",
       "   -0.028046256275868896,\n",
       "   0.9996066263830529,\n",
       "   0,\n",
       "   'dr72jjm'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   6,\n",
       "   '11:00',\n",
       "   0.46875,\n",
       "   -0.9807852804032304,\n",
       "   0.19509032201612858,\n",
       "   'Friday',\n",
       "   0.6383928571428571,\n",
       "   -0.6451719835420883,\n",
       "   -0.7640373758216069,\n",
       "   0,\n",
       "   'dr72qzb'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   16,\n",
       "   '10:00',\n",
       "   0.4270833333333333,\n",
       "   -0.8968727415326883,\n",
       "   0.4422886902190013,\n",
       "   'Monday',\n",
       "   0.06101190476190476,\n",
       "   0.9274171786569749,\n",
       "   0.37402857742682805,\n",
       "   0,\n",
       "   'dr5ryhb'),\n",
       "  1),\n",
       " ((2013,\n",
       "   12,\n",
       "   15,\n",
       "   '22:00',\n",
       "   0.9270833333333334,\n",
       "   0.8968727415326883,\n",
       "   -0.4422886902190014,\n",
       "   'Sunday',\n",
       "   0.9895833333333333,\n",
       "   0.9978589232386035,\n",
       "   -0.06540312923014369,\n",
       "   1,\n",
       "   'dr5ry6h'),\n",
       "  1)]"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "gclean_rdd.take(50)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "yclean_rdd = y_rdd.map(lambda row: (row, g, b, minutes_per_bin))\\\n",
    "                .map(data_cleaner)\\\n",
    "                .map(lambda row: (row,1))\\\n",
    "                .reduceByKey(lambda a,b: a + b)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "859235"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "yclean_rdd.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(('00:10',\n",
       "   0.6666666666666666,\n",
       "   -0.5000000000000004,\n",
       "   -0.8660254037844384,\n",
       "   'Wednesday',\n",
       "   0.28968253968253965,\n",
       "   -0.24675739769029345,\n",
       "   0.969077286229078,\n",
       "   0,\n",
       "   'dr5rumv'),\n",
       "  35),\n",
       " (('01:19',\n",
       "   1.8166666666666667,\n",
       "   0.4067366430758003,\n",
       "   -0.9135454576426009,\n",
       "   'Saturday',\n",
       "   0.7250992063492063,\n",
       "   -0.15581877706340935,\n",
       "   -0.9877856592978376,\n",
       "   1,\n",
       "   'dr5rkhf'),\n",
       "  3),\n",
       " (('00:05',\n",
       "   0.5833333333333334,\n",
       "   -0.8660254037844386,\n",
       "   -0.5000000000000001,\n",
       "   'Tuesday',\n",
       "   0.1463293650793651,\n",
       "   0.6062858642634977,\n",
       "   0.7952467860948975,\n",
       "   0,\n",
       "   'dr5revn'),\n",
       "  1),\n",
       " (('00:19',\n",
       "   0.8166666666666667,\n",
       "   0.40673664307579976,\n",
       "   -0.9135454576426011,\n",
       "   'Monday',\n",
       "   0.00486111111111111,\n",
       "   0.9995335908367129,\n",
       "   0.030538513209822652,\n",
       "   0,\n",
       "   'dr5rvth'),\n",
       "  3),\n",
       " (('00:22',\n",
       "   0.8666666666666667,\n",
       "   0.6691306063588585,\n",
       "   -0.743144825477394,\n",
       "   'Friday',\n",
       "   0.5765873015873016,\n",
       "   -0.8864344987859896,\n",
       "   -0.46285405838345145,\n",
       "   0,\n",
       "   'dr5rkw8'),\n",
       "  2)]"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "yclean_rdd.take(5)"
   ]
  },
  {
   "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
}