<jupyter_start><jupyter_text>This notebook goes through an Apache Spark tutorial to load and wrangle data from San Francisco's open datasetsThis example will particularly utilize the "fire department calls for service" dataset containing over 4.5 million rows starting from the year 2000 all the way upto 2020. the dataset can be downloaded from the following link:https://data.sfgov.org/Public-Safety/Fire-Department-Calls-for-Service/nuek-vuh3<jupyter_code>sc
# Imports
import spark
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, BooleanType
from pyspark.sql.functions import *
# Setting schema
firecalls_schema = StructType([StructField('CallNumber', IntegerType(), True),
                               StructField('UnitID', StringType(), True),
                               StructField('IncidentNumber', IntegerType(), True),
                               StructField('CallType', StringType(), True),
                               StructField('CallDate', StringType(), True),
                               StructField('WatchDate', StringType(), True),
                               StructField('ReceivedDtTm', StringType(), True),
                               StructField('EntryDtTm', StringType(), True),
                               StructField('DispatchDtTm', StringType(), True),
                               StructField('ResponseDtTm', StringType(), True),
                               StructField('OnSceneDtTm', StringType(), True),
                               StructField('TransportDtTm', StringType(), True),
                               StructField('HospitalDtTm', StringType(), True),
                               StructField('CallFinalDisposition', StringType(), True),
                               StructField('AvailableDtTm', StringType(), True),
                               StructField('Address', StringType(), True),
                               StructField('City', StringType(), True),
                               StructField('ZipcodeofIncident', IntegerType(), True),
                               StructField('Battalion', StringType(), True),
                               StructField('StationArea', StringType(), True),
                               StructField('Box', StringType(), True),
                               StructField('OriginalPriority', StringType(), True),
                               StructField('Priority', StringType(), True),
                               StructField('FinalPriority', IntegerType(), True),
                               StructField('ALSUnit', BooleanType(), True),
                               StructField('CallTypeGroup', StringType(), True),
                               StructField('NumberofAlarms', IntegerType(), True),
                               StructField('UnitType', StringType(), True),
                               StructField('Unitsequenceincalldispatch', IntegerType(), True),
                               StructField('FirePreventionDistrict', StringType(), True),
                               StructField('SupervisorDistrict', StringType(), True),
                               StructField('NeighborhoodDistrict', StringType(), True),
                               StructField('Location', StringType(), True),
                               StructField('RowID', StringType(), True),
    
]
)
# Creating Spark Session and reading CSV file
spark = SparkSession \
    .builder \
    .appName("Spark SFFireDepartmentCalls") \
    .getOrCreate()

df = spark.read.csv("Fire_Department_Calls_for_Service.csv", header=True, schema=firecalls_schema);
# View dataframe
df.show()
# Printing all columns in the dataframe
df.columns
# Getting tope 35 distinct call types in the dataframe
# 'False' argument expands column width to include the full text
df.select('CallType').distinct().show(35, False)
# Applying Spark transformations to get calls by highest count 
df.select('CallType').groupBy('CallType').count().orderBy('count', ascending=False).show()
df.printSchema()
# Spark uses Java simple date format
# Converting 'Date' and 'DateTime' columns from String to DateTime format

from_pattern1 = 'MM/dd/yyyy'
to_pattern1 = 'yyyy-MM-dd'

from_pattern2 = 'MM/dd/yyyy hh:mm:ss aa'
to_pattern2 = 'MM/dd/yyyy hh:mm:ss aa'

df = df\
    .withColumn('CallDateTS', unix_timestamp(df['CallDate'], from_pattern1).cast('timestamp'))\
    .drop('CallDate')\
    .withColumn('WatchDateTS', unix_timestamp(df['WatchDate'], from_pattern1).cast('timestamp'))\
    .drop('WatchDate')\
    .withColumn('ReceivedDtTmTS', unix_timestamp(df['ReceivedDtTm'], from_pattern2).cast('timestamp'))\
    .drop('ReceivedDtTm')\
    .withColumn('EntryDtTmTS', unix_timestamp(df['EntryDtTm'], from_pattern2).cast('timestamp'))\
    .drop('EntryDtTm')\
    .withColumn('DispatchDtTmTS', unix_timestamp(df['DispatchDtTm'], from_pattern2).cast('timestamp'))\
    .drop('DispatchDtTm')\
    .withColumn('ResponseDtTmTS', unix_timestamp(df['ResponseDtTm'], from_pattern2).cast('timestamp'))\
    .drop('ResponseDtTm')\
    .withColumn('OnSceneDtTmTS', unix_timestamp(df['OnSceneDtTm'], from_pattern2).cast('timestamp'))\
    .drop('OnSceneDtTm')\
    .withColumn('TransportDtTmTS', unix_timestamp(df['TransportDtTm'], from_pattern2).cast('timestamp'))\
    .drop('TransportDtTm')\
    .withColumn('HospitalDtTmTS', unix_timestamp(df['HospitalDtTm'], from_pattern2).cast('timestamp'))\
    .drop('HospitalDtTm')\
    .withColumn('AvailableDtTmTS', unix_timestamp(df['AvailableDtTm'], from_pattern2).cast('timestamp'))\
    .drop('AvailableDtTm')\

df.printSchema()
# Getting all the distinct years in the dataframe
df.select(year('CallDateTS')).distinct().orderBy('year(CallDateTS)').show()
# Filtering down days to the first ten days in the July of 2019
df.filter(year('CallDateTS') == '2019')\
.filter(dayofyear('CallDateTS') >= 180)\
.filter(dayofyear('CallDateTS') <= 190)\
.select(dayofyear('CallDateTS'))\
.distinct()\
.orderBy('dayofyear(CallDateTS)')\
.show()
# Coounting number of calls per day in July 2019
df.filter(year('CallDateTS') == '2019').filter(dayofyear('CallDateTS') >= 180)\
.filter(dayofyear('CallDateTS') <= 190)\
.groupBy(dayofyear('CallDateTS'))\
.count()\
.orderBy('dayofyear(CallDateTS)')\
.show()
display(df.filter(year('CallDateTS') == '2019').filter(dayofyear('CallDateTS') >= 180)\
.filter(dayofyear('CallDateTS') <= 190)\
.groupBy(dayofyear('CallDateTS'))\
.count()\
.orderBy('dayofyear(CallDateTS)'))
df.rdd.getNumPartitions()
df.repartition(6).createOrReplaceTempView("fireCallsView")<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13775355737208597, 'fraction_numerical': 0.010293672419013019, 'mean_word_length': 2.951555023923445, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 8, 'https://': 1, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '1', 'dir': 'jupyter-structured-clean-dedup', 'id': '244230', 'license': 'MIT', 'n_tokens_mistral': 1901, 'n_tokens_neox': 1808, 'n_words': 367, 'path': 'Spark_SF_fire_test (1).ipynb', 'repo_name': 'amrithravindra/Spark_Tutorial_SFOpenDataset', 'size': '6606'}