How to convert Seq to RDD when working with spark streaming context
<p>I am using <code>TestSuiteBase</code> to create some tests with <code>spark-streaming</code> (using spark streaming context <code>scc</code>). Then I create dummy data using <code>output: Seq[Seq[(Double, Double)]]</code>. Finally I want to apply some function to <code>output</code>, but this function accepts <code>RDD[(Double, Double)]</code>, not <code>Seq[Seq[(Double, Double)]]</code>. </p>

<p>To solve this issue, I am thinking to use <code>val rdd: RDD[(Double, Double)] = sc.parallelize(output.flatten)</code>, however how and where exactly should I get spark context <code>sc</code> from <code>scc</code>? Or, perhaps, there is any way to directly create dummy data in <code>RDD</code> without using <code>Seq</code>?</p>

<pre><code>class StreamingTestLR  extends SparkFunSuite
                       with TestSuiteBase {

  // use longer wait time to ensure job completion
  override def maxWaitTimeMillis: Int = 20000

  var ssc: StreamingContext = _

  override def afterFunction() {
    super.afterFunction()
    if (ssc != null) {
      ssc.stop()
    }
  }

//...

val output: Seq[Seq[(Double, Double)]] = runStreams(ssc, numBatches, numBatches)

// THE PROBLEM IS HERE!!!
// val metrics = new SomeFuncThatAcceptsRDD(rdd)

}
</code></pre>

<p>UPDATE</p>

<pre><code>  // Test if the prediction accuracy of increases when using hyper-parameter optimization
  // in order to learn Y = 10*X1 + 10*X2 on streaming data
  test("Test 1") {
    // create model initialized with zero weights
    val model = new StreamingLinearRegressionWithSGD()
      .setInitialWeights(Vectors.dense(0.0, 0.0))
      .setStepSize(0.2)
      .setNumIterations(25)

    // generate sequence of simulated data for testing
    val numBatches = 10
    val nPoints = 100
    val testInput = (0 until numBatches).map { i =&gt;
      LinearDataGenerator.generateLinearInput(0.0, Array(10.0, 10.0), nPoints, 42 * (i + 1))
    }
    val inputDStream = DStream[LabeledPoint]

    withStreamingContext(setupStreams(testInput, inputDStream)) { ssc =&gt;
      model.trainOn(inputDStream)
      model.predictOnValues(inputDStream.map(x =&gt; (x.label, x.features)))
      val output: Seq[Seq[(Double, Double)]] = runStreams(ssc, numBatches, numBatches)


      val rdd: RDD[(Double, Double)] = ssc.sparkContext.parallelize(output.flatten)

      // Instantiate metrics object
      val metrics = new RegressionMetrics(rdd)

      // Squared error
      println(s"MSE = ${metrics.meanSquaredError}")
      println(s"RMSE = ${metrics.rootMeanSquaredError}")

      // R-squared
      println(s"R-squared = ${metrics.r2}")

      // Mean absolute error
      println(s"MAE = ${metrics.meanAbsoluteError}")

      // Explained variance
      println(s"Explained variance = ${metrics.explainedVariance}")
    }
  }
</code></pre>