Liferay message boards classifier

MessageBoardsClassifier.scala

val ssc =  new StreamingContext(conf, Seconds(5))

val topicMap = topics.split(",").map((_,numThreads.toInt)).toMap

val kafkaStream = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap).map(_._2)

val model = new KMeansModel(ssc.sparkContext.objectFile[Vector](modelFile.toString).collect())

val filteredMessages = kafkaStream
  .filter(m => model.predict(Utils.featurize(m)) == clusterNumber.toInt)

filteredMessages.print()

ssc.start()
ssc.awaitTermination()