我应该如何将org.apache.spark.ml.linalg.Vector的RDD转换为Dataset？

Question

我很难理解RDD，DataSet和DataFrame之间的转换是如何工作的。我对Spark很陌生，每当我需要从数据模型传递到另一个数据模型时（特别是从RDD到Datasets和Dataframes），我都会遇到困难。任何人都能解释我正确的方法吗？

作为一个例子，现在我有一个RDD[org.apache.spark.ml.linalg.Vector]，我需要将它传递给我的机器学习算法，例如KMeans（Spark DataSet MLlib）。因此，我需要使用名为“features”的单个列将其转换为Dataset，该列应包含Vector类型的行。我该怎么做？

Answer 1

要将RDD转换为数据帧，最简单的方法是在Scala中使用toDF()。要使用此功能，必须导入使用SparkSession对象完成的含义。它可以如下完成：

val spark = SparkSession.builder().getOrCreate()
import spark.implicits._

val df = rdd.toDF("features")

toDF()采用元组的RDD。当RDD由常见的Scala对象构建时，它们将被隐式转换，即不需要做任何事情，并且当RDD有多个列时也不需要做任何事情，RDD已经包含一个元组。但是，在这种特殊情况下，您需要先将RDD[org.apache.spark.ml.linalg.Vector]转换为RDD[(org.apache.spark.ml.linalg.Vector)]。因此，有必要按如下方式转换为元组：

val df = rdd.map(Tuple1(_)).toDF("features")

以上内容将RDD转换为具有名为features的单个列的数据帧。

---

要转换为数据集，最简单的方法是使用案例类。确保在Main对象外部定义了case类。首先将RDD转换为数据帧，然后执行以下操作：

case class A(features: org.apache.spark.ml.linalg.Vector)

val ds = df.as[A]

要显示所有可能的转换，可以使用.rdd从数据框或数据集访问基础RDD：

val rdd = df.rdd

---

不是在RDD和数据帧/数据集之间来回转换，而是使用数据帧API进行所有计算通常更容易。如果没有合适的函数来执行您想要的操作，通常可以定义UDF，用户定义的函数。例如，请参见：https://jaceklaskowski.gitbooks.io/mastering-spark-sql/spark-sql-udfs.html

Answer 2

你需要的只是一个Encoder。进口

import org.apache.spark.sql.Encoder
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder
import org.apache.spark.ml.linalg

RDD：

val rdd = sc.parallelize(Seq(
  linalg.Vectors.dense(1.0, 2.0), linalg.Vectors.sparse(2, Array(), Array())
))

转换：

val ds = spark.createDataset(rdd)(ExpressionEncoder(): Encoder[linalg.Vector])
 .toDF("features")

ds.show
// +---------+
// | features|
// +---------+
// |[1.0,2.0]|
// |(2,[],[])|
// +---------+


ds.printSchema
// root
//  |-- features: vector (nullable = true)