BigDataProject-SSML/trying_with_test.py at main · kuntal0901/BigDataProject-SSML · GitHub

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
import json
from pyspark import SparkContext, SparkConf
from pyspark.streaming import StreamingContext
from pyspark.sql import SQLContext, Row, SparkSession
from pyspark.sql.types import IntegerType, StructType,StructField,StringType
from pyspark.sql.functions import *
import ast
import model
from pyspark.sql.functions import monotonically_increasing_id

Model=model.load_model()

flag=True

sc = SparkContext("local[2]", "crime_data")
fp=open('mappings/addmapp.txt','r')
fp1=open("mappings/pdmapp.txt","r")
fp2=open("mappings/dowmapp.txt","r")
fp3=open("mappings/Catmap.txt","r")
addressmappings=ast.literal_eval(fp.read())
dayofweekmappings=ast.literal_eval(fp2.read())
pddistrictmappings=ast.literal_eval(fp1.read())
categorymappings=ast.literal_eval(fp3.read())
spark = SparkSession.builder.config(conf=SparkConf()).getOrCreate()
ssc = StreamingContext(sc, 1)
sql_context = SQLContext(sc)
schema2 = StructType([
    StructField("ID", StringType(), False),
	StructField("Dates", StringType(), False),
    StructField("DayOfWeek", StringType(), False),
    StructField("PdDistrict", StringType(), False),
    StructField("Address", StringType(), False),
    StructField("X", StringType(), False),
    StructField("Y", StringType(), False)
])
schema3 = StructType([
	StructField("DayOfWeek", StringType(), False),
    StructField("PdDistrict", StringType(), False),
    StructField("X", StringType(), False),
    StructField("Y", StringType(), False),
    StructField("Year", StringType(), False),
    StructField("Month", StringType(), False),
    StructField("Date", StringType(), False),
    StructField("Hours",StringType(), False),
    StructField("Minute",StringType(), False)
])
schema4=StructType([StructField("Category", StringType(), False)])
def preprocessing(df):
   df = df.withColumn('year',year(df.Dates))
   df = df.withColumn('month',month(df.Dates))
   df = df.withColumn('date',dayofmonth(df.Dates))
   df = df.withColumn('hour',hour(df.Dates))
   df = df.withColumn('minute',minute(df.Dates))
   df=df.drop("Descript","Resolution","Dates","ID")
   df = df.select("*").withColumn("ID", monotonically_increasing_id())
   df_X_train=df
   rdd=df_X_train.rdd.map(lambda x: (dayofweekmappings[x["DayOfWeek"]],pddistrictmappings[x["PdDistrict"]],x["X"],x["Y"],x["year"],x["month"],x["date"],x["hour"],x["minute"]))
   df_X_final=spark.createDataFrame(data=rdd,schema=schema3)
   predict=model.test(Model,df_X_final)
   predict_df=spark.createDataFrame(data=predict,schema=schema4)
   predict_df = predict_df.select("*").withColumn("ID", monotonically_increasing_id())
   final_result=df_X_train.join(predict_df,df_X_train.ID==predict_df.ID,"inner")
   final_result=final_result.drop("ID")
   final_result.show()

def convert_to_df(rdd):
	global schema, spark
	records = rdd.collect()
	dicts = [i for j in records for i in list(json.loads(j).values())]
	if len(dicts) == 0:
		return
	df = spark.createDataFrame((Row(**d) for d in dicts), schema2)
	preprocessing(df)
lines = ssc.socketTextStream("localhost",6100)
json_str = lines.flatMap(lambda x: x.split('\n'))
lines.foreachRDD(convert_to_df)
ssc.start()
ssc.awaitTermination(300)