型の不一致、予想（（ダブル、ダブル））=>ブール、実際の（（ダブル、ダブル））=>どれ

Question

Iによる3つのエラーのためにコンパイルされません、次の二つの機能を持っている：

1. RegressionMetrics：Cannot resolve constructor

2. _.nonEmpty：Type mismatch, expected ((Double,Double))=>Boolean, actual ((Double,Double))=>Any

3. reduce(_+_)：Cannot resolve symbol +。

コード：

import org.apache.spark.mllib.evaluation.RegressionMetrics 

//.. 

def getRMSE (rdd: RDD): Double = { 
    val metrics = new RegressionMetrics(rdd) 
    metrics.rootMeanSquaredError 
} 

def calculateRMSE(output: DStream[(Double, Double)]): Double = { 
    output.filter(_.nonEmpty).map(getRMSE).reduce(_+_) 
} 

test("Test1") { 
// do some data preprocessing 
// call the function calculateRMSE 
} 

これらのエラーを修正するためにどのように任意のアイデア？

P.S：testの内部にval metrics = new RegressionMetrics(rdd)を入れても奇妙なことは問題なくコンパイルできます。

UPDATE：私は追加することによって、問題の＃1を解決することができた

(Double,Double)RDDへ：

def getRMSE(rdd : RDD[(Double, Double)]) : Double = { 
    val metrics = new RegressionMetrics(rdd) 
    metrics.rootMeanSquaredError 
    } 

Answer 1

reduce(func)：要素を集約 ことにより、単一要素のRDDSの新しいDSTREAMを返し ファンクションfunc（2つの引数をとり、1つを返す）を使用して、ソースDStreamの各RDDで 関数は、並列に計算できるように結合的でなければなりません。

したがって、calculateRMSEための右署名がなければならない：

def calculateRMSE(output: DStream[(Double, Double)]): DStream[Double]