-
Notifications
You must be signed in to change notification settings - Fork 1
/
suicide-analysis.py
127 lines (65 loc) · 7.29 KB
/
suicide-analysis.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
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
from pyspark.sql import SparkSession
spark = SparkSession.builder.enableHiveSupport().getOrCreate()
spark.sparkContext.setLogLevel('ERROR')
spark.sql("use indiaDB")
data = spark.sql("select * from suicides")
print("\n--------------0.Verifying the data from Hive Table--------------")
data.show(20,False)
data.createOrReplaceTempView("suicides")
from pyspark.sql.functions import sum,col,min,max
print("--------------1.List of States in the Dataset--------------")
spark.sql("select distinct(state) as STATES from suicides order by STATES").show(40,False)
print("--------------2.Total Cases according to the Year--------------")
query1 = spark.sql("select type_code, year, sum(case_count) as Total from suicides where group by type_code,year order by year").filter(col("type_code")=="Means_adopted").drop("type_code")
query1.show(40,False)
print("--------------3.Percentage of increase in suicides between 2001 and 2012--------------\n")
maxVal = query1.select(max("Total")).collect()[0][0]
minVal = query1.select(min("Total")).collect()[0][0]
print("Average increase: " + str((float(maxVal)-float(minVal))/float(minVal)*100))
print("\n--------------4.Cases every year gender wise with %share of both gender--------------")
maleData = data.filter((data.type_code == "Means_adopted") & (data.gender == "Male")).groupBy("year","gender").agg(sum("case_count").alias("Male")).orderBy("year").drop("gender")
femaleData = data.filter((data.type_code == "Means_adopted") & (data.gender == "Female")).groupBy("year","gender").agg(sum("case_count").alias("Female")).orderBy("year").drop("gender")
genderData = maleData.join(femaleData,"year")
genderData.withColumn("Male %", col("Male")/(col("Male")+col("Female"))*100).withColumn("Female %" ,col("Female")/(col("Male")+col("Female"))*100).show()
print("--------------5.Causes for suicide in descending order of total of cases--------------")
data.filter(data.type_code == "Causes").groupBy("type").agg(sum("case_count").alias("Total_Cases")).orderBy(col("Total_Cases").desc()).show(50,False)
print("--------------6.Causes for suicide based on age groups in descending order of total cases--------------")
data.filter(data.type_code == "Causes").groupBy("type","age_group").agg(sum("case_count").alias("Total_Cases")).orderBy(col("Total_Cases").desc()).show(200,False)
print("--------------7.Total cases categorised under cause of suicide gender wise--------------")
maleData = data.filter((data.type_code == "Causes") & (data.gender == "Male")).groupBy("type").agg(sum("case_count").alias("Male")).orderBy(col("Male").desc())
femaleData = data.filter((data.type_code == "Causes") & (data.gender == "Female")).groupBy("type").agg(sum("case_count").alias("Female")).orderBy(col("Female").desc())
genderData = maleData.join(femaleData,"type")
genderData.show(50,False)
print("--------------8.Total cases categorised under cause of suicide of specific age group--------------")
data.filter((data.type_code == "Causes") & (data.age_group == "15-29")).groupBy("type").agg(sum("case_count").alias("Age 15-29")).orderBy(col("Age 15-29").desc()).show(20,False)
print("--------------9.Total cases categorised under professional profile of the people who committed suicide--------------")
data.filter(data.type_code == "Professional_Profile").groupBy("type").agg(sum("case_count").alias("Total_Cases")).orderBy(col("Total_Cases").desc()).show(50,False)
print("--------------10.Total cases categorised under method adopted for suicide--------------")
data.filter(data.type_code == "Means_adopted").groupBy("type").agg(sum("case_count").alias("Total_Cases")).orderBy(col("Total_Cases").desc()).show(50,False)
print("--------------11.Total cases categorised under state--------------")
data.filter(data.type_code == "Means_adopted").groupBy("state").agg(sum("case_count").alias("Total_Cases")).orderBy(col("Total_Cases").desc()).show(50,False)
print("--------------12.Cases on the causes of suicide for Maharashtra--------------")
data.filter((data.type_code == "Causes") & (data.state == "Maharashtra")).groupBy("state","type").agg(sum("case_count").alias("Total_Cases")).orderBy(col("Total_Cases").desc()).drop("state").show(50,False)
print("--------------13.Cases on the education status of the people committed suicide--------------")
data.filter(data.type_code == "Education_Status").groupBy("type").agg(sum("case_count").alias("Total_Cases")).orderBy(col("Total_Cases").desc()).show(50,False)
print("\n--------------14.Creating Partitions based on state and type_code--------------\n")
spark.sparkContext.setLogLevel('INFO')
data.write.option("header",True).partitionBy("state","type_code").mode("overwrite").csv("/suicideData")
spark.sparkContext.setLogLevel('ERROR')
print("\n--------------15.Reading data from partition and comparing two states based on cause for suicide--------------")
MHData = spark.read.options(header='True', inferSchema='True').csv("/suicideData/state=Maharashtra/type_code=Causes/").withColumnRenamed("case_count", "MH Cases")
WBData = spark.read.options(header='True', inferSchema='True').csv("/suicideData/state=West Bengal/type_code=Causes/").withColumnRenamed("case_count", "WB Cases")
outData = MHData.join(WBData,['year','type','gender','age_group'])
outData.groupBy('type').agg(sum("MH Cases").alias("Cases in Maharashtra"), sum("WB Cases").alias("Cases in West Bengal")).orderBy(col("Cases in Maharashtra").desc(),col("Cases in West Bengal").desc()).withColumnRenamed("type","Causes for Suicide").show(50,False)
print("--------------16.Reading data from partition and comparing two states based on method adopted for suicide--------------")
WBData = spark.read.options(header='True', inferSchema='True').csv("/suicideData/state=West Bengal/type_code=Means_adopted/").withColumnRenamed("case_count", "WB Cases")
MHData = spark.read.options(header='True', inferSchema='True').csv("/suicideData/state=Maharashtra/type_code=Means_adopted/").withColumnRenamed("case_count", "MH Cases")
outData = MHData.join(WBData,['year','type','gender','age_group'])
outData.groupBy('type').agg(sum("MH Cases").alias("Cases in Maharashtra"), sum("WB Cases").alias("Cases in West Bengal")).orderBy(col("Cases in Maharashtra").desc(),col("Cases in West Bengal").desc()).withColumnRenamed("type","Means Adopted for Suicide").show(50,False)
print("--------------17.Trend of a particular cause between the timeframe along with percentage contribution--------------")
totalValue = data.filter(data.type == "Drug Abuse/Addiction").groupBy("year").agg(sum("case_count").alias("TC")).agg(sum("TC")).collect()[0][0]
data.filter(data.type == "Drug Abuse/Addiction").groupBy("year").agg(sum("case_count").alias("TC")).orderBy("year").withColumn("% Share of Total (" +str(totalValue)+ ")",(col("TC")/totalValue)*100).withColumnRenamed("TC", "Drug Abuse/Addiction Cases").show(50,False)
print("--------------18.Division of suicide cases in various age groups for Students--------------")
data.filter((data.type == "Student")).groupBy("type","age_group").agg(sum("case_count").alias("Student Cases")).orderBy(col("age_group").desc()).drop("type").show(20,False)
print("--------------19.Division of suicide cases in various age groups for Failure in Examination--------------")
data.filter((data.type == "Failure in Examination")).groupBy("type","age_group").agg(sum("case_count").alias("Student Cases")).orderBy(col("age_group").desc()).drop("type").show(20,False)