将Rdd[Vector]转换为Rdd[Double]

Question

如何将csv转换为RddDouble？在这一行中，我有一个错误:无法应用于(org.apache.spark.rdd.RDDUnit)：

val kd = new KernelDensity().setSample(rows) 

我的完整代码在这里：

   import org.apache.spark.mllib.linalg.Vectors
    import org.apache.spark.mllib.linalg.distributed.RowMatrix
    import org.apache.spark.mllib.stat.KernelDensity
    import org.apache.spark.rdd.RDD
    import org.apache.spark.{SparkContext, SparkConf}

class KdeAnalysis {
  val conf = new SparkConf().setAppName("sample").setMaster("local")
  val sc = new SparkContext(conf)

  val DATAFILE: String = "C:\\Users\\ajohn\\Desktop\\spark_R\\data\\mass_cytometry\\mass.csv"
  val rows = sc.textFile(DATAFILE).map {
    line => val values = line.split(',').map(_.toDouble)
      Vectors.dense(values)
  }.cache()



  // Construct the density estimator with the sample data and a standard deviation for the Gaussian
  // kernels
  val rdd : RDD[Double] = sc.parallelize(rows)
  val kd = new KernelDensity().setSample(rdd)
    .setBandwidth(3.0)

  // Find density estimates for the given values
  val densities = kd.estimate(Array(-1.0, 2.0, 5.0))
}

Answer 1

由于rows是一个RDD[org.apache.spark.mllib.linalg.Vector]，下面的行无法工作：

val rdd : RDD[Double] = sc.parallelize(rows)

parallelize预计Seq[T]和RDD不是Seq。

即使这个部分像你期望的那样起作用，你的输入也是完全错误的。KernelDensity.setSample的一个正确论点是RDD[Double]或JavaRDD[java.lang.Double]。目前看来，它不支持多元数据。

关于瓷砖上的一个问题，您可以flatMap

rows.flatMap(_.toArray)

或者创建rows时更好

val rows = sc.textFile(DATAFILE).flatMap(_.split(',').map(_.toDouble)).cache()

但我怀疑这真的是你需要的。

Answer 2

已经准备好了这段代码，请评估它是否能帮助您走出->

val doubleRDD = rows.map(_.toArray).flatMap(x => x)