将 [(Int， seq[Double])] RDD 转换为 LabeledPoint

Question

>我有一个以下格式的RDD，并希望将其转换为LabeledPoint RDD，以便在mllib中处理它：

Test: RDD[(Int, Seq[Double])] = Array((1,List(1.0,3.0,8.0),(2,List(3.0, 3.0,8.0),(1,List(2.0,3.0,7.0),(1,List(5.0,5.0,9.0))

我尝试使用地图

import org.apache.spark.mllib.linalg.{Vector, Vectors}
import org.apache.spark.mllib.regression.LabeledPoint
Test.map(x=> LabeledPoint(x._1, Vectors.sparse(x._2)))

但我收到此错误

mllib.linalg.Vector cannot be applied to (Seq[scala.Double])

所以大概需要先转换 Seq 元素，但我不知道是什么。

Answer 1

这里有几个问题：

• 标签应Double而不是Int
• SparseVector需要元素、索引和值的数量
• 没有一个向量构造函数接受Double列表
• 您的数据看起来密集而不是稀疏

一种可能的解决方案：

val rdd = sc.parallelize(Array(
    (1, List(1.0,3.0,8.0)),
    (2, List(3.0, 3.0,8.0)),
    (1, List(2.0,3.0,7.0)),
    (1, List(5.0,5.0,9.0))))
rdd.map { case (k, vs) => 
  LabeledPoint(k.toDouble, Vectors.dense(vs.toArray))
}

另一个：

rdd.collect { case (k, v::vs) =>
  LabeledPoint(k.toDouble, Vectors.dense(v, vs: _*)) }

Answer 2

正如你在 LabeledPoint 的文档中注意到的那样，它的构造函数接收一个Double作为标签和一个 Vector 作为特征（DenseVector 或 SparseVector）。但是，如果你查看两个继承类的构造函数，它们会收到一个Array，因此你需要将你的Seq转换为Array。

import org.apache.spark.mllib.linalg.{Vector, Vectors, DenseVector}
import org.apache.spark.mllib.regression.LabeledPoint
val rdd = sc.parallelize(Array((1, Seq(1.0,3.0,8.0)), 
                               (2, Seq(3.0, 3.0,8.0)),
                               (1, Seq(2.0,3.0, 7.0)),
                               (1, Seq(5.0, 5.0, 9.0))))
val x = rdd.map{
    case (a: Int, b:Seq[Double]) => LabeledPoint(a, new DenseVector(b.toArray))
}
x.take(2).foreach(println)
//(1.0,[1.0,3.0,8.0])
//(2.0,[3.0,3.0,8.0])