Обработка нескольких «линий» в apache-spark RDD

Question

Я очень новичок в Spark, и у меня есть вопрос.

Я пытаюсь сделать простой анализ настроения с некоторыми данными. В файле данных каждая строка содержит обзор продукта.

Вот мой код для обработки одной строки:

// wordlist 
val pos_file = "/user/cloudera/Data/pos_list.txt" 
val neg_file = "/user/cloudera/Data/neg_list.txt" 

val pos_words = sc.textFile(pos_file).cache().collect().toSet 
val neg_words = sc.textFile(neg_file).cache().collect().toSet 


val test_string = "Line with positive or negative review." 
val test_rdd = sc.parallelize(List(test_string)) 

val test_rdd2 = test_rdd.flatMap(line => "[a-zA-Z]+".r findAllIn line map (_.toLowerCase)) 

val pos = test_rdd2.filter(x => pos_words contains x) 
val neg = test_rdd2.filter(x => neg_words contains x) 

Мой вопрос теперь, как обрабатывать каждую запись в РДУ (в данном случае 3):

val file_in = "/user/cloudera/Data/teststring.txt" 
val data = sc.textFile(file_in).cache() 
val reviews = data.flatMap(_.split("\n")) 

scala> reviews.count() 
res29: Long = 3 

Следующий код

val reviews2 = reviews.flatMap(line => "[a-zA-Z]+".r findAllIn line map (_.toLowerCase)) 

дает мне все слова. Я хочу получить значения для pos и neg для каждой строки/обзора. Вычисление довольно просто: если слово находится в наборе pos_words/neg_words, тогда положите его в pos/neg. На самом деле, я просто считаю, что встречаются положительные или отрицательные слова.

Как получить что-то вроде ('line', 'posvalue', 'negvalue')?

Большое спасибо заранее

Answer 1

Попытка обобщить этот вопрос: мы хотим, чтобы подсчитать число вхождений определенных слов (положительный, отрицательный) по каждой строке входного файла после его «нормализуется»: все алфавитные символы и пробелы в нижнем регистре.

Давайте предположим, что я уже два набора положительных и отрицательных слов:

val posWords: Set[String] = ??? 
val negWords: Set[String] = ??? 

и файл ввода с 1 записей в каждой строке:

val data = sc.textFile("data.txt") 

Мы хотим, чтобы результат формы :

(text, posCount, negCount) 

Давайте сначала определим вспомогательную функцию, которая позволяет нам рассчитывать совпадения слов в последовательности Строки против набора строк:

def matches(text:Seq[String], words:Set[String]):Int = 
text.map(w => if (words.contains(w)) 1 else 0).sum 

И, наконец, мы преобразуем каждую строку в графу как положительных, так и отрицательных матчей.

val posNegData = data.map{line => 
    val cleanLine = line.toLowerCase.split("\\W") 
    (line, matches(cleanLine, posWords), matches(cleanLine, negWords)) 
} 

Здесь мы предполагаем, что ожидаемым результатом является исходная строка с положительными и отрицательными совпадениями. (Это непонятно из первоначального вопроса)