将包含 Vector 作为特征的 Spark 数据帧转换为 CSV 文件答案

【问题标题】：Converting a spark dataframe that contains Vector as a feature into CSV file将包含 Vector 作为特征的 Spark 数据帧转换为 CSV 文件
【发布时间】：2019-07-30 12:17:23
【问题描述】：

我需要将 spark 数据帧转换为 CSV 文件。

问题是其中一个特征是向量结构，我不知道如何处理它。由于潜在狄利克雷分配的lda.transform() 函数，我得到了这个数据框。我正在使用spark.ml 库。

我通常使用的通用df.coalesce(1).write.option("header", true).csv("<file output path>") 在这种情况下不起作用。

这是数据框的样子：

org.apache.spark.sql.DataFrame = [label: bigint, topicDistribution: vector]

scala> df_new.show
+-----+--------------------+
|label|   topicDistribution|
+-----+--------------------+
|    0|[6.71576085454879...|
|    1|[5.74898984641732...|
|    2|[1.50297841245588...|
|    3|[3.83251655971072...|
|    4|[8.62396858027641...|
|    5|[1.54900186503601...|
|    6|[1.61493761327978...|
|    7|[9.36938609746372...|
|    8|[1.34332824038935...|
|    9|[1.09991943420077...|
|   10|[1.29739085981008...|
|   11|[8.92839698024594...|
|   12|[4.12414455173398...|
|   13|[1.56497583389755...|
|   14|[1.80725990261967...|
|   15|[1.80725990261967...|
|   16|[9.36938609746372...|
|   17|[1.53335069918914...|
|   18|[1.51801264939744...|
|   19|[7.15931646362877...|
+-----+--------------------+
only showing top 20 rows

我不需要将这些数组扩展为单独的功能，只需将其转换为原样即可。如何将其转换为 CSV 文件？

【问题讨论】：

请问为什么df.coalesce(1).write.option("header", true).csv("<file output path>") 不起作用？错误是什么？
type Vector is not supported...

标签： scala apache-spark

【解决方案1】：

考虑到使用coalesce(1) 会影响DataFrame 之前的步骤的执行方式。仅传递 1 个分区作为 coalesce 的参数将大大降低并行度，因此您可能会遇到内存错误。

请尝试repartition，因为这不会影响先前操作的并行性，并且您会得到类似的结果（即只写入一个 csv）。

应该是这样的：

df.repartition(1).write.option("header", true).csv("<file output path>")

【讨论】：

谢谢你，我以后会和repartition一起做。但是 CSV 转换不支持类型向量，所以我需要做一些解决方法。
你为什么不把你的列映射成一个字符串呢？像df_new.map(<function_that_converts_vector_column_to_string>) 这样的东西。这样，生成的 DataFrame 将有一个字符串，由 CSV 库管理。这也可以控制您要使用的字符串表示形式。如果您需要具体示例，请告诉我，我将创建一个包含详细信息的新答案。
最后我设法使用df.map { row => (row.getAs[Long]("label"), "[" + row.getAs[org.apache.spark.ml.linalg.Vector]("topicDistribution").toArray.mkString(",") + "]") } 做到了。它有效，但这是一个糟糕的 hack。我相信一定有更好的方法，但现在就可以了。感谢您的意见，这很有帮助！

【解决方案2】：

最后我设法做到了使用

df.map { row => (row.getAs[Long]("label"), "[" + row.getAs[org.apache.spark.ml.linalg.Vector("topicDistribution").toArray.mkString(",") + "]") }

它有效，但这是一个糟糕的 hack。我相信一定有更好的方法，但现在可以了。

【讨论】：