我正在使用Scala编写一个Apache Spark应用程序。为了处理和存储数据,我使用DataFrames。我有一个不错的流水线,包括特征提取和MultiLayerPerceptron分类器,使用ML API。
我还想使用SVM(为了比较目的)。问题是(如果我说错了请纠正),只有MLLib提供SVM。而MLLib还没有准备好处理DataFrames,只能处理RDDs。
因此,我认为可以使用DataFrames维护我的应用程序核心,并使用SVM来1)仅将我需要的DataFrame列转换为RDD [LabeledPoint]
,并且2)在分类后将SVM预测添加到DataFrame作为新列。
第一部分我使用了一个小函数来完成:
private def dataFrameToRDD(dataFrame : DataFrame) : RDD[LabeledPoint] = {
val rddMl = dataFrame.select("label", "features").rdd.map(r => (r.getInt(0).toDouble, r.getAs[org.apache.spark.ml.linalg.SparseVector](1)))
rddMl.map(r => new LabeledPoint(r._1, Vectors.dense(r._2.toArray)))
}
由于特征提取方法使用的是ML API而不是MLLib,因此我必须指定和转换向量的类型。
然后,将这个RDD[LabeledPoint]
输入SVM进行分类,一切顺利,没有问题。最后,按照Spark的示例,我得到了一个RDD[Double]
:
val predictions = rdd.map(point => model.predict(point.features))
现在,我想将预测分数作为列添加到原始DataFrame中并返回它。这就是我卡住的地方。我可以使用
RDD[Double]
将其转换为DataFrame
(sql context ommited)
import sqlContext.implicits._
val plDF = predictions.toDF("prediction")
但是如何将第二个DataFrame作为原始DataFrame的一列加入其中呢?我尝试使用join
和union
方法,但由于DataFrames上没有相同的列进行连接或合并,所以遇到了SQL异常。
编辑
data.withColumn("prediction", plDF.col("prediction"))
但是我遇到了一个分析异常 :(