스칼라 스파크에서 문장을 문장 안에서 단어로 분리하는 법 (case (key, value) => ...)

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, love), (2, rose) ...와 같이 매핑 할 것입니다 ...

Answer 1

org.apache.spark.rdd.PairRDDFunctions (문서화 된 here)을 가져와 키 - 값 쌍으로보다 쉽게 ​​작업 할 수 있습니다.

그 시점에서 flatMapValues 메서드를 사용하여 원하는 것을 얻을 수 있습니다.

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

당신이 만약이 코드의 몇 줄의 출력

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) 

: 여기에 최소한의 일 예이다 (당신은 스파크 콘솔에있는 경우 val tweets을 포함하는 행 복사) Spark Streaming으로 라이브 트위터 피드를 분석하는 방법을 보여주는 작은 예를보고 싶다면 here을 찾을 수 있습니다.