(数组/ ML向量/MLlib向量) RDD到ML向量

Question

我需要将RDD转换为单列o.a.s.ml.linalg.Vector DataFrame，以便使用ML算法，特别是这种情况下的K-方法。这是我的RDD：

val parsedData = sc.textFile("/digits480x.csv").map(s => Row(org.apache.spark.mllib.linalg.Vectors.dense(s.split(',').slice(0,64).map(_.toDouble))))

我试着做this答案暗示的事情，我想，因为你最终得到了一个MLlib向量，它在运行算法时会抛出一个不匹配的错误。现在如果我改变这个：

import org.apache.spark.mllib.linalg.{Vectors, VectorUDT}

val schema = new StructType()
  .add("features", new VectorUDT())

对此：

import org.apache.spark.ml.linalg.{Vectors, VectorUDT}

val parsedData = sc.textFile("/digits480x.csv").map(s => Row(org.apache.spark.ml.linalg.Vectors.dense(s.split(',').slice(0,64).map(_.toDouble))))

val schema = new StructType()
  .add("features", new VectorUDT())

我会得到一个错误，因为ML VectorUDT是私有的。

我还尝试将RDD转换为一个双重数组到Dataframe，并获得如下所示的ML密集向量：

var parsedData = sc.textFile("/home/pililo/Documents/Mi_Memoria/Codigo/Datasets/Digits/digits480x.csv").map(s => Row(s.split(',').slice(0,64).map(_.toDouble)))

parsedData: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row]

val schema2 = new StructType().add("features", ArrayType(DoubleType))

schema2: org.apache.spark.sql.types.StructType = StructType(StructField(features,ArrayType(DoubleType,true),true))

val df = spark.createDataFrame(parsedData, schema2)

df: org.apache.spark.sql.DataFrame = [features: array<double>]

val df2 = df.map{ case Row(features: Array[Double]) => Row(org.apache.spark.ml.linalg.Vectors.dense(features)) }

即使导入了spark.implicits._，也会引发以下错误：

error: Unable to find encoder for type stored in a Dataset.  Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._  Support for serializing other types will be added in future releases.

任何帮助都是非常感谢的，谢谢！

Answer 1

从我的头顶：

1. 使用csv源代码和VectorAssembler： 导入scala.util.Try org.apache.spark.ml.linalg._导入org.apache.spark.ml.feature.VectorAssembler val路径:org.apache.spark.ml.feature.VectorAssembler=？凡恩: Int =？val m:Int =？val raw = spark.read.csv(path) val featureCols = raw.columns.slice(n，m) val exprs = featureCols.map(c => col(c).cast("double")) val汇编程序=新VectorAssembler() .setInputCols(featureCols) .setOutputCol(“功能”) assembler.transform(raw.select(exprs：_*)).select($"features")
2. 使用text源代码和UDF： def parse_(n: Int，m: Int)(s: String) = Try( Vectors.dense(s.split('，').slice(n，m) .map(_.toDouble ).toOption def解析(n: Int，m: Int) = udf(parse_(n，m) _)
3. 使用text源并删除包装Row spark.read.text(path).asString.map(parse_(n，m)).toDF