SFFireDeptDataAnalysis_AWS.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Analyzing SF Fire report using AWS EMR and Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Steps \n",
    "\n",
    "1. Create emr clsuter in aws console\n",
    "\n",
    "Open the Amazon EMR console at https://console.aws.amazon.com/elasticmapreduce/\n",
    "\n",
    "[Steps to create EMR cluster](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-gs-launch-sample-cluster.html)\n",
    "\n",
    "[Steps to create jupyter notebook](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-managed-notebooks-create.html)\n",
    "\n",
    "A S3 bucket will be created during creation of S3 EMR clsuter(You can specify an existing bucket name if you already have otherwise a new s3 bucket will be created).\n",
    "\n",
    "2. Upload Data File to S3\n",
    "    - Boot an ec2 machine \n",
    "    - Configure AWS CLI \n",
    "    \n",
    "            aws configure\n",
    "    \n",
    "        Provide access key id , secret acess key,region name(same as ec2 machine region)\n",
    "        \n",
    "    - Download File from URL to ec2 machine local disk\n",
    "    \n",
    "            wget 'https://data.sfgov.org/api/views/nuek-vuh3/rows.csv?accessType=DOWNLOAD' -O sffire.csv\n",
    "\n",
    "    - Upload File to s3 using CLI command\n",
    "    \n",
    "            aws s3 cp sffire.csv s3://bucketname/sffire.csv\n",
    "3. Upload this notebook in your jupyter server and execute the cells\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 97,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "3cd40f52d3aa4c109584ed42a8a08cb1",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "import os"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "eafc40733fa0487ab78d89cbf0f3065e",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "df = spark.read.csv('s3://aws-emr-resources-504809190933-us-west-2/Data/sffire.csv',header=True,inferSchema=True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "33b437e0ccab4c588d4535344fd0a56f",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+-----------+-------+---------------+---------+----------+----------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------+-------------+----------------------+--------------------+-------------------+-------------+-------------------+---------+------------+----+-----------------+--------+--------------+--------+---------------+----------------+---------+------------------------------+------------------------+-------------------+------------------------------------+--------------------+-------------+\n",
      "|Call Number|Unit ID|Incident Number|Call Type| Call Date|Watch Date|       Received DtTm|          Entry DtTm|       Dispatch DtTm|       Response DtTm|       On Scene DtTm|Transport DtTm|Hospital DtTm|Call Final Disposition|      Available DtTm|            Address|         City|Zipcode of Incident|Battalion|Station Area| Box|Original Priority|Priority|Final Priority|ALS Unit|Call Type Group|Number of Alarms|Unit Type|Unit sequence in call dispatch|Fire Prevention District|Supervisor District|Neighborhooods - Analysis Boundaries|            Location|        RowID|\n",
      "+-----------+-------+---------------+---------+----------+----------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------+-------------+----------------------+--------------------+-------------------+-------------+-------------------+---------+------------+----+-----------------+--------+--------------+--------+---------------+----------------+---------+------------------------------+------------------------+-------------------+------------------------------------+--------------------+-------------+\n",
      "|  192910017|    E11|       19125164|   Alarms|10/18/2019|10/17/2019|10/18/2019 12:03:...|10/18/2019 12:06:...|10/18/2019 12:07:...|10/18/2019 12:08:...|10/18/2019 12:11:...|          null|         null|                  Fire|10/18/2019 12:33:...|24TH ST/VALENCIA ST|San Francisco|              94110|      B06|          11|5525|                3|       3|             3|    true|          Alarm|               1|   ENGINE|                             1|                       6|                  9|                             Mission|(37.7521036457482...|192910017-E11|\n",
      "+-----------+-------+---------------+---------+----------+----------+--------------------+--------------------+--------------------+--------------------+--------------------+--------------+-------------+----------------------+--------------------+-------------------+-------------+-------------------+---------+------------+----+-----------------+--------+--------------+--------+---------------+----------------+---------+------------------------------+------------------------+-------------------+------------------------------------+--------------------+-------------+\n",
      "only showing top 1 row"
     ]
    }
   ],
   "source": [
    "df.show(1)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e80314a14ed940a7b655b856df661029",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "StructType(List(StructField(Call Number,IntegerType,true\n",
      "StructField(Unit ID,StringType,true\n",
      "StructField(Incident Number,IntegerType,true\n",
      "StructField(Call Type,StringType,true\n",
      "StructField(Call Date,StringType,true\n",
      "StructField(Watch Date,StringType,true\n",
      "StructField(Received DtTm,StringType,true\n",
      "StructField(Entry DtTm,StringType,true\n",
      "StructField(Dispatch DtTm,StringType,true\n",
      "StructField(Response DtTm,StringType,true\n",
      "StructField(On Scene DtTm,StringType,true\n",
      "StructField(Transport DtTm,StringType,true\n",
      "StructField(Hospital DtTm,StringType,true\n",
      "StructField(Call Final Disposition,StringType,true\n",
      "StructField(Available DtTm,StringType,true\n",
      "StructField(Address,StringType,true\n",
      "StructField(City,StringType,true\n",
      "StructField(Zipcode of Incident,IntegerType,true\n",
      "StructField(Battalion,StringType,true\n",
      "StructField(Station Area,StringType,true\n",
      "StructField(Box,StringType,true\n",
      "StructField(Original Priority,StringType,true\n",
      "StructField(Priority,StringType,true\n",
      "StructField(Final Priority,IntegerType,true\n",
      "StructField(ALS Unit,BooleanType,true\n",
      "StructField(Call Type Group,StringType,true\n",
      "StructField(Number of Alarms,IntegerType,true\n",
      "StructField(Unit Type,StringType,true\n",
      "StructField(Unit sequence in call dispatch,IntegerType,true\n",
      "StructField(Fire Prevention District,StringType,true\n",
      "StructField(Supervisor District,StringType,true\n",
      "StructField(Neighborhooods - Analysis Boundaries,StringType,true\n",
      "StructField(Location,StringType,true\n",
      "StructField(RowID,StringType,true)))"
     ]
    }
   ],
   "source": [
    "for s in str(df.schema).split('),'):\n",
    "    print(s)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Tasks\n",
    "\n",
    "1. What were all the different types of fire calls in 2018?\n",
    "\n",
    "2. What months within the year 2018 saw for the highest number of fire calls?\n",
    "\n",
    "3. Which neighborhood in SF generated the most fire calls in 2018?\n",
    "\n",
    "4. Which neighborhoods in SF had the worst response time to fire calls in 2018?\n",
    "\n",
    "5. Which week in the year in 2018 had the most fire calls?\n",
    "\n",
    "6. Is there a correlation between neighborhood, zip code, and fire calls?\n",
    "\n",
    "7. How can we use Parquet files or SQL tables to store this data and read it back?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "9c389749bcf44e9888aaf455c70316bb",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "df = df.cache()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "192fa2f9ab7a4bd08f960a3554a2a8a0",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------+\n",
      "|           call type|\n",
      "+--------------------+\n",
      "|Elevator / Escala...|\n",
      "|         Marine Fire|\n",
      "|  Aircraft Emergency|\n",
      "|Confined Space / ...|\n",
      "|      Administrative|\n",
      "|              Alarms|\n",
      "|Odor (Strange / U...|\n",
      "|Lightning Strike ...|\n",
      "|Citizen Assist / ...|\n",
      "|              HazMat|\n",
      "|Watercraft in Dis...|\n",
      "|           Explosion|\n",
      "|           Oil Spill|\n",
      "|        Vehicle Fire|\n",
      "|  Suspicious Package|\n",
      "|   Train / Rail Fire|\n",
      "|Extrication / Ent...|\n",
      "|               Other|\n",
      "|        Outside Fire|\n",
      "|   Traffic Collision|\n",
      "+--------------------+\n",
      "only showing top 20 rows"
     ]
    }
   ],
   "source": [
    "df.select('call type').distinct().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 71,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "8b7fa11f9018409bad24373516c95acb",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "sdf = df.select('call type','call date','Neighborhooods - Analysis Boundaries','Response DtTm','Zipcode of Incident','Received DtTm')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 72,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "9e814abb88044381a367171480abf434",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "from pyspark.sql.functions import *\n",
    "sdf = (sdf.withColumn('call_date',to_timestamp(col('call date'), 'MM/dd/yyyy')).drop('call date')\n",
    "    .withColumn('res_date',to_timestamp(col('Response DtTm'), 'MM/dd/yyyy hh:mm:ss')).drop('Response DtTm')\n",
    "       .withColumn('recv_date',to_timestamp(col('Received DtTm'), 'MM/dd/yyyy hh:mm:ss')).drop('Received DtTm')\n",
    "      .filter(year('call_date')==2018))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 73,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "ac59785baf7b41748342b7a0869bc1ee",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+------------------------------------+-------------------+-------------------+-------------------+-------------------+\n",
      "|       call type|Neighborhooods - Analysis Boundaries|Zipcode of Incident|          call_date|           res_date|          recv_date|\n",
      "+----------------+------------------------------------+-------------------+-------------------+-------------------+-------------------+\n",
      "|          Alarms|                          Tenderloin|              94109|2018-02-02 00:00:00|2018-02-02 00:37:53|2018-02-02 00:33:20|\n",
      "|Medical Incident|                             Mission|              94110|2018-02-01 00:00:00|2018-02-01 02:42:25|2018-02-01 02:39:08|\n",
      "|Medical Incident|                      Haight Ashbury|              94117|2018-02-01 00:00:00|2018-02-01 03:07:42|2018-02-01 03:03:58|\n",
      "|Medical Incident|                Financial Distric...|              94111|2018-02-01 00:00:00|2018-02-01 03:28:30|2018-02-01 03:25:11|\n",
      "|Medical Incident|                     South of Market|              94103|2018-02-01 00:00:00|2018-02-01 03:46:29|2018-02-01 03:44:30|\n",
      "|Medical Incident|                Bayview Hunters P...|              94124|2018-02-01 00:00:00|2018-02-01 03:57:24|2018-02-01 03:53:31|\n",
      "|Medical Incident|                      Outer Richmond|              94121|2018-02-01 00:00:00|2018-02-01 04:16:02|2018-02-01 04:12:38|\n",
      "|Medical Incident|                             Mission|              94110|2018-02-01 00:00:00|2018-02-01 04:36:07|2018-02-01 04:31:35|\n",
      "|Medical Incident|                      Outer Richmond|              94121|2018-02-01 00:00:00|2018-02-01 04:36:27|2018-02-01 04:32:02|\n",
      "|Medical Incident|                            Nob Hill|              94109|2018-02-01 00:00:00|2018-02-01 04:56:15|2018-02-01 04:54:04|\n",
      "+----------------+------------------------------------+-------------------+-------------------+-------------------+-------------------+\n",
      "only showing top 10 rows"
     ]
    }
   ],
   "source": [
    "sdf=sdf.cache()\n",
    "sdf.show(10)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Q.1 What were all the different types of fire calls in 2018?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 50,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "34e64ca8959c4e69bded7bb0ece75d22",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+--------------------------------------------+\n",
      "|call type                                   |\n",
      "+--------------------------------------------+\n",
      "|Elevator / Escalator Rescue                 |\n",
      "|Marine Fire                                 |\n",
      "|Confined Space / Structure Collapse         |\n",
      "|Administrative                              |\n",
      "|Alarms                                      |\n",
      "|Odor (Strange / Unknown)                    |\n",
      "|Citizen Assist / Service Call               |\n",
      "|HazMat                                      |\n",
      "|Watercraft in Distress                      |\n",
      "|Explosion                                   |\n",
      "|Vehicle Fire                                |\n",
      "|Suspicious Package                          |\n",
      "|Train / Rail Fire                           |\n",
      "|Extrication / Entrapped (Machinery, Vehicle)|\n",
      "|Other                                       |\n",
      "|Outside Fire                                |\n",
      "|Traffic Collision                           |\n",
      "|Assist Police                               |\n",
      "|Gas Leak (Natural and LP Gases)             |\n",
      "|Water Rescue                                |\n",
      "+--------------------------------------------+\n",
      "only showing top 20 rows"
     ]
    }
   ],
   "source": [
    "sdf.select('call type').distinct().show(truncate=False)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Q.2 What months within the year 2018 saw for the highest number of fire calls?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 56,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "f57ea107e5c240e1b895097f5623e18b",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "maxcall=sdf.groupBy(month('call_date')).count().agg(max('count')).take(1)\n",
    "maxcall=maxcall[0][0]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 65,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "1799cb175fca4520b647ddab4b5da725",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+-----+\n",
      "|month(call_date)|count|\n",
      "+----------------+-----+\n",
      "|               1|27027|\n",
      "|               2|24252|\n",
      "|               3|26606|\n",
      "|               4|25565|\n",
      "|               5|26297|\n",
      "|               6|26189|\n",
      "|               7|25964|\n",
      "|               8|25341|\n",
      "|               9|24602|\n",
      "|              10|26536|\n",
      "|              11|26307|\n",
      "|              12|26014|\n",
      "+----------------+-----+\n",
      "\n",
      "+----------------+-----+\n",
      "|month(call_date)|count|\n",
      "+----------------+-----+\n",
      "|               1|27027|\n",
      "+----------------+-----+"
     ]
    }
   ],
   "source": [
    "sdf.groupBy(month('call_date')).count().orderBy('month(call_date)').show()\n",
    "\n",
    "sdf.groupBy(month('call_date')).count().filter(col('count')==maxcall).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Q.3 Which neighborhood in SF generated the most fire calls in 2018?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 70,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cc88036f937b415d80a341953d292a0f",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[Row(Neighborhooods - Analysis Boundaries='Tenderloin', count=43894)]"
     ]
    }
   ],
   "source": [
    "sdf.groupBy('Neighborhooods - Analysis Boundaries').count().sort('count',ascending=False).take(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Q.4 Which neighborhoods in SF had the worst response time to fire calls in 2018?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 83,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "a4097db8e5e442feac406d893bb0816d",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------------+--------+\n",
      "|Neighborhooods - Analysis Boundaries|res_time|\n",
      "+------------------------------------+--------+\n",
      "|                  West of Twin Peaks|   88445|\n",
      "|                             Mission|   87956|\n",
      "|                           Chinatown|   87292|\n",
      "|                             Mission|   83365|\n",
      "|                  West of Twin Peaks|   75599|\n",
      "|                           Chinatown|   73617|\n",
      "|                           Chinatown|   72687|\n",
      "|                           Chinatown|   72676|\n",
      "|                           Chinatown|   72602|\n",
      "|                           Chinatown|   72348|\n",
      "|                     Sunset/Parkside|   70644|\n",
      "|                Financial Distric...|   68384|\n",
      "|                Financial Distric...|   68378|\n",
      "|                Financial Distric...|   68378|\n",
      "|                Financial Distric...|   68378|\n",
      "|                Financial Distric...|   68378|\n",
      "|                Financial Distric...|   68378|\n",
      "|                Financial Distric...|   68378|\n",
      "|                Financial Distric...|   68378|\n",
      "|                Financial Distric...|   68315|\n",
      "+------------------------------------+--------+\n",
      "only showing top 20 rows"
     ]
    }
   ],
   "source": [
    "resp_time_df=sdf.withColumn('res_time',col('res_date').cast(\"long\")-col('recv_date').cast(\"long\"))\n",
    "resp_time_df.select('Neighborhooods - Analysis Boundaries','res_time').sort('res_time',ascending=False).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 86,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "3bf49d51368640bdbeadbb4b7a959a93",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+------------------------------------+------------------+\n",
      "|Neighborhooods - Analysis Boundaries|     avg(res_time)|\n",
      "+------------------------------------+------------------+\n",
      "|                     Treasure Island| 695.8151029748284|\n",
      "|                            Presidio|430.21208141825343|\n",
      "|                             Portola| 328.8081374956156|\n",
      "|                        Russian Hill| 308.4644152311877|\n",
      "|                Financial Distric...| 283.6726095402547|\n",
      "|                     Sunset/Parkside|268.83954664668414|\n",
      "|                      Bernal Heights|247.84918032786885|\n",
      "|                Oceanview/Merced/...|244.77176948464077|\n",
      "|                           Chinatown|240.81956299078183|\n",
      "|                              Marina|239.95751469353485|\n",
      "|                          Noe Valley|226.03992395437263|\n",
      "|                 Castro/Upper Market| 224.5785999003488|\n",
      "|                             Mission|224.21023607176582|\n",
      "|                Bayview Hunters P...| 213.5516316787653|\n",
      "|                   Visitacion Valley|207.84369840396258|\n",
      "|                        Hayes Valley| 207.2489812015249|\n",
      "|                      Inner Richmond| 206.0422069666921|\n",
      "|                                None|204.07291666666666|\n",
      "|                  West of Twin Peaks|203.32785070377986|\n",
      "|                         Mission Bay|  197.676801332778|\n",
      "+------------------------------------+------------------+\n",
      "only showing top 20 rows"
     ]
    }
   ],
   "source": [
    "resp_time_df.groupBy('Neighborhooods - Analysis Boundaries').agg(avg('res_time')).sort('avg(res_time)',ascending=False).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Q.5 Which week in the year in 2018 had the most fire calls?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 89,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "e4913f7adf3e4af19b4554a0070c82d9",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+---------------------+-----+\n",
      "|weekofyear(call_date)|count|\n",
      "+---------------------+-----+\n",
      "|                    1| 7545|\n",
      "|                   25| 6425|\n",
      "|                   49| 6354|\n",
      "|                   22| 6328|\n",
      "|                   13| 6321|\n",
      "|                   27| 6289|\n",
      "|                   40| 6252|\n",
      "|                   44| 6250|\n",
      "|                   16| 6217|\n",
      "|                   46| 6209|\n",
      "|                   43| 6200|\n",
      "|                    5| 6160|\n",
      "|                   18| 6152|\n",
      "|                   48| 6142|\n",
      "|                    2| 6109|\n",
      "|                    9| 6079|\n",
      "|                   21| 6073|\n",
      "|                   45| 6050|\n",
      "|                    6| 6025|\n",
      "|                    8| 6014|\n",
      "+---------------------+-----+\n",
      "only showing top 20 rows"
     ]
    }
   ],
   "source": [
    "sdf.groupBy(weekofyear('call_date')).count().sort('count',ascending=False).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Q.7 How can we use Parquet files or SQL tables to store this data and read it back?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 94,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "cfc87e36f3674a82a2441990929087c0",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "sdf_new = (sdf.withColumnRenamed('call type','call_type')\n",
    ".withColumnRenamed('Neighborhooods - Analysis Boundaries','Neighborhooods')\n",
    ".withColumnRenamed('Zipcode of Incident','Zipcode'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 95,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "3e47bcf16608410d9e167392e18efff7",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "\n",
    "parquet_path = 's3://aws-emr-resources-504809190933-us-west-2/Data/SFFire.parquet'\n",
    "sdf_new.write.format('parquet').save(parquet_path)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 96,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "7b1c4acd1dd44fbcb5d4984da31806fe",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "VBox()"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "application/vnd.jupyter.widget-view+json": {
       "model_id": "",
       "version_major": 2,
       "version_minor": 0
      },
      "text/plain": [
       "FloatProgress(value=0.0, bar_style='info', description='Progress:', layout=Layout(height='25px', width='50%'),…"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "+----------------+--------------------+-------+-------------------+-------------------+-------------------+\n",
      "|       call_type|      Neighborhooods|Zipcode|          call_date|           res_date|          recv_date|\n",
      "+----------------+--------------------+-------+-------------------+-------------------+-------------------+\n",
      "|          Alarms|          Tenderloin|  94109|2018-02-02 00:00:00|2018-02-02 00:37:53|2018-02-02 00:33:20|\n",
      "|Medical Incident|             Mission|  94110|2018-02-01 00:00:00|2018-02-01 02:42:25|2018-02-01 02:39:08|\n",
      "|Medical Incident|      Haight Ashbury|  94117|2018-02-01 00:00:00|2018-02-01 03:07:42|2018-02-01 03:03:58|\n",
      "|Medical Incident|Financial Distric...|  94111|2018-02-01 00:00:00|2018-02-01 03:28:30|2018-02-01 03:25:11|\n",
      "|Medical Incident|     South of Market|  94103|2018-02-01 00:00:00|2018-02-01 03:46:29|2018-02-01 03:44:30|\n",
      "|Medical Incident|Bayview Hunters P...|  94124|2018-02-01 00:00:00|2018-02-01 03:57:24|2018-02-01 03:53:31|\n",
      "|Medical Incident|      Outer Richmond|  94121|2018-02-01 00:00:00|2018-02-01 04:16:02|2018-02-01 04:12:38|\n",
      "|Medical Incident|             Mission|  94110|2018-02-01 00:00:00|2018-02-01 04:36:07|2018-02-01 04:31:35|\n",
      "|Medical Incident|      Outer Richmond|  94121|2018-02-01 00:00:00|2018-02-01 04:36:27|2018-02-01 04:32:02|\n",
      "|Medical Incident|            Nob Hill|  94109|2018-02-01 00:00:00|2018-02-01 04:56:15|2018-02-01 04:54:04|\n",
      "|Medical Incident|  West of Twin Peaks|  94127|2018-02-01 00:00:00|2018-02-01 05:00:15|2018-02-01 04:54:21|\n",
      "|Medical Incident|         Mission Bay|  94158|2018-02-01 00:00:00|2018-02-01 05:15:40|2018-02-01 05:12:07|\n",
      "|Medical Incident|Oceanview/Merced/...|  94132|2018-02-01 00:00:00|2018-02-01 05:27:24|2018-02-01 05:23:52|\n",
      "|Medical Incident|   Visitacion Valley|  94134|2018-02-01 00:00:00|2018-02-01 05:39:00|2018-02-01 05:27:31|\n",
      "|Medical Incident|             Mission|  94110|2018-02-01 00:00:00|2018-02-01 05:50:36|2018-02-01 05:47:41|\n",
      "|Medical Incident|     South of Market|  94103|2018-02-01 00:00:00|2018-02-01 06:27:13|2018-02-01 06:22:51|\n",
      "|Medical Incident|     South of Market|  94103|2018-02-01 00:00:00|2018-02-01 06:33:55|2018-02-01 06:29:48|\n",
      "|Medical Incident|     Sunset/Parkside|  94116|2018-02-01 00:00:00|2018-02-01 06:47:58|2018-02-01 06:45:49|\n",
      "|Medical Incident|      Haight Ashbury|  94117|2018-02-01 00:00:00|2018-02-01 07:07:02|2018-02-01 07:05:38|\n",
      "|Medical Incident|    Western Addition|  94102|2018-02-01 00:00:00|2018-02-01 07:22:51|2018-02-01 07:20:20|\n",
      "+----------------+--------------------+-------+-------------------+-------------------+-------------------+\n",
      "only showing top 20 rows"
     ]
    }
   ],
   "source": [
    "spark.read.parquet(parquet_path).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "PySpark",
   "language": "",
   "name": "pysparkkernel"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "python",
    "version": 2
   },
   "mimetype": "text/x-python",
   "name": "pyspark",
   "pygments_lexer": "python2"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}