将 [(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 需要元素、索引和值的数量
none 接受列表 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's documentation its constructor receives a Double as a label and a Vector as features (DenseVector or 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])

将 [(Int, Seq[Double])] RDD 转换为 LabeledPoint

Converting a [(Int, Seq[Double])] RDD to LabeledPoint

scala

apache-spark

apache-spark-mllib