Пара ключей/значений RDD

Question

У меня есть вопрос о парах ключей/значений RDD.

У меня есть пять файлов в C:/download/input папке, которая имеет диалоги в фильмах, как содержание файлов следующим образом:

movie_horror_Conjuring.txt 
movie_comedy_eurotrip.txt 
movie_horror_insidious.txt 
movie_sci-fi_Interstellar.txt 
movie_horror_evildead.txt 

Я пытаюсь читать файлы в папке ввода с помощью СБН. wholeTextFiles(), где я получаю ключ/значение следующего

(C:/download/input/movie_horror_Conjuring.txt,values) 

Я пытаюсь сделать операцию, где я должен сгруппировать входные файлы каждый жанр вместе с помощью groupByKey(). Значения всех фильмов ужасов вместе, комедийные фильмы вместе и так далее.

Есть ли способ, что я могу генерировать пары ключей/значение таким образом (horror, values) вместо (C:/download/input/movie_horror_Conjuring.txt,values)

val ipfile = sc.wholeTextFiles("C:/download/input") 
val output = ipfile.groupByKey().map(t => (t._1,t._2)) 

Приведенный выше код дает мне выход следующим

(C:/download/input/movie_horror_Conjuring.txt,values) 
(C:/download/input/movie_comedy_eurotrip.txt,values) 
(C:/download/input/movie_horror_Conjuring.txt,values) 
(C:/download/input/movie_sci-fi_Interstellar.txt,values) 
(C:/download/input/movie_horror_evildead.txt,values) 

, где, как мне нужно выход следующим образом:

(horror, (values1, values2, values3)) 
(comedy, (values1)) 
(sci-fi, (values1)) 

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

Также я хотел бы знать, как я могу получить подсчитывать строки в values1, values2, и т.д. Значение 3

Мой окончательный вывод должен быть как

(ужас, 100)

где 100 - сумма количества строк в значениях 1 = 40 строк, значения 2 = 30 строк и значений 3 = 30 строк и т. Д.

Answer 1

Попробуйте следующее:

val output = ipfile.map{case (k, v) => (k.split("_")(1),v)}.groupByKey()  
output.collect 

Дайте мне знать, если это сработает для вас!

Update:

Чтобы получить выход в формате (horror, 100):

val output = ipfile.map{case (k, v) => (k.split("_")(1),v.count(_ == '\n'))}.reduceByKey(_ + _)  
output.collect