在 Spark-Scala 中将 Dataset[Row] 转换为 RDD[Array[String]] 的最佳方法？

Question

我正在通过读取 csv 文件创建一个 spark 数据集。此外，我需要将此 Dataset[Row] 转换为 RDD[Array[String]] 以将其传递给 FpGrowth(Spark MLLIB)。

val df: DataFrame = spark.read.format("csv").option("header", "true").load("/path/to/csv")
val ds: Dataset[Row] = df.groupBy("user").agg(collect_set("values"))

现在，我需要 select 列“values”并将结果数据集转换为 RDD[Array[String]].

val rddS: RDD[String] = ds.select(concat_ws(",", col("values")).as("items")).distinct().rdd.map(_.mkString(","))
val rddArray: RDD[Array[String]] = rddS.map(s => s.trim.split(','))

我尝试了这种方法，但不确定它是否是最好的方法。请建议我实现此目标的最佳方法。

Answer 1

为什么不简单地使用如下，你将减少concat_ws和split操作。

val rddS:RDD[Array[String]] = ds.select("values")
    .distinct()
    .rdd.map(r => r.getAs[mutable.WrappedArray[String]](0).toArray)

Answer 2

一行：

val rddArray: RDD[Array[String]] = ds.select("values").as[Array[String]].rdd

顺便说一下，我建议使用基于数据帧的 Spark ML 而不是基于 RDD 的 Spark MLLib，后者现已弃用。您可以使用 org.apache.spark.ml.fpm.FPGrowth.

Answer 3

我最终使用了 toSeq 方法

val rddArray: RDD[Array[String]] = ds.select("values").rdd.map(r => r.getSeq[String](0).toArray)

这对我的用例来说更有效（更快）。

在 Spark-Scala 中将 Dataset[Row] 转换为 RDD[Array[String]] 的最佳方法？

Best approach to transform Dataset[Row] to RDD[Array[String]] in Spark-Scala?

scala

apache-spark

apache-spark-sql

rdd

apache-spark-dataset