Ошибка паркета при сохранении от Spark

После перекомпоновки DataFrame в Spark 1.3.0 я получаю исключение .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$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)

Я хотел бы знать, в чем проблема и как ее решить.

Ответ 1

Я могу воспроизвести эту проблему с помощью Spark 1.3.1 на EMR, при сохранении на S3.

Однако сохранение в HDFS прекрасное. Сначала вы можете сохранить в HDFS, а затем использовать, например, s3distcp, чтобы переместить файлы на S3.

Ответ 2

Я столкнулся с этой ошибкой, когда saveAsParquetFile в 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 следующим образом.

Ответ 3

Вы уверены, что это происходит не из-за SPARK-6351 ( "Wrong FS" при сохранении паркета на S3 )? Если это так, то это не имеет никакого отношения к перераспределению, и оно было исправлено в искро-1.3.1. Если же, как и я, вы застряли с искро-1.3.0, потому что используете CDH-5.4.0, я только что выяснил, что прошлым вечером можно обойти это прямо из кода (без изменения файла конфигурации):

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

После этого я могу без проблем сохранить паркетные файлы на S3.

Обратите внимание, что для этого есть несколько недостатков. Я думаю (не пытался), что он тогда не сможет записать в другой FS, а не в S3, а может быть и в другое ведро. Это может также заставить Spark записывать временные файлы на S3, а не локально, но я также не проверял это.