DStream의 각 RDD에 대해 이것을 배열이나 다른 일반적인 Java 데이터 유형으로 변환하려면 어떻게해야합니까?

Question

DStream을 배열, 목록 등으로 변환하고 싶습니다. 그래서 json으로 변환하여 엔드 포인트에서 사용할 수 있습니다. 나는 아파치 스파크를 사용하고, 트위터 데이터를 주입하고있다. 이 작업을 Dstream statuses에서 어떻게 수행합니까? 나는 print() 이외의 다른 것을 얻을 수없는 것 같습니다.

import org.apache.spark._ 
import org.apache.spark.SparkContext._ 
import org.apache.spark.streaming._ 
import org.apache.spark.streaming.twitter._ 
import org.apache.spark.streaming.StreamingContext._ 
import TutorialHelper._ 
object Tutorial { 
    def main(args: Array[String]) { 

    // Location of the Spark directory 
    val sparkHome = "/opt/spark" 

    // URL of the Spark cluster 
    val sparkUrl = "local[8]" 

    // Location of the required JAR files 
    val jarFile = "target/scala-2.10/tutorial_2.10-0.1-SNAPSHOT.jar" 

    // HDFS directory for checkpointing 
    val checkpointDir = "/tmp" 

    // Configure Twitter credentials using twitter.txt 
    TutorialHelper.configureTwitterCredentials() 

    val ssc = new StreamingContext(sparkUrl, "Tutorial", Seconds(1), sparkHome, Seq(jarFile)) 

    val filters = Array("#americasgottalent", "iamawesome") 
    val tweets = TwitterUtils.createStream(ssc, None, filters) 

    val statuses = tweets.map(status => status.getText()) 

    val arry = Array("firstval") 
    statuses.foreachRDD { 
     arr :+ _.collect() 
    } 

    ssc.checkpoint(checkpointDir) 

    ssc.start() 
    ssc.awaitTermination() 
    } 
} 

Answer 1

RDD가 statuses 인 경우 수행 할 수 있습니다.

DStream이 거대 할 수 있기 때문에 드라이버에서 원하는 것보다 더 많은 데이터가 될 수 있습니다.

Answer 2

우리가 당신을 닫았지만, 내가 결국 찾았습니다.

statuses.foreachRDD(rdd => { 
    for(item <- rdd.collect().toArray) { 
     println(item); 
    } 
})