Styjun

We are building sentiment analysis application and we converted our tweets dataframe to an array. We created another array consisting of positive words. But we cannot count the number of tweets containing one of those positive words. We tried these and we get 1 as result. It must be more than 1. Apparently it did not count:

val sqlContext = new org.apache.spark.sql.SQLContext(sc)
var tweetDF = sqlContext.read.json("hdfs:///sandbox/tutorial-files/770/tweets_staging/*")
tweetDF.show()
var messages = tweetDF.select("msg").collect.map(_.toSeq) 
println("Total messages: " + messages.size)
val positive = Source.fromFile("/home/teslavm/positive.txt").getLines.toArray
var happyCount=0
for (e <- 0 until messages.size) 
 for (f <- 0 until positive.size) 
 if (messages(e).contains(positive(f)))
 happyCount=happyCount+1
 
 

print("nNumber of happy messages: " +happyCount) 

This should work.
It has the advantage that you do not have to collect the result, as well as being more functional.

val messages = tweetDF.select("msg").as[String]

val positiveWords =
 Source
 .fromFile("/home/teslavm/positive.txt")
 .getLines
 .toList
 .map(word => word.toLowerCase)

def hasPositiveWords(message: String): Boolean = 
 val _message = message.toLowerCase
 positiveWords.exists(word => _message.contains(word))
 

val positiveMessages = messages.filter(hasPositiveWords _)

println(positiveMessages.count())

I tested this code locally with:

import org.apache.spark.sql.SparkSession

val spark = SparkSession.builder.master("local[*]").getOrCreate()
import spark.implicits._

val tweetDF = List(
 (1, "Yes I am happy"),
 (2, "Sadness is a way of life"),
 (3, "No, no, no, no, yes")
).toDF("id", "msg")

val positiveWords = List("yes", "happy")

And it worked.