For each RDD in a DStream how do I convert this to an array or some other typical Java data type?

Question

I would like to convert a DStream into an array, list, etc. so I can then translate it to json and serve it on an endpoint. I'm using apache spark, injecting twitter data. How do I preform this operation on the Dstream `statuses`? I can't seem to get anything to work other than `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",…

Accepted Answer

If your RDD is `statuses` you can do.

```
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
    ...
}
```

Keep in mind this could end up being way more data than you want in your driver since a DStream can be huge.

For each RDD in a DStream how do I convert this to an array or some other typical Java data type?

Solution

Problem