notebook/02 - Simple Clustering using Apache Spark.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Assignment 02 - Simple Clustering using Apache Spark\n",
    "## 1. Preparation\n",
    "### 1.1 Requirements\n",
    "\n",
    "1. Operating System : Linux Mint 18.3 Sylvia\n",
    "2. Apache Spark 2.4.0 Binary (https://spark.apache.org/downloads.html)\n",
    "3. PySpark 2.4.0 (Apache Spark Python API)\n",
    "4. Findspark 1.3.0 (Python's library)\n",
    "5. Jupyter Notebook (https://jupyter.org/install)\n",
    "6. matplotlib-3.0.3\n",
    "7. bokeh-1.0.4\n",
    "8. pixiedust-1.1.15\n",
    "9. pandas-0.24.2\n",
    "\n",
    "### 1.2 Installation\n",
    "For Debian/Ubuntu-based Linux:\n",
    "```bash\n",
    "pip3 install matplotlib\n",
    "pip3 install bokeh\n",
    "pip3 install pixiedust\n",
    "pip3 install pandas\n",
    "```\n",
    "\n",
    "### 1.3 Description of Dataset\n",
    "* Dataset's name : [UK Road Safety : Traffic Accidents and Vehicles](https://www.kaggle.com/tsiaras/uk-road-safety-accidents-and-vehicles)\n",
    "* Description : Detailed dataset of road accidents and involved vehicles in the UK (2005-2017). Each line represents a single traffic accident (identified by the Accident_Index column) and its various properties.\n",
    "* There are two kinds of data sources in this dataset :\n",
    "\n",
    "<table>\n",
    "    <thead>\n",
    "        <tr>\n",
    "            <th>Data Sources</th>\n",
    "            <th>Number of Rows</th>\n",
    "            <th>Number of Columns</th>\n",
    "            <th>Size</th>\n",
    "            <th>File Format</th>\n",
    "        </tr>\n",
    "    </thead>\n",
    "    <tbody>\n",
    "        <tr>\n",
    "            <td>Accident_Information.csv</td>\n",
    "            <td>2047256</td>\n",
    "            <td>34</td>\n",
    "            <td>705,5 MB</td>\n",
    "            <td rowspan=3>CSV</td>\n",
    "        </tr>\n",
    "        <tr>\n",
    "            <td>Vehicle_Information.csv</td>\n",
    "            <td>2177205</td>\n",
    "            <td>24</td>\n",
    "            <td>644,4 MB</td>\n",
    "        </tr>\n",
    "        <tr>\n",
    "            <td colspan=3>Total</td>\n",
    "            <td>1,3 GB</td>\n",
    "        </tr>\n",
    "    </tbody>\n",
    "</table>\n",
    "    \n",
    "## 2. Spark Initialization"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import findspark to make pyspark importable as a regular library\n",
    "import findspark\n",
    "findspark.init('/home/mocatfrio/spark') \n",
    "\n",
    "# /home/mocatfrio/spark has symbolic link to /bin/spark-2.4.0-bin-hadoop2.7"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import SparkSession\n",
    "from pyspark.sql import SparkSession\n",
    "\n",
    "# Create Spark Session\n",
    "spark = SparkSession \\\n",
    "    .builder \\\n",
    "    .appName(\"Simple Clustering\") \\\n",
    "    .getOrCreate()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "<pyspark.sql.session.SparkSession object at 0x7f65a3a4b828>\n"
     ]
    }
   ],
   "source": [
    "# Print spark object ID\n",
    "print(spark)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. Load Dataset"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Load the dataset\n",
    "df = spark.read.load(\"/home/mocatfrio/Documents/projects/big-data/datasets/Accident_Information.csv\", \\\n",
    "                     format=\"csv\", sep=\",\", inferSchema=\"true\", header=\"true\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------+--------------+---------------+--------------+---------------+-----------------+-------------------+-------------------+-----------+-------------------------------------------+--------------------+--------------------+---------+--------------------+--------------------------+-------------------------+---------------------+----------------------+---------+-------------------------+--------------------+------------------+---------------------------------+---------------------------------------+-------------------+-----------------------+------------------+--------------------------+-----------+-----+-------------------+--------------------+----+----------+\n",
      "|Accident_Index|1st_Road_Class|1st_Road_Number|2nd_Road_Class|2nd_Road_Number|Accident_Severity|Carriageway_Hazards|               Date|Day_of_Week|Did_Police_Officer_Attend_Scene_of_Accident|    Junction_Control|     Junction_Detail| Latitude|    Light_Conditions|Local_Authority_(District)|Local_Authority_(Highway)|Location_Easting_OSGR|Location_Northing_OSGR|Longitude|LSOA_of_Accident_Location|Number_of_Casualties|Number_of_Vehicles|Pedestrian_Crossing-Human_Control|Pedestrian_Crossing-Physical_Facilities|       Police_Force|Road_Surface_Conditions|         Road_Type|Special_Conditions_at_Site|Speed_limit| Time|Urban_or_Rural_Area|  Weather_Conditions|Year|InScotland|\n",
      "+--------------+--------------+---------------+--------------+---------------+-----------------+-------------------+-------------------+-----------+-------------------------------------------+--------------------+--------------------+---------+--------------------+--------------------------+-------------------------+---------------------+----------------------+---------+-------------------------+--------------------+------------------+---------------------------------+---------------------------------------+-------------------+-----------------------+------------------+--------------------------+-----------+-----+-------------------+--------------------+----+----------+\n",
      "| 200501BS00001|             A|           3218|            NA|              0|          Serious|               None|2005-01-04 00:00:00|    Tuesday|                                          1|Data missing or o...|Not at junction o...|51.489096|            Daylight|      Kensington and Ch...|     Kensington and Ch...|               525680|                178240| -0.19117|                E01002849|                   1|                 1|                                0|                                      1|Metropolitan Police|            Wet or damp|Single carriageway|                      None|         30|17:42|              Urban|Raining no high w...|2005|        No|\n",
      "| 200501BS00002|             B|            450|             C|              0|           Slight|               None|2005-01-05 00:00:00|  Wednesday|                                          1| Auto traffic signal|          Crossroads|51.520075|Darkness - lights...|      Kensington and Ch...|     Kensington and Ch...|               524170|                181650|-0.211708|                E01002909|                   1|                 1|                                0|                                      5|Metropolitan Police|                    Dry|  Dual carriageway|                      None|         30|17:36|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00003|             C|              0|            NA|              0|           Slight|               None|2005-01-06 00:00:00|   Thursday|                                          1|Data missing or o...|Not at junction o...|51.525301|Darkness - lights...|      Kensington and Ch...|     Kensington and Ch...|               524520|                182240|-0.206458|                E01002857|                   1|                 2|                                0|                                      0|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|00:15|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00004|             A|           3220|            NA|              0|           Slight|               None|2005-01-07 00:00:00|     Friday|                                          1|Data missing or o...|Not at junction o...|51.482442|            Daylight|      Kensington and Ch...|     Kensington and Ch...|               526900|                177530|-0.173862|                E01002840|                   1|                 1|                                0|                                      0|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|10:35|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00005|  Unclassified|              0|            NA|              0|           Slight|               None|2005-01-10 00:00:00|     Monday|                                          1|Data missing or o...|Not at junction o...|51.495752|Darkness - lighti...|      Kensington and Ch...|     Kensington and Ch...|               528060|                179040|-0.156618|                E01002863|                   1|                 1|                                0|                                      0|Metropolitan Police|            Wet or damp|Single carriageway|                      None|         30|21:13|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00006|  Unclassified|              0|            NA|              0|           Slight|               None|2005-01-11 00:00:00|    Tuesday|                                          1|Data missing or o...|Not at junction o...| 51.51554|            Daylight|      Kensington and Ch...|     Kensington and Ch...|               524770|                181160|-0.203238|                E01002832|                   1|                 2|                                0|                                      0|Metropolitan Police|            Wet or damp|Single carriageway|             Oil or diesel|         30|12:40|              Urban|Raining no high w...|2005|        No|\n",
      "| 200501BS00007|             C|              0|  Unclassified|              0|           Slight|               None|2005-01-13 00:00:00|   Thursday|                                          1|Give way or uncon...|T or staggered ju...|51.512695|Darkness - lights...|      Kensington and Ch...|     Kensington and Ch...|               524220|                180830|-0.211277|                E01002875|                   1|                 2|                                0|                                      0|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|20:40|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00009|             A|            315|            NA|              0|           Slight|               None|2005-01-14 00:00:00|     Friday|                                          1|Data missing or o...|Not at junction o...| 51.50226|            Daylight|      Kensington and Ch...|     Kensington and Ch...|               525890|                179710|-0.187623|                E01002889|                   2|                 1|                                0|                                      0|Metropolitan Police|                    Dry|  Dual carriageway|                      None|         30|17:35|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00010|             A|           3212|             B|            304|           Slight|               None|2005-01-15 00:00:00|   Saturday|                                          1| Auto traffic signal|          Crossroads| 51.48342|Darkness - lights...|      Kensington and Ch...|     Kensington and Ch...|               527350|                177650|-0.167342|                E01002900|                   2|                 2|                                0|                                      5|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|22:43|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00011|             B|            450|             C|              0|           Slight|               None|2005-01-15 00:00:00|   Saturday|                                          1|Give way or uncon...|T or staggered ju...|51.512443|            Daylight|      Kensington and Ch...|     Kensington and Ch...|               524550|                180810|-0.206531|                E01002875|                   5|                 2|                                0|                                      8|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|16:00|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00012|             A|              4|             B|            325|           Slight|               None|2005-01-16 00:00:00|     Sunday|                                          1| Auto traffic signal|          Crossroads|51.494902|Darkness - lights...|      Kensington and Ch...|     Kensington and Ch...|               526240|                178900|-0.182872|                E01002835|                   1|                 1|                                0|                                      5|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|00:42|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00014|             A|           3220|             A|            308|           Slight|               None|2005-01-25 00:00:00|    Tuesday|                                          1| Auto traffic signal|          Crossroads|51.484044|Darkness - lights...|      Kensington and Ch...|     Kensington and Ch...|               526170|                177690|-0.184312|                E01002912|                   1|                 2|                                0|                                      5|Metropolitan Police|            Wet or damp|Single carriageway|                      None|         30|20:48|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00015|  Unclassified|              0|             A|           3220|           Slight|               None|2005-01-11 00:00:00|    Tuesday|                                          1|Give way or uncon...|T or staggered ju...|51.491632|            Daylight|      Kensington and Ch...|     Kensington and Ch...|               525590|                178520|-0.192366|                E01002849|                   1|                 1|                                0|                                      1|Metropolitan Police|            Wet or damp|    One way street|                      None|         30|12:55|              Urban|Raining no high w...|2005|        No|\n",
      "| 200501BS00016|             A|           3217|             A|           3216|           Slight|               None|2005-01-18 00:00:00|    Tuesday|                                          1|Give way or uncon...|T or staggered ju...|51.492622|Darkness - lights...|      Kensington and Ch...|     Kensington and Ch...|               527990|                178690|-0.157753|                E01002902|                   1|                 2|                                0|                                      0|Metropolitan Police|            Wet or damp|    One way street|                      None|         30|05:01|              Urban|Raining no high w...|2005|        No|\n",
      "| 200501BS00017|             A|              4|            NA|              0|           Slight|               None|2005-01-18 00:00:00|    Tuesday|                                          1|Data missing or o...|Not at junction o...|51.495429|            Daylight|      Kensington and Ch...|     Kensington and Ch...|               526700|                178970|-0.176224|                E01002821|                   2|                 1|                                0|                                      0|Metropolitan Police|                    Dry|  Dual carriageway|                      None|         30|11:15|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00018|             A|           3217|  Unclassified|              0|           Slight|               None|2005-01-18 00:00:00|    Tuesday|                                          1|Give way or uncon...|T or staggered ju...|51.481912|            Daylight|      Kensington and Ch...|     Kensington and Ch...|               526460|                177460| -0.18022|                E01002840|                   1|                 1|                                0|                                      1|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|10:50|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00019|  Unclassified|              0|  Unclassified|              0|          Serious|               None|2005-01-20 00:00:00|   Thursday|                                          1|Give way or uncon...|T or staggered ju...|51.500191|Darkness - lights...|      Kensington and Ch...|     Kensington and Ch...|               524680|                179450|-0.205139|                E01002864|                   1|                 2|                                0|                                      0|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|00:15|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00020|             A|           3218|             A|              4|           Slight|               None|2005-01-21 00:00:00|     Friday|                                          1|Give way or uncon...|T or staggered ju...|51.495811|            Daylight|      Kensington and Ch...|     Kensington and Ch...|               527000|                179020|-0.171887|                E01002821|                   1|                 2|                                0|                                      0|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|09:15|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00021|             B|            302|            NA|              0|           Slight|               None|2005-01-21 00:00:00|     Friday|                                          1|Data missing or o...|Not at junction o...|51.486552|Darkness - lights...|      Kensington and Ch...|     Kensington and Ch...|               527810|                178010| -0.16059|                E01002901|                   1|                 2|                                0|                                      0|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|21:16|              Urban|  Fine no high winds|2005|        No|\n",
      "| 200501BS00022|             A|              4|  Unclassified|              0|          Serious|               None|2005-01-08 00:00:00|   Saturday|                                          1|Give way or uncon...|T or staggered ju...|51.495498|Darkness - lights...|      Kensington and Ch...|     Kensington and Ch...|               526790|                178980|-0.174925|                E01002821|                   1|                 1|                                0|                                      0|Metropolitan Police|                    Dry|Single carriageway|                      None|         30|03:00|              Urban|  Fine no high winds|2005|        No|\n",
      "+--------------+--------------+---------------+--------------+---------------+-----------------+-------------------+-------------------+-----------+-------------------------------------------+--------------------+--------------------+---------+--------------------+--------------------------+-------------------------+---------------------+----------------------+---------+-------------------------+--------------------+------------------+---------------------------------+---------------------------------------+-------------------+-----------------------+------------------+--------------------------+-----------+-----+-------------------+--------------------+----+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# Print top 20 rows data\n",
    "df.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "2047256"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# Count data rows\n",
    "df.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Accident_Index: string (nullable = true)\n",
      " |-- 1st_Road_Class: string (nullable = true)\n",
      " |-- 1st_Road_Number: string (nullable = true)\n",
      " |-- 2nd_Road_Class: string (nullable = true)\n",
      " |-- 2nd_Road_Number: string (nullable = true)\n",
      " |-- Accident_Severity: string (nullable = true)\n",
      " |-- Carriageway_Hazards: string (nullable = true)\n",
      " |-- Date: timestamp (nullable = true)\n",
      " |-- Day_of_Week: string (nullable = true)\n",
      " |-- Did_Police_Officer_Attend_Scene_of_Accident: string (nullable = true)\n",
      " |-- Junction_Control: string (nullable = true)\n",
      " |-- Junction_Detail: string (nullable = true)\n",
      " |-- Latitude: string (nullable = true)\n",
      " |-- Light_Conditions: string (nullable = true)\n",
      " |-- Local_Authority_(District): string (nullable = true)\n",
      " |-- Local_Authority_(Highway): string (nullable = true)\n",
      " |-- Location_Easting_OSGR: string (nullable = true)\n",
      " |-- Location_Northing_OSGR: string (nullable = true)\n",
      " |-- Longitude: string (nullable = true)\n",
      " |-- LSOA_of_Accident_Location: string (nullable = true)\n",
      " |-- Number_of_Casualties: integer (nullable = true)\n",
      " |-- Number_of_Vehicles: integer (nullable = true)\n",
      " |-- Pedestrian_Crossing-Human_Control: string (nullable = true)\n",
      " |-- Pedestrian_Crossing-Physical_Facilities: string (nullable = true)\n",
      " |-- Police_Force: string (nullable = true)\n",
      " |-- Road_Surface_Conditions: string (nullable = true)\n",
      " |-- Road_Type: string (nullable = true)\n",
      " |-- Special_Conditions_at_Site: string (nullable = true)\n",
      " |-- Speed_limit: string (nullable = true)\n",
      " |-- Time: string (nullable = true)\n",
      " |-- Urban_or_Rural_Area: string (nullable = true)\n",
      " |-- Weather_Conditions: string (nullable = true)\n",
      " |-- Year: integer (nullable = true)\n",
      " |-- InScotland: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# inferSchema is used to inference the actual datatype of columns, especially for dates and timestamp\n",
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. Simple Clustering\n",
    "Cluster some areas or points of frequent accidents in the UK (2015-2017) according to its latitude and longitude point. \n",
    "\n",
    "Following are the steps:\n",
    "\n",
    "### 4.1 Preprocess Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import col\n",
    "from pyspark.sql.functions import col"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Convert the type of Latitude and Longitude values into float and remove null values\n",
    "df_preprocessed = (df\n",
    "       .withColumn(\"Latitude\", col(\"Latitude\").cast(\"float\"))\n",
    "       .withColumn(\"Longitude\", col(\"Longitude\").cast(\"float\"))\n",
    "       .dropna()\n",
    "       )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- Accident_Index: string (nullable = true)\n",
      " |-- 1st_Road_Class: string (nullable = true)\n",
      " |-- 1st_Road_Number: string (nullable = true)\n",
      " |-- 2nd_Road_Class: string (nullable = true)\n",
      " |-- 2nd_Road_Number: string (nullable = true)\n",
      " |-- Accident_Severity: string (nullable = true)\n",
      " |-- Carriageway_Hazards: string (nullable = true)\n",
      " |-- Date: timestamp (nullable = true)\n",
      " |-- Day_of_Week: string (nullable = true)\n",
      " |-- Did_Police_Officer_Attend_Scene_of_Accident: string (nullable = true)\n",
      " |-- Junction_Control: string (nullable = true)\n",
      " |-- Junction_Detail: string (nullable = true)\n",
      " |-- Latitude: float (nullable = true)\n",
      " |-- Light_Conditions: string (nullable = true)\n",
      " |-- Local_Authority_(District): string (nullable = true)\n",
      " |-- Local_Authority_(Highway): string (nullable = true)\n",
      " |-- Location_Easting_OSGR: string (nullable = true)\n",
      " |-- Location_Northing_OSGR: string (nullable = true)\n",
      " |-- Longitude: float (nullable = true)\n",
      " |-- LSOA_of_Accident_Location: string (nullable = true)\n",
      " |-- Number_of_Casualties: integer (nullable = true)\n",
      " |-- Number_of_Vehicles: integer (nullable = true)\n",
      " |-- Pedestrian_Crossing-Human_Control: string (nullable = true)\n",
      " |-- Pedestrian_Crossing-Physical_Facilities: string (nullable = true)\n",
      " |-- Police_Force: string (nullable = true)\n",
      " |-- Road_Surface_Conditions: string (nullable = true)\n",
      " |-- Road_Type: string (nullable = true)\n",
      " |-- Special_Conditions_at_Site: string (nullable = true)\n",
      " |-- Speed_limit: string (nullable = true)\n",
      " |-- Time: string (nullable = true)\n",
      " |-- Urban_or_Rural_Area: string (nullable = true)\n",
      " |-- Weather_Conditions: string (nullable = true)\n",
      " |-- Year: integer (nullable = true)\n",
      " |-- InScotland: string (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_preprocessed.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1902247"
      ]
     },
     "execution_count": 11,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df_preprocessed.count()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Register the dataframe as a SQL temporary view\n",
    "df_preprocessed.createOrReplaceTempView(\"traffic_accidents\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.2 Retrieve Data"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Retrieve the data needed:\n",
    "# - Latitude\n",
    "# - Longitude\n",
    "# - Value (In this case is Number of Casualties)\n",
    "\n",
    "data = spark.sql(\"SELECT `Latitude` as lat, `Longitude` as lon, `Accident_Severity`, `Number_of_Casualties`\\\n",
    "                  FROM traffic_accidents\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+-----------------+--------------------+\n",
      "|      lat|      lon|Accident_Severity|Number_of_Casualties|\n",
      "+---------+---------+-----------------+--------------------+\n",
      "|51.489098| -0.19117|          Serious|                   1|\n",
      "|51.520077|-0.211708|           Slight|                   1|\n",
      "|51.525303|-0.206458|           Slight|                   1|\n",
      "| 51.48244|-0.173862|           Slight|                   1|\n",
      "| 51.49575|-0.156618|           Slight|                   1|\n",
      "| 51.51554|-0.203238|           Slight|                   1|\n",
      "|51.512695|-0.211277|           Slight|                   1|\n",
      "| 51.50226|-0.187623|           Slight|                   2|\n",
      "| 51.48342|-0.167342|           Slight|                   2|\n",
      "|51.512444|-0.206531|           Slight|                   5|\n",
      "|51.494904|-0.182872|           Slight|                   1|\n",
      "|51.484043|-0.184312|           Slight|                   1|\n",
      "| 51.49163|-0.192366|           Slight|                   1|\n",
      "|51.492622|-0.157753|           Slight|                   1|\n",
      "| 51.49543|-0.176224|           Slight|                   2|\n",
      "| 51.48191| -0.18022|           Slight|                   1|\n",
      "| 51.50019|-0.205139|          Serious|                   1|\n",
      "| 51.49581|-0.171887|           Slight|                   1|\n",
      "|51.486553| -0.16059|           Slight|                   1|\n",
      "|  51.4955|-0.174925|          Serious|                   1|\n",
      "+---------+---------+-----------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "data.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "1902247"
      ]
     },
     "execution_count": 15,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "data.count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.3 Assemble Features"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import VectorAssembler\n",
    "from pyspark.ml.feature import VectorAssembler"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Spark Machine Learning requires the input features to be gathered in a single column of dataframe, usually named \"features\"\n",
    "vecAssembler = VectorAssembler(inputCols=[\"lat\", \"lon\"], outputCol=\"features\")\n",
    "df_assembled = vecAssembler.transform(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+-----------------+--------------------+--------------------+\n",
      "|      lat|      lon|Accident_Severity|Number_of_Casualties|            features|\n",
      "+---------+---------+-----------------+--------------------+--------------------+\n",
      "|51.489098| -0.19117|          Serious|                   1|[51.4890975952148...|\n",
      "|51.520077|-0.211708|           Slight|                   1|[51.5200767517089...|\n",
      "|51.525303|-0.206458|           Slight|                   1|[51.5253028869628...|\n",
      "| 51.48244|-0.173862|           Slight|                   1|[51.4824409484863...|\n",
      "| 51.49575|-0.156618|           Slight|                   1|[51.4957504272460...|\n",
      "| 51.51554|-0.203238|           Slight|                   1|[51.5155410766601...|\n",
      "|51.512695|-0.211277|           Slight|                   1|[51.5126953125,-0...|\n",
      "| 51.50226|-0.187623|           Slight|                   2|[51.5022583007812...|\n",
      "| 51.48342|-0.167342|           Slight|                   2|[51.4834213256835...|\n",
      "|51.512444|-0.206531|           Slight|                   5|[51.5124435424804...|\n",
      "|51.494904|-0.182872|           Slight|                   1|[51.4949035644531...|\n",
      "|51.484043|-0.184312|           Slight|                   1|[51.4840431213378...|\n",
      "| 51.49163|-0.192366|           Slight|                   1|[51.4916305541992...|\n",
      "|51.492622|-0.157753|           Slight|                   1|[51.4926223754882...|\n",
      "| 51.49543|-0.176224|           Slight|                   2|[51.4954299926757...|\n",
      "| 51.48191| -0.18022|           Slight|                   1|[51.4819107055664...|\n",
      "| 51.50019|-0.205139|          Serious|                   1|[51.5001907348632...|\n",
      "| 51.49581|-0.171887|           Slight|                   1|[51.4958114624023...|\n",
      "|51.486553| -0.16059|           Slight|                   1|[51.4865531921386...|\n",
      "|  51.4955|-0.174925|          Serious|                   1|[51.4954986572265...|\n",
      "+---------+---------+-----------------+--------------------+--------------------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "df_assembled.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.4 Cluster Data using K-Means"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import KMeans\n",
    "from pyspark.ml.clustering import KMeans"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Train a k-means model\n",
    "kmeans = KMeans().setK(3).setSeed(1)\n",
    "model = kmeans.fit(df_assembled.select('features'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.5 Transform and Show Cluster Assignment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Make predictions by transforming the initial dataframe\n",
    "transformed = model.transform(df_assembled)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------+---------+-----------------+--------------------+--------------------+----------+\n",
      "|      lat|      lon|Accident_Severity|Number_of_Casualties|            features|prediction|\n",
      "+---------+---------+-----------------+--------------------+--------------------+----------+\n",
      "|51.489098| -0.19117|          Serious|                   1|[51.4890975952148...|         0|\n",
      "|51.520077|-0.211708|           Slight|                   1|[51.5200767517089...|         0|\n",
      "|51.525303|-0.206458|           Slight|                   1|[51.5253028869628...|         0|\n",
      "| 51.48244|-0.173862|           Slight|                   1|[51.4824409484863...|         0|\n",
      "| 51.49575|-0.156618|           Slight|                   1|[51.4957504272460...|         0|\n",
      "| 51.51554|-0.203238|           Slight|                   1|[51.5155410766601...|         0|\n",
      "|51.512695|-0.211277|           Slight|                   1|[51.5126953125,-0...|         0|\n",
      "| 51.50226|-0.187623|           Slight|                   2|[51.5022583007812...|         0|\n",
      "| 51.48342|-0.167342|           Slight|                   2|[51.4834213256835...|         0|\n",
      "|51.512444|-0.206531|           Slight|                   5|[51.5124435424804...|         0|\n",
      "|51.494904|-0.182872|           Slight|                   1|[51.4949035644531...|         0|\n",
      "|51.484043|-0.184312|           Slight|                   1|[51.4840431213378...|         0|\n",
      "| 51.49163|-0.192366|           Slight|                   1|[51.4916305541992...|         0|\n",
      "|51.492622|-0.157753|           Slight|                   1|[51.4926223754882...|         0|\n",
      "| 51.49543|-0.176224|           Slight|                   2|[51.4954299926757...|         0|\n",
      "| 51.48191| -0.18022|           Slight|                   1|[51.4819107055664...|         0|\n",
      "| 51.50019|-0.205139|          Serious|                   1|[51.5001907348632...|         0|\n",
      "| 51.49581|-0.171887|           Slight|                   1|[51.4958114624023...|         0|\n",
      "|51.486553| -0.16059|           Slight|                   1|[51.4865531921386...|         0|\n",
      "|  51.4955|-0.174925|          Serious|                   1|[51.4954986572265...|         0|\n",
      "+---------+---------+-----------------+--------------------+--------------------+----------+\n",
      "only showing top 20 rows\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# \"prediction\" column shows the cluster assignment \n",
    "transformed.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.6 Evaluate the Cluster"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Import ClusteringEvaluator\n",
    "from pyspark.ml.evaluation import ClusteringEvaluator"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Evaluate clustering by computing Silhouette score\n",
    "evaluator = ClusteringEvaluator()\n",
    "silhouette = evaluator.evaluate(transformed)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Silhouette with squared euclidean distance = 0.718443635599194\n"
     ]
    }
   ],
   "source": [
    "# Print the Silhouette score\n",
    "print(\"Silhouette with squared euclidean distance = \" + str(silhouette))\n",
    "\n",
    "# The silhouette value is a measure of how similar an object is to its own cluster (cohesion) compared to other clusters (separation). \n",
    "# The silhouette ranges from −1 to +1, where a high value indicates that the object is well matched to its own cluster and poorly matched to neighboring clusters. \n",
    "# -Wikipedia"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From several experiments, I got this result\n",
    "\n",
    "| k | Silhouette Score |\n",
    "|---|---|\n",
    "| 2 | 0.6301058414017123 |\n",
    "| 3 | 0.718443635599194 |\n",
    "| 5 | 0.5516203069472314 |\n",
    "| 7 | 0.5524358973809714 |\n",
    "| 10 | 0.627209303002482 |\n",
    "\n",
    "So, I'll choose k that has the biggest silhouette score "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.7 Show the Cluster Centers"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "centers = model.clusterCenters()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Cluster Centers: \n",
      "[51.58280613 -0.13750819]\n",
      "[53.42604169 -1.8494087 ]\n",
      "[51.2695775  -3.10019804]\n"
     ]
    }
   ],
   "source": [
    "# Print the cluster centers\n",
    "print(\"Cluster Centers: \")\n",
    "for center in centers:\n",
    "    print(center)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### 4.8 Export DataFrame to CSV"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Convert to Pandas\n",
    "import pandas as pd"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [],
   "source": [
    "# \"transformed\" dataframe must be converted into CSV first, so mapbox API can render it\n",
    "transformed.toPandas().to_csv('/home/mocatfrio/Documents/projects/big-data/datasets/accident_area.csv')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 5. Visualization"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Pixiedust database opened successfully\n"
     ]
    },
    {
     "data": {
      "text/html": [
       "\n",
       "        <div style=\"margin:10px\">\n",
       "            <a href=\"https://github.com/ibm-watson-data-lab/pixiedust\" target=\"_new\">\n",
       "                <img src=\"https://github.com/ibm-watson-data-lab/pixiedust/raw/master/docs/_static/pd_icon32.png\" style=\"float:left;margin-right:10px\"/>\n",
       "            </a>\n",
       "            <span>Pixiedust version 1.1.15</span>\n",
       "        </div>\n",
       "        "
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# I'll use Pixiedust to do visualization\n",
    "import pixiedust"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Using Mapbox\n",
    "\n",
    "Mapbox needs at least the following three fields in order to work with this renderer:\n",
    "\n",
    "* a latitude field named `latitude`, `lat`, or `y`\n",
    "* longitude field named `longitude`, `lon`, `long`, or `x`\n",
    "* a numeric field for visualization\n",
    "\n",
    "To use the Mapbox renderer, you need a **free API key from Mapbox**. You can get one on their website at https://www.mapbox.com/signup/. When you get your key, enter it in the Options dialog box.\n",
    "\n",
    "In the Options dialog, drag both your **latitude** and **longitude** fields into **Keys**. Then choose any **numeric fields** for **Values**. Only the first one you choose is used to color the map thematically, but any other fields specified in Values appear in a pop-up information bubble when you hover your mouse over a data point on the map.\n",
    "\n",
    "![img3](img/img3.png)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Downloading 'file:///home/mocatfrio/Documents/projects/big-data/datasets/accident_area.csv' from file:///home/mocatfrio/Documents/projects/big-data/datasets/accident_area.csv\n",
      "Downloaded 151974608 bytes\n",
      "Creating pySpark DataFrame for 'file:///home/mocatfrio/Documents/projects/big-data/datasets/accident_area.csv'. Please wait...\n",
      "Loading file using 'SparkSession'\n",
      "Successfully created pySpark DataFrame for 'file:///home/mocatfrio/Documents/projects/big-data/datasets/accident_area.csv'\n"
     ]
    }
   ],
   "source": [
    "# Open the CSV file\n",
    "dataset = pixiedust.sampleData('file:///home/mocatfrio/Documents/projects/big-data/datasets/accident_area.csv')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {
    "pixiedust": {
     "displayParams": {
      "basemap": "light-v9",
      "chartsize": "70",
      "coloropacity": "100",
      "colorrampname": "Custom Color Range",
      "custombasecolor": "#f15454",
      "custombasecolorsecondary": "#ffc000",
      "handlerId": "mapView",
      "keyFields": "lat,lon",
      "kind": "choropleth",
      "legend": "true",
      "mapboxtoken": "pk.eyJ1IjoibW9jYXRmcmlvIiwiYSI6ImNqdGU2ZmpubzBrMjg0NHFpdTYwbTJ0MHoifQ.H4VFTZcz7HaXtBgy-m2a3g",
      "numbins": "5",
      "rowCount": "1000",
      "valueFields": "Accident_Severity,Number_of_Casualties"
     }
    }
   },
   "outputs": [
    {
     "data": {
      "text/html": [
       "<style type=\"text/css\">.pd_warning{display:none;}</style><div class=\"pd_warning\"><em>Hey, there's something awesome here! To see it, open this notebook outside GitHub, in a viewer like Jupyter</em></div>"
      ],
      "text/plain": [
       "<IPython.core.display.HTML object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "display(dataset)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Simple\n",
    "    \n",
    "    ![img4](img/img4.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Choropleth\n",
    "\n",
    "    ![img2](img/img2.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "* Choropleth Cluster\n",
    "    \n",
    "    ![img1](img/img1.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## References\n",
    "\n",
    "* https://spark.apache.org/docs/2.3.1/ml-clustering.html\n",
    "* https://stackoverflow.com/questions/47585723/kmeans-clustering-in-pyspark\n",
    "* https://pixiedust.github.io/pixiedust/displayapi.html\n",
    "* https://github.com/pixiedust/pixiedust/wiki/Tutorial:-Using-Notebooks-with-PixieDust-for-Fast,-Flexible,-and-Easier-Data-Analysis-and-Experimentation"
   ]
  }
 ],
 "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.5.2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}