spark+DataFrame (1).ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "spark\r\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file1 = spark.read.csv(\"Police_Department_Incident_Reports__2018_to_Present.csv\", header = True, inferSchema = True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.types import StructType, StructField, IntegerType, StringType, BooleanType, DoubleType, LongType"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "fileSchema = StructType([StructField('Incident_DateTime', StringType(),True),\n",
    "                        StructField('Incident_Date', StringType(),True),\n",
    "                        StructField('Incident_Time', StringType(),True),\n",
    "                        StructField('Incident_Year', IntegerType(),True),\n",
    "                        StructField('Incident_DayOfWeek', StringType(),True),\n",
    "                        StructField('Report_DateTime', StringType(),True),\n",
    "                        StructField('Row_Id', LongType(),True),\n",
    "                        StructField('Incident_Id', IntegerType(),True),\n",
    "                        StructField('Incident_Number', IntegerType(),True),\n",
    "                        StructField('Cad_Number', IntegerType(),True),\n",
    "                        StructField('Report_Type_Code', StringType(),True),\n",
    "                        StructField('Report_Type_Description', StringType(),True),\n",
    "                        StructField('File_Online', BooleanType(),True),\n",
    "                        StructField('Incident_Code', IntegerType(),True),\n",
    "                        StructField('Incident_Category', StringType(),True),\n",
    "                        StructField('Incident_Subcategory', StringType(),True),\n",
    "                        StructField('Incident_Description', StringType(),True), \n",
    "                        StructField('Resolution', StringType(),True),\n",
    "                        StructField('Intersection', StringType(),True),\n",
    "                        StructField('CNN', DoubleType(),True),\n",
    "                        StructField('Police_District', StringType(),True),\n",
    "                        StructField('Analysis_Neighbourhood', StringType(),True),\n",
    "                        StructField('Supervisor_District', IntegerType(),True),\n",
    "                        StructField('Latitude', DoubleType(),True), \n",
    "                        StructField('Longitide', DoubleType(),True),\n",
    "                        StructField('Point', StringType(),True), \n",
    "                        ])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file1 = spark.read.csv(\"Police_Department_Incident_Reports__2018_to_Present.csv\", header = True, schema = fileSchema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file1.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file1.show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file1.columns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file1.select('Incident_id', 'Incident_Category').show(5)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file1.select('Incident_Category').distinct().show(truncate = True)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file1.select('Incident_Category').groupBy('Incident_Category').count().orderBy(\"count\", ascending = False).show(52, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import *"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Analyzing datetime columns in the data set"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file1.select(\"Incident_DateTime\").show(5, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "pattern1 = 'yyyy/MM/dd hh:mm:ss aa'\n",
    "file2 = file1.withColumn('Incident_DateTime', unix_timestamp(file1['Incident_DateTime'], pattern1).cast('timestamp'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file2.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file2.select(year('Incident_DateTime')).distinct().show(truncate = False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file1.select('Incident_DateTime', 'Incident_Date','Incident_Time', 'Incident_Year', 'Report_DateTime').show(5, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "pattern1 = 'yyyy/MM/dd hh:mm:ss aa'\n",
    "pattern2 = 'yyyy/MM/dd'\n",
    "pattern3 = 'hh:mm'\n",
    "pattern4 = 'yyyy'\n",
    "file2 = file1.withColumn('Incident_DateTime', unix_timestamp(file1['Incident_DateTime'], pattern1).cast('timestamp'))\\\n",
    ".withColumn('Incident_Date', unix_timestamp(file1['Incident_Date'], pattern2).cast('timestamp'))\\\n",
    ".withColumn('Incident_Time', unix_timestamp(file1['Incident_Time'], pattern3).cast('timestamp'))\\\n",
    ".withColumn('Report_DateTime', unix_timestamp(file1['Report_DateTime'], pattern1).cast('timestamp'))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file2.printSchema()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "####   Analysis 1 ##############\n",
    "# Find the days of the week on which maximum incidents has happened\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "\n",
    "file2.select(dayofweek(\"Incident_DateTime\")).show(7)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file2.select(date_format(\"Incident_DateTime\",'E')).show(7)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Adding a new column in our dataframe,which add the day of the week in each record\n",
    "file3 =file2.withColumn('dayOfTheWeek' , date_format(\"Incident_DateTime\",'E'))\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# Aggregating based on the day of the week -- this will get us the day of the week, on which maximum incidents happened \n",
    "file3.groupBy('dayOfTheWeek').count().orderBy('count', ascending = False).show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#################  Analysis 2    ####################\n",
    "\n",
    "# What percent of the incidents has been recorded online\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    " file2.select(\"File_Online\").show()\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file3 = file2.withColumn(\"File_Online\" ,when(col(\"File_Online\") == True , True).otherwise(False) )\n",
    "file3.select('File_Online').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file4 = file3.select(\"File_Online\").groupBy('File_Online').count()\n",
    "\n",
    "file4.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "from pyspark.sql.window import Window\n",
    "\n",
    "file4.withColumn( 'colnew' ,col('count') / sum('count').over(Window.partitionBy())).show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "\n",
    "############   Analysis 3 ###################\n",
    "\n",
    "# Group by the numbers of incidents reported based on each Year\n",
    "\n",
    "incidents_reporter_per_year = file2.select(year('Incident_DateTime')).groupBy('year(Incident_DateTime)').count()\n",
    "\n",
    "incidents_reporter_per_year.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "###### Running SQL queries in spark ################"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "### creating a temporary table ######\n",
    "file2.registerTempTable(\"police_report_data\")\n",
    "                        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "spark.sql(\"select * from police_report_data\").show(3)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "## Finding the number of incidents of for each incident_category\n",
    "spark.sql(\"select Incident_Category , count(Incident_Category) from police_report_data group by  Incident_Category\").show(52, False)\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "####   Analysis 2 ##############\n",
    "# Find the days of the week on which maximum incidents has happened\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "file2.withColumn('dayOfTheWeek' , date_format(\"Incident_DateTime\",'E')).registerTempTable(\"police_report_data_with_day\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "spark.sql('select dayOfTheWeek from police_report_data_with_day ').show(12, False)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "spark.sql('select dayOfTheWeek , count(dayOfTheWeek) from police_report_data_with_day group by dayOfTheWeek order by count(dayOfTheWeek) desc ').show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#################  Analysis 3    ####################\n",
    "\n",
    "# What percent of the incidents has been recorded online\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "# file2.select(\"File_Online\").show()\n",
    "spark.sql(\"select File_Online from police_report_data\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "#file3 = file2.withColumn(\"File_Online\" ,when(col(\"File_Online\") == True , True).otherwise(False) )\n",
    "#update users set name = '*' where name is null\n",
    "spark.sql(\"select  ((count(*) - count(File_Online))/count(*))*100 as offline_percent  , \\\n",
    "(100 - ((count(*) - count(File_Online))/count(*))*100) as online_percent from police_report_data \").\\\n",
    "show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "############   Analysis 3 ###################\n",
    "\n",
    "# Group by the numbers of incidents reported based on each Year\n",
    "\n",
    "spark.sql(\"select  year(Incident_DateTime) as year ,  count(*) as no_incidents from police_report_data group by \\\n",
    "year(Incident_DateTime)\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "########### How many cases of Assault happened on particular month say in Jan 2020  ############\n",
    "\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": [
    "spark.sql(\"select  year(Incident_DateTime) as year , month (Incident_DateTime) as month,  count(*) \\\n",
    "as no_incidents from police_report_data where year(Incident_DateTime)= 2020 and month (Incident_DateTime) = 3 group by year, month \").show()\n",
    "\n",
    "\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true
   },
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}