DStream [Double、Double]をRDDに変換する（Double、Double）

Question

ストリーミングデータに対して線形回帰モデルを訓練する必要があります。私はtextFileStreamを使ってストリーミングデータを読んだ。しかし、問題は、RegressionMetricsがRDD[(Double, Double)]を受け付け、outputがDStream[Double,Double]の形式であることです。 outputをRDD[(Double, Double)]に変換する方法をRegressionMetricsにできますか？

val model = new StreamingLinearRegressionWithSGD() 
     .setInitialWeights(Vectors.dense(0.0, 0.0)) 
     .setStepSize(0.2) 
     .setNumIterations(25) 

    trainingData = ssc.textFileStream("/training/data/dir").map(LabeledPoint.parse) 
testData = ssc.textFileStream("/training/data/dir").map(LabeledPoint.parse) 

model.trainOn(trainingData) 

val output = model.predictOnValues(testData.map(lp => (lp.label, lp.features))) 

val metrics = new RegressionMetrics(output) 
val rmse = metrics.rootMeanSquaredError 

Answer 1

すべてのDSTREAMはforeachRDDメソッドを使用してアクセスすることができる基本的なRDD（すべてのデータのバッチのための独立した1）、から構成されています

model.predictOnValues(testData.map(lp => (lp.label, lp.features))).foreachRDD { rdd => 
    val metrics = new RegressionMetrics(rdd) 
    val rmse = metrics.rootMeanSquaredError 
    // do something with `rmse` here 
}