Lab 4 - Spark Mlib, Twitter streaming word count

Author: Ruthvicp

Lab/Lab4/Source/Absenteeism_at_work.csv

@@ -0,0 +1,77 @@
+from pyspark.ml import Pipeline
+from pyspark.ml.classification import DecisionTreeClassifier
+from pyspark.ml.feature import StringIndexer, VectorIndexer, VectorAssembler
+from pyspark.ml.evaluation import MulticlassClassificationEvaluator
+
+from sklearn.metrics import confusion_matrix
+from sklearn.metrics import precision_score
+from sklearn.metrics import recall_score
+import scipy
+import os
+os.environ["SPARK_HOME"] = "C:\\spark-2.3.1-bin-hadoop2.7\\spark-2.3.1-bin-hadoop2.7"
+os.environ["HADOOP_HOME"] = "C:\\winutils\\"
+
+from pyspark.python.pyspark.shell import spark
+
+data = spark.read.load("Absenteeism_at_work.csv", format="csv", header=True, delimiter=",")
+data = data.withColumn("MOA", data["Month of absence"] - 0).withColumn("label", data['Height'] - 0). \
+    withColumn("ROA", data["Reason for absence"] - 0). \
+    withColumn("distance", data["Distance from Residence to Work"] - 0). \
+    withColumn("BMI", data["Body mass index"] - 0)
+#data.show()
+
+assem = VectorAssembler(inputCols=["label", "distance"], outputCol='features')
+data = assem.transform(data)
+
+# Index labels, adding metadata to the label column.
+# Fit on whole dataset to include all labels in index.
+labelIndexer = StringIndexer(inputCol="label", outputCol="indexedLabel").fit(data)
+# Automatically identify categorical features, and index them.
+# We specify maxCategories so features with > 4 distinct values are treated as continuous.
+featureIndexer =\
+    VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)
+
+# Split the data into training and test sets (30% held out for testing)
+(trainingData, testData) = data.randomSplit([0.7, 0.3])
+
+# Train a DecisionTree model.
+dt = DecisionTreeClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures")
+
+# Chain indexers and tree in a Pipeline
+pipeline = Pipeline(stages=[labelIndexer, featureIndexer, dt])
+
+# Train model.  This also runs the indexers.
+model = pipeline.fit(trainingData)
+
+# Make predictions.
+predictions = model.transform(testData)
+
+# Select example rows to display.
+predictions.select("prediction", "indexedLabel", "features").show(5)
+
+# Select (prediction, true label) and compute test error
+evaluator = MulticlassClassificationEvaluator(
+    labelCol="indexedLabel", predictionCol="prediction", metricName="accuracy")
+
+accuracy = evaluator.evaluate(predictions)
+
+y_true = data.select("BMI").rdd.flatMap(lambda x: x).collect()
+y_pred = data.select("ROA").rdd.flatMap(lambda x: x).collect()
+
+confusionmatrix = confusion_matrix(y_true, y_pred)
+
+precision = precision_score(y_true, y_pred, average='micro')
+
+recall = recall_score(y_true, y_pred, average='micro')
+
+treeModel = model.stages[2]
+# summary only
+print(treeModel)
+print("Decision Tree - Test Accuracy = %g" % (accuracy))
+print("Decision Tree - Test Error = %g" % (1.0 - accuracy))
+
+print("The Confusion Matrix for Decision Tree Model is :\n" + str(confusionmatrix))
+
+print("The precision score for Decision Tree Model is: " + str(precision))
+
+print("The recall score for Decision Tree Model is: " + str(recall))

Lab/Lab4/Source/DecisionTree.py

@@ -0,0 +1,65 @@
+import os
+import numpy as np
+from pyspark.ml.feature import VectorAssembler
+from sklearn.metrics import confusion_matrix
+from sklearn.metrics import precision_score
+from sklearn.metrics import recall_score
+import scipy
+os.environ["SPARK_HOME"] = "C:\\spark-2.3.1-bin-hadoop2.7\\spark-2.3.1-bin-hadoop2.7"
+os.environ["HADOOP_HOME"] = "C:\\winutils\\"
+from pyspark.ml.classification import NaiveBayes
+from pyspark.ml.evaluation import MulticlassClassificationEvaluator
+
+# import numpy
+# Load training data
+from pyspark.ml.linalg import SparseVector
+# from pyspark.python.pyspark.shell import spark
+from pyspark.sql import SparkSession
+spark = SparkSession.builder.getOrCreate()
+data = spark.read.load("Absenteeism_at_work.csv", format="csv", header=True, delimiter=",")
+data = data.withColumn("MOA", data["Month of absence"] - 0).withColumn("label", data['Seasons'] - 0). \
+    withColumn("ROA", data["Reason for absence"] - 0). \
+    withColumn("distance", data["Distance from Residence to Work"] - 0). \
+    withColumn("BMI", data["Body mass index"] - 0)
+
+assem = VectorAssembler(inputCols=["label", "MOA"], outputCol='features')
+
+data = assem.transform(data)
+# Split the data into train and test
+splits = data.randomSplit([0.7, 0.3], 1000)
+train = splits[0]
+test = splits[1]
+
+# create the trainer and set its parameters
+nb = NaiveBayes(smoothing=1.0, modelType="multinomial")
+
+# train the model
+model = nb.fit(train)
+
+# select example rows to display.
+predictions = model.transform(test)
+
+# compute accuracy on the test set
+evaluator = MulticlassClassificationEvaluator(labelCol="label", predictionCol="prediction",
+                                              metricName="accuracy")
+
+y_true = data.select("BMI").rdd.flatMap(lambda x: x).collect()
+y_pred = data.select("ROA").rdd.flatMap(lambda x: x).collect()
+
+
+accuracy = evaluator.evaluate(predictions)
+
+confusionmatrix = confusion_matrix(y_true, y_pred)
+
+precision = precision_score(y_true, y_pred, average='micro')
+
+recall = recall_score(y_true, y_pred, average='micro')
+
+
+print("Naive Bayes - Test set accuracy = " + str(accuracy))
+
+print("The Confusion Matrix for Naive Bayes Model is :\n" + str(confusionmatrix))
+
+print("The precision score for Naive Bayes Model is: " + str(precision))
+
+print("The recall score for Naive Bayes Model is: " + str(recall))

Lab/Lab4/Source/NaiveBayes.py

@@ -0,0 +1,82 @@
+from pyspark.ml import Pipeline
+from pyspark.ml.classification import RandomForestClassifier
+from pyspark.ml.feature import IndexToString, StringIndexer, VectorIndexer, VectorAssembler
+from pyspark.ml.evaluation import MulticlassClassificationEvaluator
+
+from sklearn.metrics import confusion_matrix
+from sklearn.metrics import precision_score
+from sklearn.metrics import recall_score
+import os
+os.environ["SPARK_HOME"] = "C:\\spark-2.3.1-bin-hadoop2.7\\spark-2.3.1-bin-hadoop2.7"
+os.environ["HADOOP_HOME"] = "C:\\winutils\\"
+# Load and parse the data file, converting it to a DataFrame.
+from pyspark.python.pyspark.shell import spark
+
+data = spark.read.load("Absenteeism_at_work.csv", format="csv", header=True, delimiter=",")
+
+data = data.withColumn("MOA", data["Month of absence"] - 0).withColumn("label", data['Transportation expense'] - 0). \
+    withColumn("ROA", data["Reason for absence"] - 0). \
+    withColumn("distance", data["Distance from Residence to Work"] - 0). \
+    withColumn("BMI", data["Body mass index"] - 0)
+
+# Index labels, adding metadata to the label column.
+# Fit on whole dataset to include all labels in index.
+
+assem = VectorAssembler(inputCols=["label", "distance"], outputCol='features')
+
+data = assem.transform(data)
+
+labelIndexer = StringIndexer(inputCol="label", outputCol="indexedLabel").fit(data)
+
+# Automatically identify categorical features, and index them.
+# Set maxCategories so features with > 4 distinct values are treated as continuous.
+featureIndexer =\
+    VectorIndexer(inputCol="features", outputCol="indexedFeatures", maxCategories=4).fit(data)
+
+# Split the data into training and test sets (30% held out for testing)
+(trainingData, testData) = data.randomSplit([0.7, 0.3])
+
+# Train a RandomForest model.
+rf = RandomForestClassifier(labelCol="indexedLabel", featuresCol="indexedFeatures", numTrees=10)
+
+# Convert indexed labels back to original labels.
+labelConverter = IndexToString(inputCol="prediction", outputCol="predictedLabel",
+                               labels=labelIndexer.labels)
+
+y_true = data.select("BMI").rdd.flatMap(lambda x: x).collect()
+y_pred = data.select("ROA").rdd.flatMap(lambda x: x).collect()
+
+# Chain indexers and forest in a Pipeline
+pipeline = Pipeline(stages=[labelIndexer, featureIndexer, rf, labelConverter])
+
+# Train model.  This also runs the indexers.
+model = pipeline.fit(trainingData)
+
+# Make predictions.
+predictions = model.transform(testData)
+
+# Select example rows to display.
+predictions.select("predictedLabel", "label", "features").show(5)
+
+# Select (prediction, true label) and compute test error
+evaluator = MulticlassClassificationEvaluator(
+    labelCol="indexedLabel", predictionCol="prediction", metricName="accuracy")
+
+accuracy = evaluator.evaluate(predictions)
+
+confusionmatrix = confusion_matrix(y_true, y_pred)
+
+precision = precision_score(y_true, y_pred, average='micro')
+
+recall = recall_score(y_true, y_pred, average='micro')
+
+rfModel = model.stages[2]
+print(rfModel)  # summary only
+print("Random Forest - Test Accuracy = %g" % (accuracy))
+print("Random Forest - Test Error = %g" % (1.0 - accuracy))
+
+print("The Confusion Matrix for Random Forest Model is :\n" + str(confusionmatrix))
+
+print("The precision score for Random Forest Model is: " + str(precision))
+
+print("The recall score for Random Forest Model is: " + str(recall))

Lab/Lab4/Source/RandomForest.py

@@ -0,0 +1,25 @@
+from pyspark import SparkContext
+from pyspark.streaming import StreamingContext
+from pyspark.sql.functions import desc
+from collections import namedtuple
+import os
+os.environ["SPARK_HOME"] = "C:\\spark-2.3.1-bin-hadoop2.7\\spark-2.3.1-bin-hadoop2.7"
+os.environ["HADOOP_HOME"] = "C:\\winutils\\"
+
+def main():
+    sc = SparkContext(appName="PysparkStreaming")
+    wordcount = {}
+    ssc = StreamingContext(sc, 5)
+    lines = ssc.socketTextStream("localhost", 1234)
+    fields = ("word", "count")
+    Tweet = namedtuple('Text', fields)
+    # lines = socket_stream.window(20)
+    counts = lines.flatMap(lambda text: text.split(" "))\
+        .map(lambda x: (x, 1))\
+        .reduceByKey(lambda a, b: a + b).map(lambda rec: Tweet(rec[0], rec[1]))
+    counts.pprint()
+    ssc.start()
+    ssc.awaitTermination()
+
+if __name__ == "__main__":
+    main()

Lab/Lab4/Source/TSWordCount.py

@@ -0,0 +1,56 @@
+import tweepy
+from tweepy import OAuthHandler
+from tweepy import Stream
+from tweepy.streaming import StreamListener
+import socket
+import json
+import time
+
+consumer_key = 'mSg1xMRSz8a8EqPY1sHcE0FPo'
+consumer_secret = 'VfalDVqLVhJ4CQb15Qb7YTx2hgkfPtpO6ogm6Vz07iIR7TqAaN'
+access_token = '3240653721-pmylXVMr6dX32S0omzX4hptZI2a2Aj79Popop7P'
+access_secret = 'DR8D9bYt2aZeh9jt8pugStIMMipZ9muS2oDsRT34MKxw3'
+
+
+auth = OAuthHandler(consumer_key, consumer_secret)
+auth.set_access_token(access_token, access_secret)
+
+class TweetsListener(StreamListener):
+
+    def __init__(self, csocket):
+        self.client_socket = csocket
+
+    def on_data(self, data):
+        try:
+            msg = json.loads(data)
+            print(msg['text'].encode('utf-8'))
+            self.client_socket.send(msg['text'].encode('utf-8'))
+            return True
+        except BaseException as e:
+            print("Error on_data: %s" % str(e))
+        return True
+
+    def on_error(self, status):
+        print(status)
+        return True
+
+
+def sendData(c_socket):
+    auth = OAuthHandler(consumer_key, consumer_secret)
+    auth.set_access_token(access_token, access_secret)
+
+    twitter_stream = Stream(auth, TweetsListener(c_socket))
+    twitter_stream.filter(track=['fifa'])
+
+
+if __name__ == "__main__":
+    s = socket.socket()  # Create a socket object
+    host = "localhost"  # Get local machine name
+    port = 1234  # Reserve a port for your service.
+    s.bind((host, port))  # Bind to the port
+    print("Listening on port: %s" % str(port))
+    s.listen(5)  # Now wait for client connection.
+    c, addr = s.accept()  # Establish connection with client.
+    print("Received request from: " + str(addr))
+    time.sleep(5)
+    sendData(c)

Lab/Lab4/Source/TwitterListener.py

@@ -0,0 +1,13 @@
+----------------------------------------------------------------
+Fri Jul 27 12:03:24 CDT 2018:
+Booting Derby version The Apache Software Foundation - Apache Derby - 10.12.1.1 - (1704137): instance a816c00e-0164-dcb1-e9dd-00000d7b2d68 
+on database directory C:\Users\ruthv\PycharmProjects\CS5590_BigDataProgramming\Spark_Python_Code\LAB4\New\metastore_db with class loader org.apache.spark.sql.hive.client.IsolatedClientLoader$$anon$1@74ed851c 
+Loaded from file:/C:/spark-2.3.1-bin-hadoop2.7/spark-2.3.1-bin-hadoop2.7/jars/derby-10.12.1.1.jar
+java.vendor=Oracle Corporation
+java.runtime.version=1.8.0_162-b12
+user.dir=C:\Users\ruthv\PycharmProjects\CS5590_BigDataProgramming\Spark_Python_Code\LAB4\New
+os.name=Windows 10
+os.arch=amd64
+os.version=10.0
+derby.system.home=null
+Database Class Loader started - derby.database.classpath=''

Lab/Lab4/Source/derby.log

@@ -0,0 +1,9 @@
+
+# *************************************************************************
+# ***              DO NOT TOUCH FILES IN THIS DIRECTORY!                ***
+# *** FILES IN THIS DIRECTORY AND SUBDIRECTORIES CONSTITUTE A DERBY     ***
+# *** DATABASE, WHICH INCLUDES THE DATA (USER AND SYSTEM) AND THE       ***
+# *** FILES NECESSARY FOR DATABASE RECOVERY.                            ***
+# *** EDITING, ADDING, OR DELETING ANY OF THESE FILES MAY CAUSE DATA    ***
+# *** CORRUPTION AND LEAVE THE DATABASE IN A NON-RECOVERABLE STATE.     ***
+# *************************************************************************

Lab/Lab4/Source/metastore_db/README_DO_NOT_TOUCH_FILES.txt

@@ -0,0 +1,8 @@
+
+# *************************************************************************
+# ***              DO NOT TOUCH FILES IN THIS DIRECTORY!                ***
+# *** FILES IN THIS DIRECTORY ARE USED BY THE DERBY DATABASE RECOVERY   ***
+# *** SYSTEM. EDITING, ADDING, OR DELETING FILES IN THIS DIRECTORY      ***
+# *** WILL CAUSE THE DERBY RECOVERY SYSTEM TO FAIL, LEADING TO          ***
+# *** NON-RECOVERABLE CORRUPT DATABASES.                                ***
+# *************************************************************************

Lab/Lab4/Source/metastore_db/db.lck

@@ -0,0 +1,8 @@
+
+# *************************************************************************
+# ***              DO NOT TOUCH FILES IN THIS DIRECTORY!                ***
+# *** FILES IN THIS DIRECTORY ARE USED BY THE DERBY DATABASE TO STORE   *** 
+# *** USER AND SYSTEM DATA. EDITING, ADDING, OR DELETING FILES IN THIS  ***
+# *** DIRECTORY WILL CORRUPT THE ASSOCIATED DERBY DATABASE AND MAKE     ***
+# *** IT NON-RECOVERABLE.                                               ***
+# *************************************************************************

Lab/Lab4/Source/metastore_db/log/README_DO_NOT_TOUCH_FILES.txt

@@ -0,0 +1,23 @@
+#C:\Users\ruthv\PycharmProjects\CS5590_BigDataProgramming\Spark_Python_Code\LAB4\New\metastore_db
+# ********************************************************************
+# ***                Please do NOT edit this file.                 ***
+# *** CHANGING THE CONTENT OF THIS FILE MAY CAUSE DATA CORRUPTION. ***
+# ********************************************************************
+#Fri Jul 27 08:44:04 CDT 2018
+SysschemasIndex2Identifier=225
+SyscolumnsIdentifier=144
+SysconglomeratesIndex1Identifier=49
+SysconglomeratesIdentifier=32
+SyscolumnsIndex2Identifier=177
+SysschemasIndex1Identifier=209
+SysconglomeratesIndex3Identifier=81
+SystablesIndex2Identifier=129
+SyscolumnsIndex1Identifier=161
+derby.serviceProtocol=org.apache.derby.database.Database
+SysschemasIdentifier=192
+derby.storage.propertiesId=16
+SysconglomeratesIndex2Identifier=65
+derby.serviceLocale=en_US
+SystablesIdentifier=96
+SystablesIndex1Identifier=113
+#--- last line, don't put anything after this line ---