Как разбить предложения на слова внутри карты (case (key, value) => ...) in scala spark

Question

val sparkConf = new SparkConf().setAppName("ShortTwitterAnalysis").setMaster("local[2]") 
val sc = new SparkContext(sparkConf) 
val text = sc.textFile("/home/tobbyj/HW1_INF553/shortTwitter.txt") 
val twitter = text 
    .map(_.toLowerCase) 
    .map(_.replace("\t", "")) 
    .map(_.replace("\"", "")) 
    .map(_.replace("\n", "")) 
    .map(_.replace(".", "")) 
    .map(_.replaceAll("[\\p{C}]", "")) 
    .map(_.split("text:")(1).split(",source:")(0)) 
    .zipWithIndex.map(_.swap) 

Использование приведенного выше кода У меня есть результаты, как показано ниже.

(0,a rose by any other name would smell as sweet) 
(1,a rose is a rose is a rose) 
(4,rt @nba2k: the battle of two young teams tough season but one will emerge victorious who will it be? lakers or 76ers? https:\/\/tco\/nukkjq\u2026) 
(2,love is like a rose the joy of all the earth) 
(5,i was going to bake a cake and listen to the football flour refund?) 
(3,at christmas i no more desire a rose than wish a snow in may’s new-fangled mirth) 

Однако результат я хочу «ключ», начиная с 1 и «значение» разделены на слова, как показано ниже для понимания, хотя я не уверен, что это будет выглядеть, как показано ниже.

(1,(a, rose, by, any, other, name, would, smell, as, sweet)) 
(2,(a, rose, is, a, rose, is, a, rose)) 
... 

код Я устал это

.map{case(key, value)=>(key+1, value.split(" "))} 

но дают мне результаты, как показано ниже

(1,[Ljava.lang.String;@1dff58b) 
(2,[Ljava.lang.String;@167179a3) 
(3,[Ljava.lang.String;@73e8c7d7) 
(4,[Ljava.lang.String;@7bffa418) 
(5,[Ljava.lang.String;@2d385beb) 
(6,[Ljava.lang.String;@4f1ab87e) 

Любые предложения? После этого шага я буду отображать их как (1, a), (1, роза), (1, by) ... (2, любовь), (2, роза), ....

Answer 1

Вы можете импортировать org.apache.spark.rdd.PairRDDFunctions (задокументированный here), чтобы легче работать с парами ключ-значение.

В этот момент вы можете использовать метод flatMapValues, чтобы получить то, что вы хотите; здесь минимальный рабочий пример (просто скопировать из строки, содержащей val tweets, если вы находитесь в консоли Спарк):

import org.apache.spark._ 
import org.apache.spark.rdd.PairRDDFunctions 

val conf = new SparkConf().setAppName("test").setMaster("local[*]") 
val sc = new SparkContext(conf) 

val tweets = sc.parallelize(Seq(
    "this is my first tweet", 
    "and this is my second", 
    "ok this is getting boring")) 

val results = 
    tweets. 
    zipWithIndex. 
    map(_.swap). 
    flatMapValues(_.split(" ")) 

results.collect.foreach(println) 

Это выход из этого несколько строк кода:

(0,this) 
(0,is) 
(0,my) 
(0,first) 
(0,tweet) 
(1,and) 
(1,this) 
(1,is) 
(1,my) 
(1,second) 
(2,ok) 
(2,this) 
(2,is) 
(2,getting) 
(2,boring) 

Если вы заинтересованы в том, чтобы увидеть небольшой пример, показывающий, как анализировать живой канал Twitter с помощью Spark Streaming, вы можете найти один here.