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

【问题标题】：Converting a [(Int, Seq[Double])] RDD to LabeledPoint将 [(Int, Seq[Double])] RDD 转换为 LabeledPoint
【发布时间】：2016-06-29 10:27:17
【问题描述】：

我有一个以下格式的 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 元素转换成什么，但我不知道。

【问题讨论】：

标签： scala apache-spark apache-spark-mllib

【解决方案1】：

您可以在LabeledPoint's documentation 中注意到，它的构造函数接收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])

【讨论】：

【解决方案2】：

这里有几个问题：

标签应该是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: _*)) }

【讨论】：