Spark 2.1 无法在 CSV 上写入 Vector 字段答案

【问题标题】：Spark 2.1 cannot write Vector field on CSVSpark 2.1 无法在 CSV 上写入 Vector 字段
【发布时间】：2017-05-24 14:05:03
【问题描述】：

我在将代码从 Spark 2.0 迁移到 2.1 时偶然发现了与数据帧保存相关的问题。

这是代码

import org.apache.spark.sql.types._
import org.apache.spark.ml.linalg.VectorUDT
val df = spark.createDataFrame(Seq(Tuple1(1))).toDF("values")
val toSave = new org.apache.spark.ml.feature.VectorAssembler().setInputCols(Array("values")).transform(df)
toSave.write.csv(path)

此代码在使用 Spark 2.0.0 时成功

使用 Spark 2.1.0.cloudera1，我收到以下错误：

java.lang.UnsupportedOperationException: CSV data source does not support struct<type:tinyint,size:int,indices:array<int>,values:array<double>> data type.
  at org.apache.spark.sql.execution.datasources.csv.CSVFileFormat.org$apache$spark$sql$execution$datasources$csv$CSVFileFormat$$verifyType$1(CSVFileFormat.scala:233)
  at org.apache.spark.sql.execution.datasources.csv.CSVFileFormat$$anonfun$verifySchema$1.apply(CSVFileFormat.scala:237)
  at org.apache.spark.sql.execution.datasources.csv.CSVFileFormat$$anonfun$verifySchema$1.apply(CSVFileFormat.scala:237)
  at scala.collection.Iterator$class.foreach(Iterator.scala:893)
  at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
  at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
  at org.apache.spark.sql.types.StructType.foreach(StructType.scala:96)
  at org.apache.spark.sql.execution.datasources.csv.CSVFileFormat.verifySchema(CSVFileFormat.scala:237)
  at org.apache.spark.sql.execution.datasources.csv.CSVFileFormat.prepareWrite(CSVFileFormat.scala:121)
  at org.apache.spark.sql.execution.datasources.FileFormatWriter$.write(FileFormatWriter.scala:108)
  at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelationCommand.run(InsertIntoHadoopFsRelationCommand.scala:101)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult$lzycompute(commands.scala:58)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.sideEffectResult(commands.scala:56)
  at org.apache.spark.sql.execution.command.ExecutedCommandExec.doExecute(commands.scala:74)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:114)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:114)
  at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:135)
  at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
  at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:132)
  at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:113)
  at org.apache.spark.sql.execution.QueryExecution.toRdd$lzycompute(QueryExecution.scala:87)
  at org.apache.spark.sql.execution.QueryExecution.toRdd(QueryExecution.scala:87)
  at org.apache.spark.sql.execution.datasources.DataSource.writeInFileFormat(DataSource.scala:484)
  at org.apache.spark.sql.execution.datasources.DataSource.write(DataSource.scala:520)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:215)
  at org.apache.spark.sql.DataFrameWriter.save(DataFrameWriter.scala:198)
  at org.apache.spark.sql.DataFrameWriter.csv(DataFrameWriter.scala:579)
  ... 50 elided

这只是在我身边吗？

这与 Spark 2.1 的 cloudera 版本有关吗？（从他们的仓库来看，他们似乎没有弄乱 spark.sql 所以也许不是）

谢谢！

【问题讨论】：

这是预期的。 CSV 源不支持复杂对象。与您的例外情况完全相同：CSV 数据源不支持 struct,values:array> 数据类型。
是的，我想，但为什么它可以与 Spark 2.0 一起使用？
它在 2.0 中不起作用。它曾经在 1.x 中与 spark-csv 一起使用，其中向量已转换为字符串。
好吧，我刚刚重新启动了脚本，它可以工作了。
让我们换一种说法 - 使用干净的 2.0 (2.0.2) 二进制文件，它不起作用并且它假设会失败:)

标签： csv apache-spark apache-spark-sql spark-csv

【解决方案1】：

以下答案由@zero323 的评论组成。

CSV 源不支持复杂对象。与您的例外情况完全相同：CSV 数据源不支持 struct,values:array‌> 数据类型。 是预期行为。它不适用于 Spark 2.x，尽管它曾经在 1.x 中与 spark-csv 一起使用，其中向量已转换为字符串。

此行为在以下 jira SPARK-16216 中是正确的。

【讨论】：

只是补充一点，它适用于 Spark 2.0.0。它从 2.0.1 及更高版本停止工作。

【解决方案2】：

作为一种解决方法，您可以使用 fork 中的 VectorDisassembler 类，或采用 here 中描述的解决方案。

我使用 VectorDisassembler 将 ml.feature.StandardScaler.fit 方法的结果数据框存储到 CSV 中。

代码大致如下：

val disassembler = new org.apache.spark.ml.feature.VectorDisassembler()
val disassembledDF = disassembler.setInputCol("scaledFeatures").transform(df)
disassembledDF.show()

【讨论】：