-
Notifications
You must be signed in to change notification settings - Fork 0
/
pyspark-sql.py
62 lines (48 loc) · 2.1 KB
/
pyspark-sql.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
# cevheri
# 2020
# NYC Open Data
# Fire Incident Dispatch Data
# The Fire Incident Dispatch Data file contains data that is generated by the
# Starfire Computer Aided Dispatch System. The data spans from the time
# the incident is created in the system to the time the incident is closed in the system.
# It covers information about the incident as it relates to the assignment
# of resources and the Fire Department’s response to the emergency.
# To protect personal identifying information in accordance with the Health Insurance Portability
# and Accountability Act (HIPAA), specific locations of incidents are not included
# and have been aggregated to a higher level of detail.
from pyspark.sql import SparkSession
from pyspark.sql import Row
import sys
import os
os.environ['SPARK_HOME'] = "/home/cevher/apps/spark-3.0.1-bin-hadoop2.7"
# os.environ['HADOOP_HOME'] = "/home/cevher/app/spark-3.0.1-bin-hadoop2.7/hadoop"
sys.path.append("/home/cevher/apps/spark-3.0.1-bin-hadoop2.7/python")
sys.path.append("/home/cevher/apps/spark-3.0.1-bin-hadoop2.7/python/lib")
if __name__ == '__main__':
spark = SparkSession.builder.appName("sql").getOrCreate()
def parse_line(lines):
fields = lines.split(',')
return Row(
indate=fields[1][:10],
incls=fields[15]
)
def remove_header(lines):
header = lines.first() # extract header
return lines.filter(lambda row: row != header) # filter out header
lines = spark.sparkContext.textFile("Fire_Incident_Dispatch_Data.csv") # all data
data = remove_header(lines)
rows = data.map(parse_line)
schema = spark.createDataFrame(rows).cache()
schema.createOrReplaceTempView("fires")
schema.show(2)
rows = spark.sql("SELECT incls, indate, count(*) "
"FROM fires "
"group by incls, indate"
)
# date example
spark.sql("""
select from_unixtime(unix_timestamp('08/26/2016', 'MM/dd/yyyy'), 'yyyy:MM:dd') as new_format
""").show()
for row in rows:
print(row)
schema.groupBy("incls", "indate").count().show()