Some updates.

Author: xuwenyihust

README.md

@@ -115,9 +115,6 @@ python3.4 $Visor_HOME/src/socket/fake_log_stream.py -m access
 python3.4 $Visor_HOME/src/socket/fake_log_stream.py -m error
 ```
 
-## Requirements
-* Python 3.4
-
 ## Resources
 * [Apache Log Files](https://httpd.apache.org/docs/1.3/logs.html)
 

requirements.txt

@@ -1,3 +1,4 @@
 pymongo==2.6.3
 numpy==1.8.2
 matplotlib==1.5.1
+kafka-python==1.3.2

src/fake_log_gen/fake_log_gen.py

@@ -1,6 +1,6 @@
 # Fake log files generator
 
-
+import os
 import random
 import json
 import logging
@@ -181,7 +181,7 @@ def main():
 	'''
 
 	# Load the configure json file to a dict
-	with open("../config/fake_log_gen.json") as config_file:
+	with open(os.environ['VISORHOME']+"/config/fake_log_gen.json") as config_file:
 		config = json.load(config_file)
 
 	# Instantiate a fake log generator

src/fake_log_gen/fake_log_producer.py

@@ -3,6 +3,7 @@
 from kafka import KafkaProducer
 from kafka.errors import KafkaError
 
+import os
 import random
 import json
 import logging
@@ -128,7 +129,7 @@ def main():
 	log.addHandler(out)
 
 	# Load the configure json file to a dict
-	with open("../../config/fake_log_gen.json") as config_file:
+	with open(os.environ['VISORHOME']+"/config/fake_log_gen.json") as config_file:
 		config = json.load(config_file)
 
 	# Instantiate a fake log generator

src/fake_log_gen/fake_log_stream.py

@@ -1,6 +1,7 @@
 from src.fake_log_gen import fake_log_gen
 import socket
 
+import os
 import random
 import json
 import logging
@@ -163,7 +164,7 @@ def main():
 	log.addHandler(out)
 
 	# Load the configure json file to a dict
-	with open("../../config/fake_log_gen.json") as config_file:
+	with open(os.environ['VISORHOME']+"/config/fake_log_gen.json") as config_file:
 		config = json.load(config_file)
 
 	# Instantiate a fake log generator

src/kafka_monitor/kafka_monitor.py

@@ -2,42 +2,29 @@
 from pyspark.streaming import StreamingContext
 from pyspark.streaming.kafka import KafkaUtils
 
-#def main():
 
+class kafka_monitor(object):
 
+	def __init__(self):
+		# Define Spark configuration
+		conf = SparkConf()
+		conf.setMaster("local[4]")
+		conf.setAppName("Kafka Monitor")
+		# Initialize a SparkContext
+		sc = SparkContext(conf=conf)
+		self.ssc = StreamingContext(sc, 10)
 
-if __name__=="__main__":
-	# Define Spark configuration
-	conf = SparkConf()
-	conf.setMaster("local[4]")
-	conf.setAppName("Stream Analysis")
-	# Initialize a SparkContext
-	sc = SparkContext(conf=conf)
-
-	#batch_interval = 10
-	#window_time = 10
-	#process_times = 1
-
-	# Compute the whole time for displaying
-	#total_time = batch_interval * process_times
-
-	#main(sc)
-
-	ssc = StreamingContext(sc, 10)
-
-	#zkQuorum, topic = sys.argv[1:]
-	zkQuorum = 'localhost:9092'
-	topic = 'TutorialTopic'
-	kvs = KafkaUtils.createDirectStream(ssc, [topic], {"metadata.broker.list": 'localhost:9092'})
-	kvs.foreachRDD(lambda x: print(x.collect()))
-	#lines = kvs.map(lambda x: x[1])
-	#counts = lines.flatMap(lambda line: line.split(" ")).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a+b)
-	#counts.pprint()
-	#kvs.pprint()
-
-	ssc.start()
-	ssc.awaitTermination()	
+		self.addr = 'localhost:9092'
+		self.topic = 'TutorialTopic'	
 
+	def run(self):
+		lines = KafkaUtils.createDirectStream(self.ssc, [self.topic], {"metadata.broker.list": self.addr})
+		lines.foreachRDD(lambda x: print(x.collect()))
 
+		self.ssc.start()
+		self.ssc.awaitTermination()
 
+if __name__=="__main__":
+	monitor = kafka_monitor()
+	monitor.run()