（阵列/ ML矢量/ MLlib矢量）RDD到ML矢量数据帧coulmn

Question

我需要一个RDD转换为单个柱o.a.s.ml.linalg.Vector数据帧中，为了使用ML算法，特别KMEANS对于这种情况。这是我的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是私人的，我会得到一个错误。

我ALGO试图转换RDD为双打，以数据帧的数组，并获得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：

   ​​

2. 使用text源和UDF：

   def parse_(n: Int, m: Int)(s: String) = Try(
       Vectors.dense(s.split(',').slice(n, m).map(_.toDouble)) 
   ).toOption 
   
   def parse(n: Int, m: Int) = udf(parse_(n, m) _) 
   
   val raw = spark.read.text(path) 
   
   raw.select(parse(n, m)(col(raw.columns.head)).alias("features")) 
   

3. 使用text源并放下包装

   spark.read.text(path).as[String].map(parse_(n, m)).toDF