从 Spark 保存时出现 Parquet 错误

Question

在 Spark 1.3.0 中对 DataFrame 重新分区后，我得到一个 .parquet 异常保存到 Amazon 的 S3.

时

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(newParquet.scala:635)
at org.apache.spark.sql.parquet.ParquetRelation2$$anonfun$insert.apply(newParquet.scala:649)
at org.apache.spark.sql.parquet.ParquetRelation2$$anonfun$insert.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)

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

Answer 1

我在将 saveAsParquetFile 保存到 HDFS 时遇到了这个错误。是因为datanode socket write timeout，所以我在Hadoop Settings里把它改长了：

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

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

Answer 2

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

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

Answer 3

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

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

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

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

从 Spark 保存时出现 Parquet 错误

Parquet error when saving from Spark

apache-spark

parquet