update streaming process

Author: qiaoliuhub

evaluator.py

@@ -81,10 +81,10 @@ def shutdown_hook(spark_session):
 
 # tokenize post texts and get term frequency and inverted document frequency
 logger.debug("Start to generate TFIDF features")
-tokenizer=Tokenizer.load(tokenizer_file)
+tokenizer=Tokenizer.load(tokenizer_file)h
 tokenized_words=tokenizer.transform(training_df.na.drop(how = 'any'))
 hashing_TF=HashingTF.load(hashing_tf_file)
-TFfeatures=hashing_TF.transform(tokenized_words.na.drop(how = 'any'))
+TFfeatures=hashing_TF.transform(tokenized_words.na.drop(how = 'any'))i
 
 idfModel=IDFModel.load(idf_model_file)
 TFIDFfeatures=idfModel.transform(TFfeatures.na.drop(how = 'any'))
@@ -118,16 +118,3 @@ def shutdown_hook(spark_session):
 
 
 
-
-
-
-
-
-
-
-
-
-
-
-
-

model_generation.cfg

@@ -1,5 +1,5 @@
 [io]
-post_file = /Users/QiaoLiu1/Autotag/dataset/Questions.json
+post_file = /Users/QiaoLiu1/Autotag/test/Questions.json
 tags_file = /Users/QiaoLiu1/Autotag/dataset/Tags.csv
 selected_tags_file = /Users/QiaoLiu1/Autotag/dataset/top100Tags.csv
 idf_model_file = /Users/QiaoLiu1/Autotag/models/idfModel

model_generation.py

@@ -14,6 +14,7 @@
 from pyspark.ml.classification import NaiveBayes, NaiveBayesModel
 from pyspark.ml.evaluation import MulticlassClassificationEvaluator
 from pyspark.sql.functions import udf, col
+from pyspark.sql.types import FloatType, StringType
 
 
 logging.basicConfig()
@@ -85,7 +86,7 @@ def shutdown_hook(spark_session):
 tokenizer=Tokenizer(inputCol="Body", outputCol="Words")
 tokenized_words=tokenizer.transform(training_df.na.drop(how = 'any'))
 tokenizer.save(tokenizer_file)
-hashing_TF=HashingTF(inputCol="Words", outputCol="Features", numFeatures=200000)#, numFeatures=200
+hashing_TF=HashingTF(inputCol="Words", outputCol="Features", numFeatures=200)#, numFeatures=200
 hashing_TF.save(hashing_tf_file)
 TFfeatures=hashing_TF.transform(tokenized_words.na.drop(how = 'any'))
 
@@ -104,11 +105,14 @@ def shutdown_hook(spark_session):
 # Row(IDF_features=SparseVector(200, {7: 2.3773, 9: 2.1588, 32: 2.0067, 37: 1.7143, 49: 2.6727, 59: 2.9361, 114: 1.0654, 145: 2.9522, 167: 2.3751}), Tag=u'asp.net')
 # Trasfer data to be in labeled point format
 
-labeled_points=TFIDFfeatures.rdd.map(lambda row: (float(tags_to_catId.value[row.Tag]), row.IDF_features, row.Id)).toDF()
+tags_to_catId_transform = udf(lambda tag: float(tags_to_catId.value[tag]), FloatType())
+catId_to_tags_transform = udf(lambda catId: catId_to_tags.value[int(catId)], StringType())
+
+labeled_points=TFIDFfeatures.withColumn('CatId', tags_to_catId_transform('Tag')) # rdd.map(lambda row: (float(tags_to_catId.value[row.Tag]), row.IDF_features, row.Id)).toDF()
 training, test=labeled_points.randomSplit([0.7, 0.3], seed=0)
 
 # Train Naive Bayes model
-nb=NaiveBayes(smoothing=1.0, modelType="multinomial", labelCol='_1', featuresCol='_2')
+nb=NaiveBayes(smoothing=1.0, modelType="multinomial", labelCol='TagId', featuresCol='IDF_features')
 nb_model=nb.fit(training)
 nb_model.save(nb_model_file)
 
@@ -117,7 +121,7 @@ def shutdown_hook(spark_session):
 # print test_df.collect()
 
 predictions=nb_model.transform(test)
-evaluator=MulticlassClassificationEvaluator(labelCol="_1", predictionCol="prediction", metricName="accuracy")
+evaluator=MulticlassClassificationEvaluator(labelCol="TagId", predictionCol="prediction", metricName="accuracy")
 accuracy = evaluator.evaluate(predictions)
 print("Test set accuracy = " + str(accuracy/0.6023699978752843))
 
@@ -126,16 +130,3 @@ def shutdown_hook(spark_session):
 
 
 
-
-
-
-
-
-
-
-
-
-
-
-
-

streaming_prediction.cfg

@@ -0,0 +1,13 @@
+[spark]
+master = 
+
+[io]
+tokenizer_file = 
+hashing_tf_file = 
+idf_model_file = 
+nb_model_file 
+
+[kafka]
+broker_ip = 
+kafka_topic = 
+kafka_output_topic = 

streaming_prediction.py

@@ -1,84 +1,175 @@
 import logging
 import ConfigParser
-import pandas as pd
 import atexit
+import json
+import pandas as pd
 
-from pyspark import SparkContext, SparkConf
+from pyspark import SparkContext, SparkConf 
 from pyspark.streaming import StreamingContext
-from pyspark.ml.feature import HashingTF, IDF, Tokenizer, IDFModel
-from pyspark.ml.linalg import Vectors
-from pyspark.ml.classification import NaiveBayes, NaiveBayesModel
-from pyspark.ml.evaluation import MulticlassClassificationEvaluator
-from pyspark.sql.functions import udf, col
+from pyspark.sql import SparkSession
 from kafka import KafkaProducer
 from pyspark.streaming.kafka import KafkaUtils
 from kafka.errors import KafkaError
-
-
-logging.basicConfig()
-logger=logging.getLogger('model_generation')
+from pyspark.ml.feature import Tokenizer, HashingTF, IDFModel
+from pyspark.ml.classification import NaiveBayesModel
+from pyspark.sql.functions import udf
+from pyspark.sql.types import StringType, FloatTypex
+
+#Set up logger
+logging.BasicConfig()
+logger = logging.getLogger('Streaming_prediction')
 logger.setLevel(logging.DEBUG)
 
-config=ConfigParser.ConfigParser()
-config.read('model_generation.cfg')
-
-master=config.get('spark','master')
-
-idf_model_file=config.get('io','idf_model_file')
-nb_model_file=config.get('io','nb_model_file')
-hashing_tf_file=config.get('io', 'hashing_tf_file')
-tokenizer_file=config.get('io', 'tokenizer_file')
+# set up configuration file parser
+config = ConfigParser.ConfigParser()
+config.read('streaming_prediction')
+
+master = config.get('spark', 'master')
+broker_ip = config.get('kafka', 'broker_ip')
+kafka_topic = config.get('kafka', 'kafka_topic')
+kafka_output_topic = config.get('kafka', 'kafka_output_topic')
+
+tokenizer_file = config.get('io', 'tokenizer_file')
+hashing_tf_file = config.get('io', 'hashing_tf_file')
+idf_model_file = config.get('io', 'idf_model_file')
+nb_model_file = config.get('io', 'nb_model_file')
+
+idf_model = None
+nb_model = None
+hashing_tf = None
+tokenizer = None
+
+tags_to_catId_transform = None
+catId_to_tags_transform = None
+
+
+def process_data(dStream, kafka_producer):
+
+def features_extraction(df):
+# Extract featrues
+try:
+logger.debug('Extracting features from data')
+words_df = tokenizer.transform(df)
+tf_features_df = hashing_tf.transform(words_df)
+tf_idf_features_df = idf_model.transform(tf_features_df)
+logger.debug('Extract features successfully')
+return tf_features_df
+except:
+logger.warn('Fail to extract features from Questions')
+
+def predict_tag(df):
+# Predict the tags according to extracted features
+try:
+logger.debug('Predicting data tag')
+post_data = df.withColumn('CatId', tags_to_catId_transform('Tag'))
+prediction = nb_model.transform(post_data)
+output_data = prediction.withColumn('Predicted_tag', catId_to_tags('CatId'))
+logger.debug('Predicted tags are generated')
+return output_data
+except:
+logger.warn('Fail to predict tags')
+
+# Write data back to Kafka producer
+def persist_data(row):
+tagged_data = json.dumps(row.asDict())
+try:
+logger.debug('Wrting data to Kafka topic %s' % kafka_output_topic)
+kafka_producer.send(kafka_output_topic, value=tagged_data)
+logger.info('sent data successfully')
+except:
+logger.debug('Fail to send stock data %s' % tagged_data)
+
+stream_df = spark.read.json(dStream)
+features_df = features_extraction(stream_df)
+predictions = predict_tag(features_df)
+predictions.foreach(persist_data)
+
+
+# Create shut down hook
+def shutdown_hook(kafka_producer, spark):
+# Shut down kafka producer
+try:
+logger.debug('Closing kafka producer')
+kafka_producer.flush(10)
+kafka_producer.close()
+logger.debug('Stop kafka producer successfully')
+except KafkaError as ke:
+logger.warn('Fail to stop kafka producer, caused by %s' % ke.message)
 
-def process_dStream(dStream):
+try;
+logger.debug('Shut down spark context')
+spark.close()
+logger.debug('Stop spark successfully')
+except:
+logger.warn('Fail to stop spark')
 
 
 if __name__ == '__main__':
-
-# Try to initialize a spark cluster with master, master can be local or mesos URL, which is configurable in config file 
+
+#build spark context
 try:
-logger.debug("Initializing Spark cluster")
-conf=SparkConf().setAppName('model_generation').setMaster(master)
-sc=SparkContext(conf=conf)
+logger.debug('Set up sparkcontext and sparkstreamingcontext')
+conf = SparkConf().setAppName('Streaming_prediction').setMaster(master)
+sc = SparkContext(conf=conf)
 sc.setLogLevel('INFO')
-ssc=StreamingContext(sc, 5)
-logger.debug("Created Spark cluster successfully")
-except:
-logger.error("Fail to initialize spark cluster")
-
+ssc = StreamingContext(sc, 5)
+logger.debug('Initialize spark context and sparkstreamingcontext successfully')
+except Exception as e:
+logger.debug('Fail to start spark context and sparkstreamingcontext')
+raise
+finally:
+sc.close()
+
+# Start a sparksession
 try:
-spark=SparkSession.builder.config(conf=conf).getOrCreate()
-logger.debug("Initialized spark session successfully")
+logger.debug('Set up SparkSession')
+spark = SparkSession.builder.getOrCreate()
+logger.debug('Start spark session successfully')
 except:
-logger.error("Fail to start spark session")
+logger.debug('Fail to start sparksession')
 
+# Connect to Kafka broker
 try:
-# Cread Dstream from multiple kafka topics and create a microbatch every 5 seconds
-directKafkaStream=KafkaUtils.createDirectStream(ssc, READ_TOPICS, {'metadata.broker.list':BROKER})
-logger.info('Create spark direct stream successfully')
+# Create kafka producer
+logger.debug('Initialize kafka producer')
+kafka_producer = KafkaProducer(bootstrap_servers=broker_ip)
+logger.debug('Start kafka producer successfully')
+except KafkaError as ke:
+logger.debug('Fail to start kafka producer, caused by %s' % ke.message)
 
+try:
+# Create dstream from kafka topic
+directKafkaStream = KafkaUtils.createDirectStream(ssc, kafka_topic, {'metadata.broker.list' = broker_ip})
+logger.debug('Create direct dstream from kafka successfully')
 except:
-logger.debug('Fail to create direct stream')
+logger.debug('Unable to create dstream from kafka')
 
-logger.info('Start to process data')
-process_dStream(directKafkaStream)
+atexit.register(shutdown_hook, kafka_producer, spark)
 
+# Load in idf_model, nb_model, hashing_tf, idf_model and tag_catId map
 try:
-# Create Kafka producer
-KAFKA_PRODUCER=KafkaProducer(bootstrap_servers=BROKER)
-logger.info('Create kafka producer successfully')
-
-except KafkaError as ke:
-logger.debug('Fail to create kafka producer, caused by %s' % ke.message)
-
-
-
-
-
-
-
-
+logger.debug('Loading models')
+tokenizer = Tokenizer.load(tokenizer_file)
+hashing_tf = HashingTF.load(hashing_tf_file)
+idf_model = IDFModel.load(idf_model_file)
+nb_model = NaiveBayesModel.load(nb_model_file)
+selected_tags = pd.read_csv(selected_tags_file, header=None)
+local_catId_to_tags = dict(zip(list(selected_tags.index), selected_tags[0]))
+local_tags_to_catId=dict(zip(selected_tags[0], list(selected_tags.index)))
+catId_to_tags = sc.broadcast(local_catId_to_tags)
+tags_to_catId = sc.broadcast(local_tags_to_catId)
+tags_to_catId_transform = udf(lambda tag: float(tags_to_catId.value[tag]), FloatType())
+catId_to_tags_transform = udf(lambda catId: catId_to_tags.value[catId], StringType())
+logger.debug('loaded models successfully')
+except:
+logger.debug('Fail to load models')
 
 
+logger.debug('Start to process data')
+process_data(directKafkaStream, kafka_producer)
+ssc.start()
+ssc.awaitTermination()
 
+