2
我使用TestSuiteBase
創建了一些使用spark-streaming
(使用火花上下文scc
)的測試。然後我使用output: Seq[Seq[(Double, Double)]]
創建虛擬數據。最後,我想將一些函數應用於output
,但此函數接受RDD[(Double, Double)]
,而不是Seq[Seq[(Double, Double)]]
。如何使用火花上下文處理將Seq轉換爲RDD
要解決此問題,我在考慮使用val rdd: RDD[(Double, Double)] = sc.parallelize(output.flatten)
,但是究竟應該如何以及在哪裏得到scc
的火花上下文sc
?或者,也許有什麼辦法可以在不使用Seq
的情況下直接在RDD
中創建虛擬數據?
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)
}
UPDATE
// 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 =>
LinearDataGenerator.generateLinearInput(0.0, Array(10.0, 10.0), nPoints, 42 * (i + 1))
}
val inputDStream = DStream[LabeledPoint]
withStreamingContext(setupStreams(testInput, inputDStream)) { ssc =>
model.trainOn(inputDStream)
model.predictOnValues(inputDStream.map(x => (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}")
}
}
它說:'不能呼籲停止SparkContext java.lang.IllegalStateException方法:無法調用已停止的SparkContext方法# – Klue
could y ou用完整的代碼示例編輯你的問題?你是否嘗試在withStreamingContext {ssc =>}中運行測試代碼? –
哪裏定義了TestServer? – Klue