从 Spark 保存时 Parquet 错误答案

【问题标题】：Parquet error when saving from Spark从 Spark 保存时 Parquet 错误
【发布时间】：2015-07-09 17:45:06
【问题描述】：

在 Spark 1.3.0 中对 DataFrame 重新分区后，保存到 Amazon 的 S3 时出现 .parquet 异常>.

logsForDate
    .repartition(10)
    .saveAsParquetFile(destination) // <-- Exception here

我收到的异常是：

java.io.IOException: The file being written is in an invalid state. Probably caused by an error thrown previously. Current state: COLUMN
at parquet.hadoop.ParquetFileWriter$STATE.error(ParquetFileWriter.java:137)
at parquet.hadoop.ParquetFileWriter$STATE.startBlock(ParquetFileWriter.java:129)
at parquet.hadoop.ParquetFileWriter.startBlock(ParquetFileWriter.java:173)
at parquet.hadoop.InternalParquetRecordWriter.flushRowGroupToStore(InternalParquetRecordWriter.java:152)
at parquet.hadoop.InternalParquetRecordWriter.close(InternalParquetRecordWriter.java:112)
at parquet.hadoop.ParquetRecordWriter.close(ParquetRecordWriter.java:73)
at org.apache.spark.sql.parquet.ParquetRelation2.org$apache$spark$sql$parquet$ParquetRelation2$$writeShard$1(newParquet.scala:635)
at org.apache.spark.sql.parquet.ParquetRelation2$$anonfun$insert$2.apply(newParquet.scala:649)
at org.apache.spark.sql.parquet.ParquetRelation2$$anonfun$insert$2.apply(newParquet.scala:649)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:61)
at org.apache.spark.scheduler.Task.run(Task.scala:64)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:203)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)

我想知道是什么问题以及如何解决。

【问题讨论】：

您是每次还是有时都会收到错误消息？您是否也为较小的文件获得它？您是否也只能在 S3 或其他文件系统上获得它？你试过 Apache Spark 1.3.1 吗？ Its release notes 提及一些与 Parquet 相关的修复。
在超过特定文件大小工作时，我总是收到错误消息。我只试过S3。我试过 1.3.0.d.
我能够在 EMR 上使用 Spark 1.3.1 重现此错误，并写入 S3。使用旧的 Parquet api (sqlContext.setConf("spark.sql.parquet.useDataSourceApi", "false")) 没有帮助。写入 HDFS 工作正常。
您是否尝试过在 us-west-1 区域使用存储桶？或使用 emfrs
@Interfector 我们可以解决这个问题。我有同样的问题

标签： apache-spark parquet

【解决方案1】：

我在将 asParquetFile 保存到 HDFS 时遇到了这个错误。因为datanode socket write timeout，所以我在Hadoop设置里把它改成更长的：

<property>
  <name>dfs.datanode.socket.write.timeout</name>
  <value>3000000</value>
</property>
<property>
  <name>dfs.socket.timeout</name>
  <value>3000000</value>
</property>

如果您可以像这样设置 S3，希望这会有所帮助。

【讨论】：

s3n 似乎没有任何等效设置
@Interfector，请检查工作日志，看看那里是否有有用的消息。我稍后会删除答案
提供的堆栈跟踪是我能找到的唯一一条信息。
@Interfector，请同时检查 s3n datanode 日志。
在 master 和 slave 上运行 grep -lir "The file being written is in an invalid state" /mnt 没有成功。

【解决方案2】：

当保存到 S3 时，我实际上可以在 EMR 上使用 Spark 1.3.1 重现此问题。

但是，保存到 HDFS 可以正常工作。您可以先保存到 HDFS，然后使用例如s3distcp 将文件移动到 S3。

【讨论】：

不过，我更喜欢直接保存到 S3 的解决方案。

【解决方案3】：

您确定这不是由于SPARK-6351（将镶木地板保存到 S3 时出现“错误的 FS”）吗？如果是，则与重新分区无关，在spark-1.3.1中已修复。但是，如果您像我一样因为使用 CDH-5.4.0 而被 spark-1.3.0 卡住，那么我昨晚刚刚想出了一种直接从代码中绕过它的方法（无需更改配置文件）：

spark.hadoopConfiguration.set("fs.defaultFS", "s3n://mybucket")

之后，我可以毫无问题地将镶木地板文件保存到 S3。

但请注意，这样做有几个缺点。我认为（没有尝试）它将无法写入到 S3 之外的另一个 FS，也可能无法写入另一个存储桶。它也可能会强制 Spark 将临时文件写入 S3 而不是本地，但我也没有检查过。

【讨论】：

您可以在 CDH 5.4 上运行 Spark 1.3.1 或 1.4.0 就好了。只需像任何其他 YARN 应用程序一样运行它
听起来不错。你怎么做呢？我试图在我们的 build.sbt 中更改 spark 版本，但它坏了。我们使用 CDH 的原因之一是避免因依赖项版本不兼容而头疼。
我不确定是不是因为这个，但我已经尝试使用 Spark 1.3.1 并且它不起作用。