scala - 对于 DStream 中的每个 RDD，如何将其转换为数组或其他一些典型的 Java 数据类型？

Question

我想将 DStream 转换为数组、列表等，然后我可以将其转换为 json 并在端点上提供它。我正在使用 apache spark，注入 twitter 数据。如何在 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你可以做到的。

val arr = new ArrayBuffer[String]();
statuses.foreachRDD {
    arr ++= _.collect() //you can now put it in an array or d w/e you want with it
    ...
}

请记住，由于 DStream 可能很大，因此最终可能会比您的驱动程序中的数据多得多。

Answer 2

转身我们的你很接近，但我最终寻找的是。

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