ENSF612_Spark_Project_1.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# ![Spark Logo](http://spark-mooc.github.io/web-assets/images/ta_Spark-logo-small.png) + ![Python Logo](http://spark-mooc.github.io/web-assets/images/python-logo-master-v3-TM-flattened_small.png)\n",
    " **Exploratory Analysis of Deerfoot Trail Commute Times**\n",
    "#### This lab will build on the techniques covered in the Spark tutorial to develop a simple application to compute some stats on commute times on Deerfoot Trail.  We will use the commute times and accidents data collected for Deerfoot Trail for the period September 2013 to April 2014.\n",
    "#### ** During this lab we will cover: **\n",
    "#### *Part 1:* Creating a base RDD and pair RDDs\n",
    "#### *Part 2:* Counting with pair RDDs\n",
    "#### *Part 3:* Finding mean values\n",
    "#### *Part 4:* Compute basic stats about the Deerfoot Trail data\n",
    "#### Note that, for reference, you can look up the details of the relevant methods in [Spark's Python API](https://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ** Part 1: Creating a base RDD and pair RDDs **"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### In this part of the lab, we will explore creating a base RDD with `parallelize` and using pair RDDs to count words."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (1a) Create a base RDD **\n",
    "#### We'll start by generating a base RDD by using a Python list and the `sc.parallelize` method.  Then we'll print out the type of the base RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:sparkhpc.sparkjob:Submitted batch job 3404\n",
      "\n",
      "INFO:sparkhpc.sparkjob:Submitted cluster 1\n"
     ]
    }
   ],
   "source": [
    "import os\n",
    "import atexit\n",
    "import sys\n",
    "\n",
    "import pyspark\n",
    "from pyspark.context import SparkContext\n",
    "from pyspark.sql import SQLContext\n",
    "import findspark\n",
    "from sparkhpc import sparkjob\n",
    "\n",
    "#Exit handler to clean up the Spark cluster if the script exits or crashes\n",
    "def exitHandler(sj,sc):\n",
    "    try:\n",
    "        print('Trapped Exit cleaning up Spark Context')\n",
    "        sc.stop()\n",
    "    except:\n",
    "        pass\n",
    "    try:\n",
    "        print('Trapped Exit cleaning up Spark Job')\n",
    "        sj.stop()\n",
    "    except:\n",
    "        pass\n",
    "\n",
    "findspark.init()\n",
    "\n",
    "#Parameters for the Spark cluster\n",
    "nodes=3\n",
    "tasks_per_node=8 \n",
    "memory_per_task=1024 #1 gig per process, adjust accordingly\n",
    "# Please estimate walltime carefully to keep unused Spark clusters from sitting \n",
    "# idle so that others may use the resources.\n",
    "walltime=\"1:00\" #1 hour\n",
    "os.environ['SBATCH_PARTITION']='cpu24' #Set the appropriate ARC partition\n",
    "\n",
    "sj = sparkjob.sparkjob(\n",
    "     ncores=nodes*tasks_per_node,\n",
    "     cores_per_executor=tasks_per_node,\n",
    "     memory_per_core=memory_per_task,\n",
    "     walltime=walltime\n",
    "    )\n",
    "\n",
    "sj.wait_to_start()\n",
    "sc = sj.start_spark()\n",
    "\n",
    "#Register the exit handler                                                                                                     \n",
    "atexit.register(exitHandler,sj,sc)\n",
    "\n",
    "#You need this line if you want to use SparkSQL\n",
    "sqlCtx=SQLContext(sc)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<class 'pyspark.rdd.RDD'>\n"
     ]
    }
   ],
   "source": [
    "daysList = ['sunday', 'monday', 'tuesday', 'tuesday', 'friday']\n",
    "daysRDD = sc.parallelize(daysList, 4)\n",
    "# Print out the type of daysRDD\n",
    "print(type(daysRDD))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:195\n",
      "2.4.0\n",
      "vaibhav.jadhav\n",
      "['sunday', 'monday', 'tuesday', 'tuesday', 'friday']\n"
     ]
    }
   ],
   "source": [
    "print(daysRDD) # prints RDD\n",
    "print(sc.version) #retrieves the version of spark\n",
    "print(str(sc.sparkUser())) #retrieves the name of user running spark context\n",
    "print(daysRDD.collect()) #returns the list of elements in RDD"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (1b) Pluralize and test **\n",
    "#### Let's use a `map()` transformation to add the letter 's' to each string in the base RDD we just created. We'll define a Python function that returns the word with an 's' at the end of the word.  Please replace `<FILL IN>` with your solution.  The print statement is a test of your function.\n",
    "#### This is the general form that exercises will take.  Exercises will include an explanation of what is expected, followed by code cells where one cell will have one or more `<FILL IN>` sections.  The cell that needs to be modified will have `# TODO: Replace <FILL IN> with appropriate code` on its first line."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "mondays\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "def makePlural(word):\n",
    "    \"\"\"Adds an 's' to `word`.\n",
    "\n",
    "    Note:\n",
    "        This is a simple function that only adds an 's'.  No attempt is made to follow proper\n",
    "        pluralization rules.\n",
    "\n",
    "    Args:\n",
    "        word (str): A string.\n",
    "\n",
    "    Returns:\n",
    "        str: A string with 's' added to it.\n",
    "    \"\"\"\n",
    "    return word + 's'\n",
    "    \n",
    "\n",
    "print(makePlural('monday'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "##### Using RDD map() function"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "['sundays', 'mondays', 'tuesdays', 'tuesdays', 'fridays']"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "daysRDD.map(lambda x: x + 's').collect()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (1c) Apply `makePlural` to the base RDD **\n",
    "#### Now pass each item in the base RDD into a [map()](http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD.map) transformation that applies the `makePlural()` function to each element. And then call the [collect()](http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD.collect) action to see the transformed RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['sundays', 'mondays', 'tuesdays', 'tuesdays', 'fridays']\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "pluralRDD = daysRDD.map(makePlural)\n",
    "print(pluralRDD.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (1d) Pass a `lambda` function to `map` **\n",
    "#### Let's create the same RDD using a `lambda` function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['sundays', 'mondays', 'tuesdays', 'tuesdays', 'fridays']\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "pluralLambdaRDD = daysRDD.map(lambda x: makePlural(x))\n",
    "print(pluralLambdaRDD.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (1e) Length of each word **\n",
    "#### Now use `map()` and a `lambda` function to return the first character in each word.  We'll `collect` this result directly into a variable."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "['s', 'm', 't', 't', 'f']\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "pluralFirstChars = (pluralRDD.map(lambda x: x[0]).collect())\n",
    "print(pluralFirstChars)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (1f) Pair RDDs **\n",
    "#### Often we would need to deal with pair RDDs.  A pair RDD is an RDD where each element is a pair tuple `(k, v)` where `k` is the key and `v` is the value. In this example, we will create a pair consisting of `('<day>', 1)` for each word element in the RDD.\n",
    "#### We can create the pair RDD using the `map()` transformation with a `lambda()` function to create a new RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('sunday', 1), ('monday', 1), ('tuesday', 1), ('tuesday', 1), ('friday', 1)]\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "dayPairs = daysRDD.map(lambda x: (x, 1))\n",
    "print(dayPairs.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ** Part 2: Counting with pair RDDs **"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### Now, let's count the number of times a particular day appears in the RDD. There are multiple ways to perform the counting, but some are much less efficient than others.\n",
    "#### A naive approach would be to `collect()` all of the elements and count them in the driver program. While this approach could work for small datasets, we want an approach that will work for any size dataset including terabyte- or petabyte-sized datasets. In addition, performing all of the work in the driver program is slower than performing it in parallel in the workers. For these reasons, we will use data parallel operations."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (2a) `groupByKey()` approach **\n",
    "#### An approach you might first consider (we'll see shortly that there are better ways) is based on using the [groupByKey()](http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD.groupByKey) transformation. As the name implies, the `groupByKey()` transformation groups all the elements of the RDD with the same key into a single list in one of the partitions. There are two problems with using `groupByKey()`:\n",
    "  + #### The operation requires a lot of data movement to move all the values into the appropriate partitions.\n",
    "  + #### The lists can be very large. Consider a word count of English Wikipedia: the lists for common words (e.g., the, a, etc.) would be huge and could exhaust the available memory in a worker.\n",
    " \n",
    "#### Use `groupByKey()` to generate a pair RDD of type `('day', iterator)`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "monday: [1]\n",
      "tuesday: [1, 1]\n",
      "friday: [1]\n",
      "sunday: [1]\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "# Note that groupByKey requires no parameters\n",
    "daysGrouped = dayPairs.groupByKey()\n",
    "for key, value in daysGrouped.collect():\n",
    "    print('{0}: {1}'.format(key, list(value)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (2b) Use `groupByKey()` to obtain the counts **\n",
    "#### Using the `groupByKey()` transformation creates an RDD containing 3 elements, each of which is a pair of a day and a Python iterator.\n",
    "#### Now sum the iterator using a `map()` transformation.  The result should be a pair RDD consisting of (day, count) pairs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('tuesday', 2), ('monday', 1), ('friday', 1), ('sunday', 1)]\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "dayCountsGrouped = daysGrouped.map(lambda x: (x[0], len(x[1])))\n",
    "print(dayCountsGrouped.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (2c) Counting using `reduceByKey` **\n",
    "#### A better approach is to start from the pair RDD and then use the [reduceByKey()](http://spark.apache.org/docs/latest/api/python/pyspark.html#pyspark.RDD.reduceByKey) transformation to create a new pair RDD. The `reduceByKey()` transformation gathers together pairs that have the same key and applies the function provided to two values at a time, iteratively reducing all of the values to a single value. `reduceByKey()` operates by applying the function first within each partition on a per-key basis and then across the partitions, allowing it to scale efficiently to large datasets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('tuesday', 2), ('friday', 1), ('monday', 1), ('sunday', 1)]\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "# Note that reduceByKey takes in a function that accepts two values and returns a single value\n",
    "def sum(x, y):\n",
    "    return x + y\n",
    "\n",
    "dayCounts = dayPairs.reduceByKey(lambda x,y : sum(x, y))  #sum(x, y) is optional, we can sum x + y there itself instead of \n",
    "                                                          #calling a function, I just wanted to check function call in lambda LOL\n",
    "print(dayCounts.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (2d) All together **\n",
    "#### The expert version of the code performs the `map()` to pair RDD, `reduceByKey()` transformation, and `collect` in one statement."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('tuesday', 2), ('friday', 1), ('monday', 1), ('sunday', 1)]\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "dayCountsCollected = (daysRDD.map(lambda x: (x, 1)).reduceByKey(lambda a, b: a + b).collect())\n",
    "print(dayCountsCollected)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ** Part 3: Finding unique days and a mean value **"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (3a) Unique words **\n",
    "#### Calculate the number of unique days in `daysRDD`.  You can use other RDDs that you have already created to make this easier."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "uniqueDays = dayPairs.distinct().count()\n",
    "print(uniqueDays)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (3b) Mean using `reduce` **\n",
    "#### Find the mean number of days per unique day in `dayCounts`.\n",
    "#### Use a `reduce()` action to sum the counts in `dayCounts` and then divide by the number of unique days.  First `map()` the pair RDD `dayCounts`, which consists of (key, value) pairs, to an RDD of values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "5\n",
      "1.25\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "from operator import add\n",
    "totalCount = (dayCounts.map(lambda x: x[1]).reduce(lambda a, b: a + b))\n",
    "average = totalCount / float(uniqueDays)\n",
    "print(totalCount)\n",
    "print(round(average, 2))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### ** Part 4: Compute Deerfoot Trail stats **"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### In this section we will apply some of the above concepts towards analyzing commute time and accidents data collected for Deerfoot Trail."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (4a) Loading the data **\n",
    "#### We will first load the data.  The data was collected in the period September 2013 to April 2014.  It was obtained by querying Google Maps for commute times and Twitter for accident reports.  Although this data set is very small, because we are using parallel computation via Spark the functions we develop will scale for larger data sets.  To convert a text file into an RDD, we use the `SparkContext.textFile()` method. We will use `take(15)` to print 15 lines from this file."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "0: 21/09/2013,Saturday,34,34,34,34,35,34,35,36,38,36,36,35,35,35,35,35,36,34,34,0,0,0,2,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,2,0,2\n",
      "1: 22/09/2013,Sunday,34,34,34,34,34,34,34,35,35,35,34,35,34,35,34,34,34,34,34,0,0,0,1,0,0,0,1,1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,3\n",
      "2: 23/09/2013,Monday,35,36,41,43,45,41,36,35,35,35,37,40,43,46,43,37,34,34,35,0,0,0,0,0,0,0,0,0,1,0,0,0,0,0,0,0,1,0,0,0,1,0,0,0,2\n",
      "3: 24/09/2013,Tuesday,35,36,40,44,52,41,38,36,36,36,37,40,44,47,42,39,34,35,35,0,0,0,1,1,0,0,0,0,0,0,1,1,0,1,0,0,0,0,0,0,4,1,1,0,5\n",
      "4: 25/09/2013,Wednesday,35,36,40,39,39,37,36,35,36,37,37,40,44,45,41,38,35,35,35,0,0,0,0,0,0,0,0,0,0,0,0,2,2,1,0,0,0,0,0,0,1,4,0,0,5\n",
      "5: 26/09/2013,Thursday,34,36,50,56,49,37,37,35,36,36,39,56,59,46,42,38,35,34,35,0,0,1,1,1,0,1,0,0,0,0,2,1,5,1,0,0,0,0,0,3,4,5,1,0,13\n",
      "6: 27/09/2013,Friday,34,35,37,37,36,35,36,36,36,38,40,43,47,48,42,38,35,35,35,0,0,0,2,0,0,0,0,0,1,2,0,0,0,0,0,1,0,0,0,0,1,2,0,0,6\n",
      "7: 28/09/2013,Saturday,34,34,34,34,34,34,35,35,35,35,35,35,35,49,44,36,34,34,35,0,0,0,0,0,0,0,0,0,0,0,0,0,1,0,0,0,0,0,0,0,1,1,0,0,1\n",
      "8: 29/09/2013,Sunday,34,34,34,34,34,34,34,35,35,35,35,35,35,34,34,34,34,34,35,0,1,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,1\n",
      "9: 30/09/2013,Monday,34,35,40,41,47,40,43,35,35,36,37,39,42,44,44,43,35,34,35,0,0,1,0,0,0,2,0,0,0,0,0,1,0,3,0,0,0,0,0,1,1,1,1,0,7\n",
      "10: 01/10/2013,Tuesday,35,36,40,41,40,36,36,35,36,36,37,41,44,47,41,37,35,35,35,0,0,0,0,0,1,1,0,1,0,0,0,0,0,0,0,0,0,0,0,0,1,1,0,0,3\n",
      "11: 02/10/2013,Wednesday,35,36,41,44,45,45,40,36,36,36,37,40,44,46,43,38,35,35,35,0,0,1,3,0,0,0,0,0,0,0,0,0,2,0,0,0,0,0,0,1,0,1,0,0,6\n",
      "12: 03/10/2013,Thursday,35,36,39,39,37,36,36,36,35,36,39,40,43,45,43,38,35,35,35,0,0,0,0,0,0,1,1,0,0,0,0,0,0,2,4,0,0,0,0,0,4,3,1,0,8\n",
      "13: 04/10/2013,Friday,35,35,36,36,36,36,36,36,36,39,43,44,45,45,43,39,35,35,35,0,0,0,0,0,0,0,0,0,1,0,0,0,1,2,0,0,0,0,0,0,1,1,0,0,4\n",
      "14: 05/10/2013,Saturday,34,34,34,34,34,35,35,35,35,35,35,35,35,35,35,35,34,35,36,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0,0\n"
     ]
    }
   ],
   "source": [
    "# Just run this code\n",
    "import os.path\n",
    "baseDir = os.path.join('data')\n",
    "inputPath = os.path.join('SENG501', 'lab1', 'deerfoot.csv')\n",
    "fileName = 'deerfoot.csv'\n",
    "\n",
    "deerfootRDD = (sc.textFile(fileName, 8))\n",
    "#print('\\n'.join(deerfootRDD.zipWithIndex().map(lambda (l, num): '{0}: {1}'.format(num, l)).take(15)))\n",
    "print('\\n'.join(deerfootRDD.zipWithIndex().map(lambda x: '{0}: {1}'.format(x[1], x[0])).take(15)))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (4b) Extracting fields relevant to the analysis **\n",
    "#### We will extract only those fields that will be useful for our further analysis in this lab.  Specifically, we are interested in field 2 (day), field 7 (commute time at 8 AM), and field 14 (commute time at 4 PM).  We consider only these 2 times since these best represent the morning and afternoon rush traffic.  Write a function `extractFields` that takes as input each record of `deerfootRDD` and produces a record for another RDD that only contains these 3 fields."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('Saturday', '35', '35')\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "def extractFields(deerfootRDDRecord):\n",
    "    \"\"\"Creates a record consisting of day, 8 AM commute time, and 4 PM commute time.\n",
    "\n",
    "    Args:\n",
    "        deerfootRDDRecord : a comma separated string consisting of all fields in the data set.\n",
    "\n",
    "    Returns:\n",
    "        extracted record: a comma separated record (day, 8 AM commute time, 4 PM commute time)\n",
    "    \"\"\"\n",
    "    record = deerfootRDDRecord.split(\",\")\n",
    "    return record[1], record[6], record[13]\n",
    "   \n",
    "print(extractFields(deerfootRDD.take(1)[0]))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (4c) Obtaining extracted RDD **\n",
    "#### Transform the `deerfootRDD` so that we get a resulting `deerfootPeakRDD` that only has peak hour commute times."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Saturday', '35', '35')]\n",
      "\n",
      "---------------Printing deerfootPeakRDD----------------\n",
      "\n",
      "[('Saturday', '35', '35'), ('Sunday', '34', '35'), ('Monday', '45', '40'), ('Tuesday', '52', '40'), ('Wednesday', '39', '40'), ('Thursday', '49', '56'), ('Friday', '36', '43'), ('Saturday', '34', '35'), ('Sunday', '34', '35'), ('Monday', '47', '39'), ('Tuesday', '40', '41'), ('Wednesday', '45', '40'), ('Thursday', '37', '40'), ('Friday', '36', '44'), ('Saturday', '34', '35'), ('Sunday', '34', '35'), ('Monday', '45', '40'), ('Tuesday', '40', '46'), ('Wednesday', '39', '41'), ('Thursday', '40', '44'), ('Friday', '36', '44'), ('Saturday', '35', '35'), ('Sunday', '34', '35'), ('Monday', '35', '35'), ('Tuesday', '42', '41'), ('Wednesday', '44', '41'), ('Thursday', '40', '43'), ('Friday', '40', '49'), ('Saturday', '35', '36'), ('Sunday', '34', '36'), ('Monday', '40', '40'), ('Tuesday', '46', '41'), ('Wednesday', '44', '41'), ('Thursday', '41', '43'), ('Friday', '38', '44'), ('Saturday', '35', '35'), ('Sunday', '34', '35'), ('Monday', '40', '39'), ('Tuesday', '45', '41'), ('Wednesday', '40', '43'), ('Thursday', '42', '46'), ('Friday', '37', '47'), ('Saturday', '35', '36'), ('Sunday', '36', '35'), ('Monday', '41', '39'), ('Tuesday', '41', '41'), ('Wednesday', '39', '42'), ('Thursday', '39', '43'), ('Friday', '36', '44'), ('Saturday', '34', '35'), ('Sunday', '36', '35'), ('Monday', '37', '39'), ('Tuesday', '46', '43'), ('Wednesday', '40', '43'), ('Thursday', '40', '44'), ('Friday', '39', '45'), ('Saturday', '38', '35'), ('Sunday', '35', '35'), ('Monday', '42', '42'), ('Tuesday', '39', '46'), ('Wednesday', '52', '43'), ('Thursday', '46', '46'), ('Friday', '37', '45'), ('Saturday', '35', '35'), ('Sunday', '35', '35'), ('Monday', '35', '39'), ('Tuesday', '48', '40'), ('Wednesday', '38', '41'), ('Thursday', '39', '44'), ('Friday', '38', '43'), ('Saturday', '34', '35'), ('Sunday', '34', '34'), ('Monday', '47', '54'), ('Tuesday', '45', '43'), ('Wednesday', '51', '43'), ('Thursday', '49', '43'), ('Friday', '40', '46'), ('Saturday', '35', '36'), ('Sunday', '37', '36'), ('Monday', '52', '46'), ('Tuesday', '60', '79'), ('Wednesday', '55', '44'), ('Thursday', '41', '42'), ('Friday', '39', '52'), ('Saturday', '35', '35'), ('Sunday', '35', '35'), ('Monday', '49', '44'), ('Tuesday', '44', '43'), ('Wednesday', '60', '64'), ('Thursday', '57', '48'), ('Friday', '41', '49'), ('Saturday', '35', '36'), ('Sunday', '36', '36'), ('Monday', '38', '46'), ('Tuesday', '35', '35'), ('Wednesday', '35', '35'), ('Thursday', '35', '35'), ('Friday', '35', '37'), ('Saturday', '37', '35'), ('Sunday', '35', '35'), ('Monday', '37', '40'), ('Tuesday', '37', '35'), ('Wednesday', '38', '35'), ('Thursday', '37', '42'), ('Friday', '57', '40'), ('Saturday', '37', '36'), ('Sunday', '37', '36'), ('Monday', '44', '43'), ('Tuesday', '39', '40'), ('Wednesday', '49', '40'), ('Thursday', '38', '41'), ('Friday', '49', '66'), ('Saturday', '35', '35'), ('Sunday', '35', '35'), ('Monday', '43', '45'), ('Tuesday', '41', '41'), ('Wednesday', '40', '43'), ('Thursday', '38', '41'), ('Friday', '37', '44'), ('Saturday', '35', '35'), ('Sunday', '35', '35'), ('Monday', '41', '37'), ('Tuesday', '40', '41'), ('Wednesday', '61', '39'), ('Thursday', '38', '41'), ('Friday', '38', '44'), ('Saturday', '35', '36'), ('Sunday', '35', '35'), ('Monday', '38', '38'), ('Tuesday', '38', '38'), ('Wednesday', '40', '45'), ('Thursday', '42', '37'), ('Friday', '35', '38'), ('Saturday', '34', '34'), ('Sunday', '34', '34'), ('Monday', '56', '36'), ('Tuesday', '48', '35'), ('Wednesday', '57', '39'), ('Thursday', '38', '37'), ('Friday', '36', '35'), ('Saturday', '35', '34'), ('Sunday', '34', '34'), ('Monday', '40', '57'), ('Tuesday', '38', '38'), ('Wednesday', '42', '36'), ('Thursday', '41', '39'), ('Friday', '40', '39'), ('Saturday', '34', '34'), ('Sunday', '34', '34'), ('Monday', '34', '34'), ('Tuesday', '39', '37'), ('Wednesday', '38', '37'), ('Thursday', '37', '38'), ('Friday', '35', '36'), ('Saturday', '34', '34'), ('Sunday', '34', '34'), ('Monday', '64', '36'), ('Tuesday', '39', '37'), ('Wednesday', '43', '39'), ('Thursday', '38', '36'), ('Friday', '38', '36'), ('Saturday', '36', '36'), ('Sunday', '35', '35'), ('Monday', '46', '37'), ('Tuesday', '44', '38'), ('Wednesday', '40', '37'), ('Thursday', '40', '40'), ('Friday', '36', '40'), ('Saturday', '35', '35'), ('Sunday', '35', '35'), ('Monday', '38', '39'), ('Tuesday', '43', '40'), ('Wednesday', '42', '41'), ('Thursday', '41', '40'), ('Friday', '38', '39'), ('Saturday', '35', '35'), ('Sunday', '35', '35'), ('Monday', '40', '37'), ('Tuesday', '59', '38'), ('Wednesday', '42', '39'), ('Thursday', '46', '36'), ('Friday', '43', '37'), ('Saturday', '34', '34'), ('Sunday', '35', '33'), ('Monday', '37', '36'), ('Tuesday', '35', '50'), ('Wednesday', '54', '36'), ('Thursday', '43', '35'), ('Friday', '35', '37'), ('Saturday', '34', '34'), ('Sunday', '35', '34'), ('Monday', '40', '36'), ('Tuesday', '87', '35'), ('Wednesday', '35', '35'), ('Thursday', '40', '36'), ('Friday', '35', '41'), ('Saturday', '33', '34'), ('Sunday', '34', '33'), ('Monday', '40', '36'), ('Tuesday', '40', '39'), ('Wednesday', '40', '45'), ('Thursday', '40', '44')]\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "deerfootPeakRDD = deerfootRDD.map(extractFields)\n",
    "\n",
    "print(deerfootPeakRDD.take(1))\n",
    "\n",
    "print('\\n---------------Printing deerfootPeakRDD----------------\\n')\n",
    "print(deerfootPeakRDD.collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (4d) Obtaining stats - counting number of occurrences of each day of the week **\n",
    "#### Start with the `deerfootPeakRDD`.  Create a pair RDD `deerfootDayPairRDD` that contains records where day is the key and 1 is the value. Apply another transformation on `deerfootDayPairRDD` to get a `deerfootDayCounts` RDD"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Thursday', 29), ('Friday', 28), ('Monday', 29), ('Wednesday', 29), ('Saturday', 29), ('Sunday', 29), ('Tuesday', 29)]\n",
      "\n",
      " getting day count of friday: 28\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "deerfootDayPairRDD = deerfootPeakRDD.map(lambda x: (x[0], 1))\n",
    "#print(deerfootDayPairRDD.collect())\n",
    "deerfootDayCounts = deerfootDayPairRDD.reduceByKey(lambda a, b: a + b)\n",
    "#print(deerfootDayCounts.collect())\n",
    "deerfootDayCountsList = deerfootDayCounts.collect()\n",
    "print(deerfootDayCountsList)\n",
    "deerfootDayCountsDict = dict(deerfootDayCountsList)\n",
    "print(\"\\n getting day count of friday:\", deerfootDayCountsDict.get('Friday'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (4e) Filtering out Saturdays and Sundays **\n",
    "#### As we can see from the previous result, there is almost an equal number of days of each type in the data set, which suggests that there is no big gap in the data collection.  Let's say we are now only interested in commute time stats for Monday to Friday.  Write a function called `filterSatSun` that filters out records for Saturdays and Sundays in `deerfootPeakRDD`.  Apply this transformation on `deerfootPeakRDD` to obtain an RDD called `deerfootPeakMFRDD`. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Monday', '45', '40'), ('Tuesday', '52', '40'), ('Wednesday', '39', '40'), ('Thursday', '49', '56'), ('Friday', '36', '43')]\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "\n",
    "def checkDay(record):\n",
    "    if 'Saturday' in record or 'Sunday' in record:\n",
    "        return False\n",
    "    else:\n",
    "        return True\n",
    "\n",
    "def filterSatSun(deerfootPeakRDDRecord):\n",
    "    \"\"\"Ignores \"Saturday\" and \"Sunday\" records.\n",
    "\n",
    "    Args:\n",
    "        deerfootPeakRDDRecord: A comma separated string (day, 8 AM commute time, 4 PM commute time).\n",
    "\n",
    "    Returns:\n",
    "        false if day is \"Saturday\" or \"Sunday\". true if otherwise\n",
    "    \"\"\"\n",
    "    return checkDay(deerfootPeakRDDRecord)\n",
    "\n",
    "deerfootPeakMFRDD = deerfootPeakRDD.filter(filterSatSun)\n",
    "print(deerfootPeakMFRDD.take(5))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (4f) Computing average commute times for each day of the week **\n",
    "#### We will now compute the average of commute times for each day of the week for both 8 AM and 4 PM. To do this, first create a pair RDD called `deerfootPeakAMRDD` where each record has day as the key and 8 AM commute time as value.  Apply one or more appropriate transformations to compute average.  Repeat the process for the evening rush hour.  You can use the previously computed `deerfootDayCountsDict' in the average calculation. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[('Thursday', 41.10344827586207), ('Monday', 42.44827586206897), ('Friday', 38.57142857142857), ('Wednesday', 44.206896551724135), ('Tuesday', 44.48275862068966)]\n",
      "[('Thursday', 41.37931034482759), ('Monday', 40.310344827586206), ('Friday', 43.0), ('Wednesday', 40.93103448275862), ('Tuesday', 41.44827586206897)]\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "deerfootPeakAMRDD = deerfootPeakMFRDD.map(lambda x: (x[0], int(x[1])))\n",
    "deerfootPeakAMreduceByDay = deerfootPeakAMRDD.reduceByKey(lambda x, y: x + y)\n",
    "\n",
    "amAverages = list()\n",
    "\n",
    "deerfootPeakAMreduceByDay_dict = dict(deerfootPeakAMreduceByDay.collect())\n",
    "\n",
    "for key, item in deerfootPeakAMreduceByDay_dict.items():\n",
    "    avg = item/float(deerfootDayCountsDict.get(key))     #total time on friday divided by total number of fridays...eg. 1080/28\n",
    "    amAverages.append((key,avg))\n",
    "\n",
    "deerfootPeakPMRDD = deerfootPeakMFRDD.map(lambda x: (x[0], int(x[2])))\n",
    "deerfootPeakPMreduceByDay = deerfootPeakPMRDD.reduceByKey(lambda x, y: x + y)\n",
    "\n",
    "pmAverages = list()\n",
    "\n",
    "deerfootPeakPMreduceByDay_dict = dict(deerfootPeakPMreduceByDay.collect())\n",
    "\n",
    "for key, item in deerfootPeakPMreduceByDay_dict.items():\n",
    "    avg = item/float(deerfootDayCountsDict.get(key))\n",
    "    pmAverages.append((key, avg))\n",
    "\n",
    "print(amAverages)\n",
    "print(pmAverages)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "#### ** (4g) Computing max morning hour rush commute times for each day of the week **\n",
    "#### For 8 AM, find the maximum commute time for each day of the week. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "('Thursday', 57)\n",
      "('Monday', 64)\n",
      "('Friday', 57)\n",
      "('Wednesday', 61)\n",
      "('Tuesday', 87)\n"
     ]
    }
   ],
   "source": [
    "# TODO: Replace <FILL IN> with appropriate code\n",
    "\n",
    "deerfootPeakAMMaxreduceByDay = deerfootPeakAMRDD.groupByKey().map(lambda x : (x[0], list(x[1]))).map(lambda x: (x[0], max(x[1])))\n",
    "\n",
    "#deerfootPeakAMRDD.groupByKey().map(lambda x : (x[0], list(x[1]))).map(lambda x: (x[0], max(x[1])))\n",
    "#print(deerfootPeakAMMaxreduceByDay.collect())\n",
    "for item in deerfootPeakAMMaxreduceByDay.collect():\n",
    "    print(item)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [],
   "source": [
    "#stop spark context and spark job\n",
    "sc.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO:sparkhpc.sparkjob:\n"
     ]
    }
   ],
   "source": [
    "sj.stop()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}