Проверка, содержат ли элементы массива твитов один из элементов массива положительных слов и подсчет

Question

Мы создаем приложение для анализа настроений и преобразовали наш фрейм данных твитов в массив.Мы создали еще один массив, состоящий из положительных слов.Но мы не можем подсчитать количество твитов, содержащих одно из этих позитивных слов.Мы попробовали это, и в результате получаем 1.Оно должно быть больше 1. Видимо, оно не учитывалось:

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) 

Answer 1

Это должно работать. Преимущество состоит в том, что вам не нужно собирать результат, а также в том, что он более функциональный.

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())

Я тестировал этот код локально с:

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")

И это сработало.