import findspark
findspark.init()
from pyspark.sql import SparkSession
spark = SparkSession \
    .builder \
    .appName("Analyzing London crime data") \
    .getOrCreate()
data = spark.read\
            .format("csv")\
            .option("header", "true")\
            .load("../datasets/london_crime_by_lsoa.csv")
data.printSchema()
data.count()
data.limit(5).show()
data.dropna()
data = data.drop("lsoa_code")
data.show(5)
total_boroughs = data.select('borough')\
                     .distinct()\
        
total_boroughs.show()
total_boroughs.count()
hackney_data = data.filter(data['borough'] == "Hackney")
hackney_data.show(5)
data_2015_2016 = data.filter(data['year'].isin(["2015", "2016"]))
data_2015_2016.sample(fraction=0.1).show()
data_2014_onwards = data.filter(data['year'] >= 2014 )
data_2014_onwards.sample(fraction=0.1).show()
borough_crime_count = data.groupBy('borough')\
                          .count()
    
borough_crime_count.show(5)
borough_conviction_sum = data.groupBy('borough')\
                             .agg({"value":"sum"})
borough_conviction_sum.show(5)
borough_conviction_sum = data.groupBy('borough')\
                             .agg({"value":"sum"})\
                             .withColumnRenamed("sum(value)","convictions")
borough_conviction_sum.show(5)
Total convictions
total_borough_convictions = borough_conviction_sum.agg({"convictions":"sum"})
total_borough_convictions.show()
Extracting total convictions into a variable
total_convictions = total_borough_convictions.collect()[0][0]
total_convictions
A new column which contains the % convictions for each borough
import pyspark.sql.functions as func
borough_percentage_contribution = borough_conviction_sum.withColumn(
    "% contribution",
    func.round(borough_conviction_sum.convictions / total_convictions * 100, 2))
borough_percentage_contribution.printSchema()
borough_percentage_contribution.orderBy(borough_percentage_contribution[2].desc())\
                               .show(10)
conviction_monthly = data.filter(data['year'] == 2014)\
                         .groupBy('month')\
                         .agg({"value":"sum"})\
                         .withColumnRenamed("sum(value)","convictions")
total_conviction_monthly = conviction_monthly.agg({"convictions":"sum"})\
                                             .collect()[0][0]
total_conviction_monthly
total_conviction_monthly = conviction_monthly\
    .withColumn("percent",
                func.round(conviction_monthly.convictions/total_conviction_monthly * 100, 2))
total_conviction_monthly.columns
total_conviction_monthly.orderBy(total_conviction_monthly.percent.desc()).show()
crimes_category = data.groupBy('major_category')\
                      .agg({"value":"sum"})\
                      .withColumnRenamed("sum(value)","convictions")
crimes_category.orderBy(crimes_category.convictions.desc()).show()
year_df = data.select('year')
year_df.agg({'year':'min'}).show()
year_df.agg({'year':'max'}).show()
year_df.describe().show()
data.crosstab('borough', 'major_category')\
    .select('borough_major_category', 'Burglary', 'Drugs', 'Fraud or Forgery', 'Robbery')\
    .show()
%matplotlib inline
import matplotlib.pyplot as plt
plt.style.use('ggplot')
def describe_year(year):
    yearly_details = data.filter(data.year == year)\
                         .groupBy('borough')\
                         .agg({'value':'sum'})\
                         .withColumnRenamed("sum(value)","convictions")
    
    borough_list = [x[0] for x in yearly_details.toLocalIterator()]
    convictions_list = [x[1] for x in yearly_details.toLocalIterator()]
  
    plt.figure(figsize=(33, 10)) 
    plt.bar(borough_list, convictions_list)
    
    plt.title('Crime for the year: ' + year, fontsize=30)
    plt.xlabel('Boroughs',fontsize=30)
    plt.ylabel('Convictions', fontsize=30)
    plt.xticks(rotation=90, fontsize=30)
    plt.yticks(fontsize=30)
    plt.autoscale()
    plt.show()
describe_year('2014')