NYPD_Motor_Vehicle_Accidents_Scala.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "# Draw insights from car accident reports\n",
    "\n",
    "(C) Copyright IBM Corp. 2016\n",
    "\n",
    "This notebook shows you how to analyze car vehicle accidents based on accident reports for New York. The analysis steps in the notebook show how you can use the information about accidents to learn more about the possible causes for collisions. You will learn how to install additional Scala jars and how to perform descriptive data analysis.\n",
    "\n",
    "## Table of contents\n",
    "- [Get data](#get_data)\n",
    "- [Access data](#access_data)\n",
    "- [Load data](#load_data)\n",
    "- [Load visualization packages](#load_visualization_packages)\n",
    "- [Explore data](#explore_data)\n",
    "- [Clean and shape the data](#data_cleaning)\n",
    "- [Summary](#summary)\n",
    "\n",
    "    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"get_data\"></a>\n",
    "## Get data\n",
    "\n",
    "Begin by getting the data about car accidents in the New York area. Click [NYPD Motor Vehicle Collisions](https://data.cityofnewyork.us/Public-Safety/NYPD-Motor-Vehicle-Collisions/h9gi-nx95) to access the data set. Then click **Export** to download the data as a CSV file. \n",
    "\n",
    "This data set covers all reported vehicle collisions in New York starting in July 2012 until today  and contains detailed information about the incidents.\n",
    "\n",
    "After you download the file, load the data file to the notebook by clicking **Palette>Data Sources**. Click **Add Source**, select **From file**, and browse to the data file. \n",
    "\n",
    "**Note**: Because the CSV file is relatively large, it may take a few minutes until the data file is loaded. The process bar indicates the load status of the file. \n",
    "\n",
    "This file is stored in the Object Storage instance that is associated with your IBM Analytics for Apache Spark service."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"access_data\"></a>\n",
    "## Access data\n",
    "\n",
    "Before you can access data in the data file in the Object Storage instance, you must set the Hadoop configuration with the Object Storage instance service credentials by using the following configuration function:\n",
    "\n",
    "Note: You will not be using Hadoop in this sample; however Spark leverages some Hadoop components."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "/*found at http://stackoverflow.com/questions/33725500/load-data-from-bluemix-object-store-in-spark*/\n",
    "import scala.collection.breakOut\n",
    "\n",
    "def setConfig(name:String, dsConfiguration:String) : Unit = {\n",
    "    val pfx = \"fs.swift.service.\" + name\n",
    "    val settings:Map[String,String] = dsConfiguration.split(\"\\\\n\").\n",
    "        map(l=>(l.split(\":\",2)(0).trim(), l.split(\":\",2)(1).trim()))(breakOut)\n",
    "\n",
    "    val conf = sc.getConf\n",
    "    conf.set(pfx + \"auth.url\", settings.getOrElse(\"auth_url\",\"\"))\n",
    "    conf.set(pfx + \"tenant\", settings.getOrElse(\"tenantId\", \"\"))\n",
    "    conf.set(pfx + \"username\", settings.getOrElse(\"username\", \"\"))\n",
    "    conf.set(pfx + \"password\", settings.getOrElse(\"password\", \"\"))\n",
    "    conf.set(pfx + \"apikey\", settings.getOrElse(\"password\", \"\"))\n",
    "    conf.set(pfx + \"auth.endpoint.prefix\", \"endpoints\")\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Click the next code cell to set the focus on the cell. Now add the credentials to access the data  file to this code cell by selecting **Palette>Data Sources** and clicking the `Insert to code` function below the data file in the **Data Source** pane.  \n",
    "\n",
    "When you select the `Insert to code` function, a code cell with a Scala Map is created for you. Adjust the credentials in the dictionary to correspond with the credentials inserted by the `Insert to code` function and run the dictionary code cell. The access credentials to the Object Storage instance in the dictionary are provided for later usage. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "setConfig(\"spark\", credentials_1.toString())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"load_data\"></a> \n",
    "## Load data\n",
    "\n",
    "Instead of specifying the schema for a Spark DataFrame programatically, you can use the `spark-csv` module. \n",
    "\n",
    "Load the NYPD motor vehicle collisions data set from Object Storage into an RDD and then convert the RDD into a Spark DataFrame:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "created df\n"
     ]
    }
   ],
   "source": [
    "/*found at https://github.com/databricks/spark-csv*/\n",
    "import org.apache.spark.sql.SQLContext\n",
    "\n",
    "val sqlctx = new SQLContext(sc);\n",
    "import sqlctx.implicits._\n",
    "sqlctx.setConf(\"spark.sql.shuffle.partitions\", \"10\");\n",
    "\n",
    "/* Parenthesis allows line continuation which helps with readability */\n",
    "val df = (sqlctx.read\n",
    "    .format(\"com.databricks.spark.csv\")\n",
    "    .option(\"delimiter\",\",\")\n",
    "    .option(\"header\",\"true\")\n",
    "    .option(\"inferschema\",\"true\")\n",
    "    .option(\"mode\",\"DROPMALFORMED\")\n",
    "    .option(\"treatEmptyValuesAsNulls\", \"true\")\n",
    "    .load(\"swift://notebooks.spark/NYPD_Motor_Vehicle_Collisions.csv\")\n",
    ");\n",
    "\n",
    "println(\"created df\")\n",
    "println(\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now that the data is loaded, check the inferred schema:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- DATE: string (nullable = true)\n",
      " |-- TIME: string (nullable = true)\n",
      " |-- BOROUGH: string (nullable = true)\n",
      " |-- ZIP CODE: integer (nullable = true)\n",
      " |-- LATITUDE: double (nullable = true)\n",
      " |-- LONGITUDE: double (nullable = true)\n",
      " |-- LOCATION: string (nullable = true)\n",
      " |-- ON STREET NAME: string (nullable = true)\n",
      " |-- CROSS STREET NAME: string (nullable = true)\n",
      " |-- OFF STREET NAME: string (nullable = true)\n",
      " |-- NUMBER OF PERSONS INJURED: string (nullable = true)\n",
      " |-- NUMBER OF PERSONS KILLED: integer (nullable = true)\n",
      " |-- NUMBER OF PEDESTRIANS INJURED: integer (nullable = true)\n",
      " |-- NUMBER OF PEDESTRIANS KILLED: integer (nullable = true)\n",
      " |-- NUMBER OF CYCLIST INJURED: integer (nullable = true)\n",
      " |-- NUMBER OF CYCLIST KILLED: string (nullable = true)\n",
      " |-- NUMBER OF MOTORIST INJURED: string (nullable = true)\n",
      " |-- NUMBER OF MOTORIST KILLED: integer (nullable = true)\n",
      " |-- CONTRIBUTING FACTOR VEHICLE 1: string (nullable = true)\n",
      " |-- CONTRIBUTING FACTOR VEHICLE 2: string (nullable = true)\n",
      " |-- CONTRIBUTING FACTOR VEHICLE 3: string (nullable = true)\n",
      " |-- CONTRIBUTING FACTOR VEHICLE 4: string (nullable = true)\n",
      " |-- CONTRIBUTING FACTOR VEHICLE 5: string (nullable = true)\n",
      " |-- UNIQUE KEY: integer (nullable = true)\n",
      " |-- VEHICLE TYPE CODE 1: string (nullable = true)\n",
      " |-- VEHICLE TYPE CODE 2: string (nullable = true)\n",
      " |-- VEHICLE TYPE CODE 3: string (nullable = true)\n",
      " |-- VEHICLE TYPE CODE 4: string (nullable = true)\n",
      " |-- VEHICLE TYPE CODE 5: string (nullable = true)\n",
      "\n",
      "()\n"
     ]
    }
   ],
   "source": [
    "println(df.printSchema())\n",
    "println(\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Run the next code cell to look at the data itself:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array([03/07/2016,10:40,BROOKLYN,11208,40.6801678,-73.8765734,(40.6801678, -73.8765734),ATLANTIC AVENUE,FOUNTAIN AVENUE,null,1,0,0,0,0,0,1,0,Fatigued/Drowsy,Unspecified,Unspecified,null,null,3401775,TAXI,PASSENGER VEHICLE,TAXI,null,null])"
      ]
     },
     "execution_count": 33,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Run the following cell to count the number of vehicle collisions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "769053\n"
     ]
    }
   ],
   "source": [
    "println(df.count())\n",
    "println(\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"load_visualization_packages\"></a> \n",
    "## Load visualization packages\n",
    "\n",
    "This notebook makes use of the [Brunel](https://github.com/Brunel-Visualization/Brunel) visualizaton package. `Brunel` is \"a highly succinct and novel language that defines interactive data visualizations based on tabular data.\" IBM Bluemix adds this library by default, but if you are not on IBM Bluemix, you will need to uncomment the line below in order to add the Brunel JarFile."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "//%AddJar -magic https://brunelvis.org/jar/spark-kernel-brunel-all-1.2.jar"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"explore_data\"></a> \n",
    "## Explore data\n",
    "\n",
    "Now that the data is loaded, you can start exploring it and visualizing patterns by using scatter plots. \n",
    "\n",
    "In the next code cell, select the columns with data that you want to explore, for example, NUMBER OF PERSONS INJURED or CONTRIBUTING FACTOR VEHICLE, and transform these columns into a DataFrame."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val trimmedDF = df.filter(df(\"LATITUDE\") !== 0).select(\"LATITUDE\", \"LONGITUDE\", \"DATE\", \"TIME\", \"BOROUGH\", \"ON STREET NAME\", \"CROSS STREET NAME\", \n",
    "                                                        \"NUMBER OF PERSONS INJURED\", \"NUMBER OF PERSONS KILLED\", \"CONTRIBUTING FACTOR VEHICLE 1\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "//found at http://stackoverflow.com/questions/35592917/renaming-column-names-of-a-data-frame-in-spark-scala\n",
    "val columnNames = Seq(\"Latitude\", \"Longitude\", \"Date\", \"Time\", \"Borough\", \"On Street\", \n",
    "                      \"Cross Street\", \"Persons Injured\", \"Persons Killed\", \"Contributing Factor\")\n",
    "\n",
    "var collisionsDF = trimmedDF.toDF(columnNames: _*)\n",
    "\n",
    "//this is being done to take a subset because Brunel can not handle this large of a data set\n",
    "collisionsDF = collisionsDF.limit(25000)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create an explorative scatter plot of the data\n",
    "\n",
    "Using an explorative scatter plot is a way to analyze certain characteristics of the data set. \n",
    "\n",
    "Create an intial explorative scatter plot of all collisions by using the latitude and longitude information in the raw data:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "collapsed": true
   },
   "outputs": [],
   "source": [
    "//creating a new dataframe is not required, but I am trying to minimize the amount of data passed to Brunel\n",
    "val allCollisions = collisionsDF.select(\"Latitude\", \"Longitude\", \"Borough\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "data('allCollisions') x(Longitude) y(Latitude) style('size:20%') interaction(none)\n",
    "          ::\n",
    "width=1200, height=600"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Although this scatter plot is not a street map of New York City, you will notice that the scatter plot dots roughly correspond to the street map of New York City. We can see that very few collisions happen in Central Park and on bridges. We can see higher density of collisions in areas such street crossings and curves."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Enhance the scatter plot with information about city boroughs\n",
    "\n",
    "Now add information about the city boroughs and use a different color to depict each borough on the scatter plot:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "//remove collisions with null values for Borough\n",
    "val collisionsByBorough = allCollisions.na.drop(\"any\", Seq(\"Borough\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "data('collisionsByBorough') x(Longitude) y(Latitude) color(Borough) style('size:20%') interaction(none) \n",
    "title('Motor Vehicle Collisions in New York City by Borough') axes(x:'Longitude',y:'Latitude')\n",
    "          ::\n",
    "width=1200, height=600"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create a bar graph to visualize patterns\n",
    "\n",
    "Create a bar graph to show the total number of collisions by borough:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "//replace null values with 'None' because Brunel will not display null values\n",
    "val boroughBarDF = collisionsDF.groupBy(\"Borough\").count().sort(\"count\").na.fill(\"NONE\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "data('boroughBarDF') transpose bar x(Borough) y(count) color(Borough) legends(none) sort(count:ascending) interaction(none) \n",
    "          ::\n",
    "width=600, height=300"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The bar graph clearly shows that the most collisions happen in Brooklyn and the least on Staten Island."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Adjust the scatter plot settings to use color codes to indicate collisions resulting in car body damage, personal injury, and fatal accidents:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "import org.apache.spark.sql._\n",
    "import org.apache.spark.sql.types._\n",
    "\n",
    "var severityRDD = collisionsDF.select(\"Latitude\", \"Longitude\", \"Persons Injured\", \"Persons Killed\").rdd.map(row => \n",
    "    if (row(3) != 0) {\n",
    "        Row.fromSeq(row.toSeq :+ \"fatal accidents\")\n",
    "    } else if (row(2) != \"0\" && row(3) == 0){\n",
    "        //may need to change the type of row(2) in the future\n",
    "        Row.fromSeq(row.toSeq :+ \"personal injury\")\n",
    "    } else {\n",
    "        Row.fromSeq(row.toSeq :+ \"car body damage\")\n",
    "    })\n",
    "\n",
    "var struct =\n",
    "    StructType(\n",
    "    StructField(\"Latitude\", DoubleType, true) ::\n",
    "    StructField(\"Longitude\", DoubleType, true) ::\n",
    "    StructField(\"Persons Injured\", StringType, true) ::\n",
    "    StructField(\"Persons Killed\", IntegerType, true) ::\n",
    "    StructField(\"Severity\", StringType, true) :: Nil)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val severityDF = sqlctx.createDataFrame(severityRDD, struct)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false,
    "scrolled": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "data('severityDF') x(Longitude) y(Latitude) color(Severity) size(Severity:[30%,80%,40%]) interaction(none)\n",
    "          ::\n",
    "width=1200, height=600"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The resulting scatter plot shows that there are fatal accident hot spots throughout the city. You can see that in some areas car body damage is prevalent, while in other areas personal injuries happen more often."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<a id=\"data_cleaning\"></a> \n",
    "## Clean and shape the data\n",
    "\n",
    "After using scatter plots to analyze certain characteristics of the raw data set, you will now learn how to clean and shape the data set to enable more plotting and further analysis. \n",
    "\n",
    "Begin by looking at the column names again to better assess which information you can use:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array(DATE, TIME, BOROUGH, ZIP CODE, LATITUDE, LONGITUDE, LOCATION, ON STREET NAME, CROSS STREET NAME, OFF STREET NAME, NUMBER OF PERSONS INJURED, NUMBER OF PERSONS KILLED, NUMBER OF PEDESTRIANS INJURED, NUMBER OF PEDESTRIANS KILLED, NUMBER OF CYCLIST INJURED, NUMBER OF CYCLIST KILLED, NUMBER OF MOTORIST INJURED, NUMBER OF MOTORIST KILLED, CONTRIBUTING FACTOR VEHICLE 1, CONTRIBUTING FACTOR VEHICLE 2, CONTRIBUTING FACTOR VEHICLE 3, CONTRIBUTING FACTOR VEHICLE 4, CONTRIBUTING FACTOR VEHICLE 5, UNIQUE KEY, VEHICLE TYPE CODE 1, VEHICLE TYPE CODE 2, VEHICLE TYPE CODE 3, VEHICLE TYPE CODE 4, VEHICLE TYPE CODE 5)"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "/*\n",
    "Many of the following charts have less data points than the scatter plots above. \n",
    "For this reason, I went back to using the entire dataframe instead of the 25000\n",
    "subset used above.\n",
    "*/\n",
    "df.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Run the next cell to obtain the number of entries with valid information about the street and borough:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "577477"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "val collisions = df.na.drop(\"any\", Seq(\"ON STREET NAME\", \"BOROUGH\"))\n",
    "collisions.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "### Spatial and temporal normalization by using Spark\n",
    "\n",
    "To obtain a consistent representation of the spatial and temporal information about collisions, you have to normalize the data. Normalization is the process of organizing the columns (attributes) and tables (relations) to minimize data redundancy. This step will help you in future analyses."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import org.apache.spark.sql.Row\n",
    "\n",
    "var collisionsOutArray = Array(\"Time\", \"Street\", \"Borough\")\n",
    "for(col <- df.columns){\n",
    "    if (!(Array(\"ON STREET NAME\", \"OFF STREET NAME\", \"CROSS STREET NAME\", \"BOROUGH\", \"DATE\", \"TIME\") contains col)) {\n",
    "        collisionsOutArray = collisionsOutArray :+ col\n",
    "    }\n",
    "}\n",
    "\n",
    "val normalizationCode = scala.collection.mutable.HashMap[String, String](\n",
    "    \"avenue\" -> \"av\",\n",
    "    \"ave\" -> \"av\",\n",
    "    \"avnue\" -> \"av\",\n",
    "    \"street\" -> \"st\",\n",
    "    \"road\" -> \"rd\",\n",
    "    \"boulevard\" -> \"blvd\",\n",
    "    \"place\" -> \"pl\",\n",
    "    \"plaza\" -> \"pl\",\n",
    "    \"square\" -> \"sq\",\n",
    "    \"drive\" -> \"dr\",\n",
    "    \"lane\" -> \"ln\",\n",
    "    \"parkway\" -> \"pkwy\",\n",
    "    \"turnpike\" -> \"tp\",\n",
    "    \"terrace\" -> \"ter\",\n",
    "    \"1st\" -> \"1\",\n",
    "    \"2nd\" -> \"2\",\n",
    "    \"3rd\" -> \"3\",\n",
    "    \"1th\" -> \"1\",\n",
    "    \"2th\" -> \"2\",\n",
    "    \"3th\" -> \"3\",\n",
    "    \"4th\" -> \"4\",\n",
    "    \"5th\" -> \"5\",\n",
    "    \"6th\" -> \"6\",\n",
    "    \"7th\" -> \"7\", \n",
    "    \"8th\" -> \"8\",\n",
    "    \"9th\" -> \"9\",\n",
    "    \"0th\" -> \"0\",\n",
    "    \"west \" -> \"w \",\n",
    "    \"north \" -> \"n \",\n",
    "    \"east \" -> \"e \",\n",
    "    \"south \" -> \"s \"\n",
    ")\n",
    "\n",
    "def isalnum(c: Char) : Boolean = {\n",
    "    if (c.isLetter && c <= 'z') {\n",
    "        return true\n",
    "    } else if (c.isDigit) {\n",
    "        return true\n",
    "    }\n",
    "    return false;\n",
    "}\n",
    "\n",
    "def normalizeStreet(s:String) : String = {\n",
    "    // Lowercase\n",
    "    var str = s.toLowerCase\n",
    "\n",
    "    // Delete all non-alphanumeric characters\n",
    "    if (!str.matches(\"[a-zA-Z0-9]*\")) {\n",
    "        str = str.filter(isalnum(_))\n",
    "    }\n",
    "\n",
    "    // Replace common abbreviations\n",
    "    for (k <- normalizationCode.keys) {\n",
    "        str = str.replace(k, normalizationCode(k))\n",
    "    }\n",
    "\n",
    "    return str\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "collapsed": false,
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "def getSpatial(row:Row) : List[Any] = {\n",
    "    /*\n",
    "    Computes the location identifier from the input row\n",
    "\n",
    "    Returns:\n",
    "        List of spatial key columns\n",
    "    */\n",
    "    val rowMap = row.getValuesMap(Array(\"ON STREET NAME\", \"BOROUGH\"))\n",
    "    var loc : String = rowMap(\"ON STREET NAME\")\n",
    "    loc = normalizeStreet(loc)\n",
    "    var borough : String = rowMap(\"BOROUGH\")\n",
    "    borough = borough.toLowerCase\n",
    "    \n",
    "    return List(loc, borough) \n",
    "}\n",
    "\n",
    "def getTemporal(row:Row) : List[Any] = {\n",
    "    /*\n",
    "    Computes the temporal key from a given row. \n",
    "    Unlike the python example, I do not return a Datetime object. \n",
    "    I ran into multiple issues with Datetimes, and there seemed to be no real advantage to\n",
    "    having them since later examples returned the date and time to multiple columns.\n",
    "    */\n",
    "    val rowMap = row.getValuesMap(Array(\"DATE\", \"TIME\"))\n",
    "    var date : String = rowMap(\"DATE\")\n",
    "    var time : String = rowMap(\"TIME\")\n",
    "    \n",
    "    //return \n",
    "    return List(date.split(\"/\")(2).toInt, date.split(\"/\")(0).toInt, date.split(\"/\")(1).toInt, time.split(\":\")(0).toInt)\n",
    "\n",
    "}\n",
    "\n",
    "def getRest(row:Row) : List[Any] = {\n",
    "    /*\n",
    "    Computes the rest from a given row\n",
    "    */\n",
    "    var columns = Seq.empty[String]\n",
    "    for (field <- row.schema) {\n",
    "        columns :+= field.name\n",
    "    }\n",
    "    val rowMap = row.getValuesMap(columns)\n",
    "    var rest = List.empty[Any]\n",
    "    for (col <- columns) {\n",
    "        if (!(Array(\"ON STREET NAME\", \"OFF STREET NAME\", \"CROSS STREET NAME\", \"BOROUGH\", \"DATE\", \"TIME\")  contains col)) {\n",
    "            rest :+= rowMap(col)\n",
    "        }\n",
    "    }\n",
    "    return rest\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def getCollisionsOutStruct(row:Row) : StructType = {\n",
    "    var struct = StructType(\n",
    "    StructField(\"Year\", IntegerType, true) ::\n",
    "    StructField(\"Month\", IntegerType, true) ::\n",
    "    StructField(\"Day\", IntegerType, true) ::\n",
    "    StructField(\"Hour\", IntegerType, true) ::\n",
    "    StructField(\"Street\", StringType, true) ::\n",
    "    StructField(\"Borough\", StringType, true) :: Nil)\n",
    "    for (field <- row.schema) {\n",
    "        if (collisionsOutArray contains field.name) {\n",
    "            struct = struct.add(field)\n",
    "        }\n",
    "    }\n",
    "    return struct\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val collisionsOut = collisions.rdd.map(row => Row(getTemporal(row) ::: getSpatial(row) ::: getRest(row):_*))\n",
    "var row = collisions.take(1)(0)\n",
    "var struct = getCollisionsOutStruct(row)\n",
    "var collisionsOutDF = sqlctx.createDataFrame(collisionsOut, struct)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "collapsed": false
   },
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Array(Year, Month, Day, Hour, Street, Borough, ZIP CODE, LATITUDE, LONGITUDE, LOCATION, NUMBER OF PERSONS INJURED, NUMBER OF PERSONS KILLED, NUMBER OF PEDESTRIANS INJURED, NUMBER OF PEDESTRIANS KILLED, NUMBER OF CYCLIST INJURED, NUMBER OF CYCLIST KILLED, NUMBER OF MOTORIST INJURED, NUMBER OF MOTORIST KILLED, CONTRIBUTING FACTOR VEHICLE 1, CONTRIBUTING FACTOR VEHICLE 2, CONTRIBUTING FACTOR VEHICLE 3, CONTRIBUTING FACTOR VEHICLE 4, CONTRIBUTING FACTOR VEHICLE 5, UNIQUE KEY, VEHICLE TYPE CODE 1, VEHICLE TYPE CODE 2, VEHICLE TYPE CODE 3, VEHICLE TYPE CODE 4, VEHICLE TYPE CODE 5)"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "collisionsOutDF.columns"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Investigating Variables\n",
    "\n",
    "We have to investigate the variables and find out wheather they are useful or not. We begin with plotting one of the contributing factors."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Contributing factors to collisions"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var collisionsByFactor = (collisionsOutDF\n",
    "                            .groupBy(collisionsOutDF(\"CONTRIBUTING FACTOR VEHICLE 1\").alias(\"contributingFactor\"))\n",
    "                            .count().sort($\"count\".desc).limit(24))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "data('collisionsByFactor') transpose bar x(contributingFactor) y(count:linear) sort(count:ascending) interaction(none)\n",
    "          ::\n",
    "width=800, height=600"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Running the code cell above shows you that the contributing factor can't be specified in most cases. However, factors like distraction, failure to yield right-of-way and fatigue can play a role. You can investigate and plot the other contribution factos by modifying the code above."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "### Sorting the vehicle types into groups"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "The data set has entries for a large number of car types. The following code cell regroups the car types into main categories like auto, bus, truck, taxi or other."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "val grouping = scala.collection.mutable.HashMap[String, String](\n",
    "    \"TAXI\" -> \"Taxi\",\n",
    "    \"AMBULANCE\" -> \"Other\",\n",
    "    \"BICYCLE\" -> \"Other\",\n",
    "    \"BUS\" -> \"Bus\",\n",
    "    \"FIRE TRUCK\" -> \"Other\", \n",
    "    \"LARGE COM VEH(6 OR MORE TIRES)\" -> \"Truck\",\n",
    "    \"LIVERY VEHICLE\" -> \"Truck\",\n",
    "    \"MOTORCYCLE\" -> \"Other\", \n",
    "    \"OTHER\" -> \"Other\",\n",
    "    \"PASSENGER VEHICLE\" -> \"Auto\",\n",
    "    \"PICK-UP TRUCK\" -> \"Other\",\n",
    "    \"PEDICAB\" -> \"Other\", \n",
    "    \"SCOOTER\" -> \"Other\",\n",
    "    \"SMALL COM VEH(4 TIRES)\" -> \"Truck\",\n",
    "    \"SPORT UTILITY / STATION WAGON\" -> \"Auto\", \n",
    "    \"UNKNOWN\" -> \"Other\",\n",
    "    \"VAN\" -> \"Auto\",\n",
    "    \"UNSPECIFIED\" -> \"Other\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "def groupVehicle(row : Row) : Row = {\n",
    "    var rowMap = row.getValuesMap(row.schema.fieldNames)\n",
    "    var rowSeq = row.toSeq\n",
    "    var resultSeq = Seq.empty[Any]\n",
    "    for (field <- row.schema.fieldNames) {\n",
    "        if (field.startsWith(\"VEHICLE TYPE CODE\")) {\n",
    "            if (rowMap(field) != null) {\n",
    "                rowSeq = rowSeq.updated(row.fieldIndex(field), grouping(rowMap(field)))\n",
    "            }\n",
    "        }\n",
    "    }\n",
    "    for (field <- List(\"Year\", \"Month\", \"Day\", \"Hour\", \"Street\", \"Borough\", \"NUMBER OF PERSONS INJURED\", \"NUMBER OF PERSONS KILLED\")) {\n",
    "        resultSeq :+= rowMap(field)\n",
    "    }\n",
    "    for (vehicle <- List(\"Auto\", \"Bus\", \"Truck\", \"Taxi\", \"Other\")) {\n",
    "        resultSeq :+= rowSeq.count(_ == vehicle)\n",
    "    }\n",
    "    resultSeq :+= (if (rowMap(\"NUMBER OF PERSONS INJURED\").asInstanceOf[String].toInt > 0) 1 else 0)\n",
    "    resultSeq :+= (if (rowMap(\"NUMBER OF PERSONS KILLED\").asInstanceOf[Integer] > 0) 1 else 0)\n",
    "    return Row(resultSeq:_*)\n",
    "}\n",
    "\n",
    "def getCollisionsOutCategoriesStruct(row:Row) : StructType = {\n",
    "    var struct = StructType(Nil)\n",
    "    for (field <- row.schema) {\n",
    "        if (List(\"Year\", \"Month\", \"Day\", \"Hour\", \"Street\", \"Borough\", \"NUMBER OF PERSONS INJURED\", \"NUMBER OF PERSONS KILLED\") contains field.name) {\n",
    "            struct = struct.add(field)\n",
    "        }\n",
    "    }\n",
    "    for (vehicle <- List(\"Auto\", \"Bus\", \"Truck\", \"Taxi\", \"Other\")) {\n",
    "        struct = struct.add(StructField(vehicle, IntegerType, true))\n",
    "    }\n",
    "    struct = struct.add(StructField(\"AccidentsWithInjuries\", IntegerType, true))\n",
    "    struct = struct.add(StructField(\"AccidentsWithDeaths\", IntegerType, true))\n",
    "    return struct\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var collisionsOutCategories = collisionsOutDF.map(row => groupVehicle(row))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var row = collisionsOutDF.take(1)(0)\n",
    "var struct = getCollisionsOutCategoriesStruct(row)\n",
    "var collisionsOutCategoriesDF = sqlctx.createDataFrame(collisionsOutCategories, struct)\n",
    "collisionsOutCategoriesDF = collisionsOutCategoriesDF.withColumnRenamed(\"NUMBER OF PERSONS INJURED\", \"Injured\")\n",
    "collisionsOutCategoriesDF = collisionsOutCategoriesDF.withColumnRenamed(\"NUMBER OF PERSONS KILLED\", \"Killed\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Count the number of accidents by car type, severity, street name, and borough that occurred down to the hour:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var aggregationColumns = scala.collection.mutable.HashMap[String, String](\"*\"->\"count\")\n",
    "for (field <- List(\"AccidentsWithInjuries\", \"AccidentsWithDeaths\", \"Auto\", \"Bus\", \"Truck\", \"Taxi\", \"Other\", \"Injured\", \"Killed\")) {\n",
    "    aggregationColumns(field) = \"sum\"\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var collisionsGrouped = collisionsOutCategoriesDF.groupBy(\"Year\", \"Month\", \"Day\", \"Hour\", \"Street\", \"Borough\").agg(aggregationColumns.toMap)\n",
    "for (c <- collisionsGrouped.columns) {\n",
    "    if (c.startsWith(\"sum\")) {\n",
    "        collisionsGrouped = collisionsGrouped.withColumnRenamed(c, c.substring(4,c.length -1))\n",
    "    }\n",
    "    if (c.startsWith(\"count\")) {\n",
    "        collisionsGrouped = collisionsGrouped.withColumnRenamed(c, \"NumberOfAccidents\")\n",
    "    }\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Determine the streets with the most collisions\n",
    "\n",
    "Find the top ten streets in New York where the most vehicle collisions occurred. Display the results in a bar graph and as a scatter plot:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var collisionsByStreet = collisionsOutDF.groupBy(\"Borough\", \"Street\").count().sort($\"count\".desc).limit(10)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "data('collisionsByStreet') transpose bar x(Street) y(count:linear) sort(count:ascending) interaction(none)\n",
    "          ::\n",
    "width=800, height=400"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Now you can add the information about the top 10 streets into the scatter plot."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var top10Streets = collisionsByStreet.select(\"Borough\", \"Street\").rdd.map(row => (row(0), row(1))).collect()\n",
    "\n",
    "def filterStreets(row : Row) : Row = {\n",
    "    /**\n",
    "        We only want the top 10 streets to be colored. To accomplish this we\n",
    "        will remove all street values that are not in the top 10. The nulls will\n",
    "        all be colored the same in the display generated by Brunel.\n",
    "    */\n",
    "    var rowMap = row.getValuesMap(Array(\"Borough\", \"Street\"))\n",
    "    var rowSeq = row.toSeq\n",
    "    if (!(top10Streets contains (rowMap(\"Borough\"),rowMap(\"Street\")))) {\n",
    "      return Row(rowSeq.updated(row.fieldIndex(\"Street\"), null):_*)\n",
    "    }\n",
    "    return Row(rowSeq:_*)\n",
    "}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "var top10StreetsRDD = collisionsOutDF.map(row => filterStreets(row))\n",
    "var top10StreetsDF = sqlctx.createDataFrame(top10StreetsRDD, collisionsOutDF.schema)\n",
    "//this is being done to take a subset because Brunel can not handle this large of a data set\n",
    "top10StreetsDF = top10StreetsDF.limit(25000)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "data('top10StreetsDF') x(Longitude) y(Latitude) color(Street) size(Street:[80%,80%,80%,80%,80%,80%,80%,80%,80%,80%,20%]) interaction(none)\n",
    "          ::\n",
    "width=1200, height=600"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Determining when the most collisions occurred\n",
    "\n",
    "Now find out at what time of the day the most accidents occurred and see if you can detect any interesting patterns by running the following cell:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "import org.apache.spark.sql.functions._\n",
    "collisionsGrouped = (collisionsGrouped\n",
    "                      .select(\"Bus\", \"Truck\", \"Taxi\", \"Other\", \"Hour\", \"Auto\")\n",
    "                      .groupBy(\"Hour\")\n",
    "                      .agg(sum(\"Bus\").alias(\"Bus\"), sum(\"Truck\").alias(\"Truck\"), \n",
    "                           sum(\"Taxi\").alias(\"Taxi\"), sum(\"Other\").alias(\"Other\"), sum(\"Auto\").alias(\"Auto\")))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "collapsed": false
   },
   "outputs": [],
   "source": [
    "%%brunel\n",
    "data(\"collisionsGrouped\") stack bar x(Hour) y(Auto, Taxi, Truck, Bus) color(#series) interaction(none)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": false
   },
   "source": [
    "This plot shows collisions spread across a day, with peaks during the morning and afternoon rush hours. You can see that significantly more collisions occurred during the afternoon rush hour than during the morning rush hour. Also, the most collisions involve cars by far, while buses, taxis, and trucks are involved in accidents a lot less frequently."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true
   },
   "source": [
    "<a id=\"summary\"></a>\n",
    "## Summary\n",
    "\n",
    "This notebook showed you how to analyze car vehicle accidents based on accident reports for New York and how you can use this information to learn more about the causes for collisions. If you extract  this type of information from the data, you can use it to help develop measures for preventing  vehicle accidents in accident hotspots."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Scala 2.10",
   "language": "scala",
   "name": "spark"
  },
  "language_info": {
   "name": "scala"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 0
}