Parquet file in invalid state while writing to S3 from EMR

[asfimport]

I'm writing parquet to S3 from Spark 1.6.1 on EMR. And when it got to the last few files to write to S3, I received this stacktrace in the log with no other errors before or after it. It's very consistent. This particular batch keeps erroring the same way.

2016-06-10 01:46:05,282] WARN org.apache.spark.scheduler.TaskSetManager [task-result-getter-2hread] - Lost task 3737.0 in stage 2.0 (TID 10585, ip-172-16-96-32.ec2.internal): org.apache.spark.SparkException: Task failed while writing rows.
	at org.apache.spark.sql.execution.datasources.DynamicPartitionWriterContainer.writeRows(WriterContainer.scala:414)
	at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:150)
	at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1$$anonfun$apply$mcV$sp$3.apply(InsertIntoHadoopFsRelation.scala:150)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66)
	at org.apache.spark.scheduler.Task.run(Task.scala:89)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:214)
	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)
Caused by: java.io.IOException: The file being written is in an invalid state. Probably caused by an error thrown previously. Current state: COLUMN
	at org.apache.parquet.hadoop.ParquetFileWriter$STATE.error(ParquetFileWriter.java:146)
	at org.apache.parquet.hadoop.ParquetFileWriter$STATE.startBlock(ParquetFileWriter.java:138)
	at org.apache.parquet.hadoop.ParquetFileWriter.startBlock(ParquetFileWriter.java:195)
	at org.apache.parquet.hadoop.InternalParquetRecordWriter.flushRowGroupToStore(InternalParquetRecordWriter.java:153)
	at org.apache.parquet.hadoop.InternalParquetRecordWriter.close(InternalParquetRecordWriter.java:113)
	at org.apache.parquet.hadoop.ParquetRecordWriter.close(ParquetRecordWriter.java:112)
	at org.apache.spark.sql.execution.datasources.parquet.ParquetOutputWriter.close(ParquetRelation.scala:101)
	at org.apache.spark.sql.execution.datasources.DynamicPartitionWriterContainer.writeRows(WriterContainer.scala:405)
	... 8 more

Reporter: Peter Halliday

_{Note: This issue was originally created as PARQUET-632. Please see the migration documentation for further details.}

[asfimport]

James (Mac) Lockard:
+1 on this, getting a similar issue on my end!

[asfimport]

viru reddy:
I am having the same issue in spark 1.6

[asfimport]

Gajulapalli Praveen Kumar:
i'm also facing this issue. I'm using spark 2.2.0

[asfimport]

Ryan Blue / @rdblue:
[~pkgajulapalli], there isn't enough information here to know what's happening. Can you post the schema of the dataframe you're writing and a stack trace?

[asfimport]

Gajulapalli Praveen Kumar:
@rdblue, the schema of dataframe is pretty simple.

scala> df.schema
res2: org.apache.spark.sql.types.StructType = StructType(StructField(col1,StringType,true), StructField(col2,IntegerType,true), StructField(col3,DateType,true), StructField(col4,StringType,true), StructField(col5,DecimalType(18,4),true))
The stack trace is same as in description.

[asfimport]

Ryan Blue / @rdblue:
[~pkgajulapalli], can you go ahead and post the stack trace? I thought you said you were using 2.2.0. These classes definitely changed.

[asfimport]

Tristan Davolt:
I am having a similar issue when calling ParquetWriter.write directly with parquet-hadoop 1.10.0.

EDIT: I resolved my issue by ensuring no race conditions were present on the instance of ParquetWriter I was using. Note that I am not using Spark. Perhaps there is a race condition within Spark?

[asfimport]

t oo:
I am facing same issue writing to S3A path with spark standalone (v2.3.4) but no EMR.

2020-02-17 21:26:35,367 [task-result-getter-3] WARN org.apache.spark.scheduler.TaskSetManager - Lost task 0.0 in stage 41.0 (TID 47, xxx, executor 1): org.apache.spark.SparkException: Task failed while writing rows.
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:288)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:198)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$write$1.apply(FileFormatWriter.scala:197)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:87)
at org.apache.spark.scheduler.Task.run(Task.scala:109)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:345)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.io.IOException: No space left on device
at java.io.FileOutputStream.writeBytes(Native Method)
at java.io.FileOutputStream.write(FileOutputStream.java:326)
at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82)
at java.io.BufferedOutputStream.write(BufferedOutputStream.java:126)
at org.apache.hadoop.fs.s3a.S3AOutputStream.write(S3AOutputStream.java:140)
at org.apache.hadoop.fs.FSDataOutputStream$PositionCache.write(FSDataOutputStream.java:58)
at java.io.DataOutputStream.write(DataOutputStream.java:107)
at java.io.FilterOutputStream.write(FilterOutputStream.java:97)
at org.apache.parquet.bytes.ConcatenatingByteArrayCollector.writeAllTo(ConcatenatingByteArrayCollector.java:46)
at org.apache.parquet.hadoop.ParquetFileWriter.writeDataPages(ParquetFileWriter.java:443)
at org.apache.parquet.hadoop.ColumnChunkPageWriteStore$ColumnChunkPageWriter.writeToFileWriter(ColumnChunkPageWriteStore.java:186)
at org.apache.parquet.hadoop.ColumnChunkPageWriteStore.flushToFileWriter(ColumnChunkPageWriteStore.java:245)
at org.apache.parquet.hadoop.InternalParquetRecordWriter.flushRowGroupToStore(InternalParquetRecordWriter.java:168)
at org.apache.parquet.hadoop.InternalParquetRecordWriter.checkBlockSizeReached(InternalParquetRecordWriter.java:143)
at org.apache.parquet.hadoop.InternalParquetRecordWriter.write(InternalParquetRecordWriter.java:125)
at org.apache.parquet.hadoop.ParquetRecordWriter.write(ParquetRecordWriter.java:180)
at org.apache.parquet.hadoop.ParquetRecordWriter.write(ParquetRecordWriter.java:46)
at org.apache.spark.sql.execution.datasources.parquet.ParquetOutputWriter.write(ParquetOutputWriter.scala:40)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$DynamicPartitionWriteTask$$anonfun$execute$5.apply(FileFormatWriter.scala:563)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$DynamicPartitionWriteTask$$anonfun$execute$5.apply(FileFormatWriter.scala:530)
at scala.collection.Iterator$class.foreach(Iterator.scala:893)
at scala.collection.AbstractIterator.foreach(Iterator.scala:1336)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$DynamicPartitionWriteTask.execute(FileFormatWriter.scala:530)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:272)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$3.apply(FileFormatWriter.scala:270)
at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1417)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$.org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask(FileFormatWriter.scala:275)
... 8 more
Suppressed: java.io.IOException: The file being written is in an invalid state. Probably caused by an error thrown previously. Current state: COLUMN
at org.apache.parquet.hadoop.ParquetFileWriter$STATE.error(ParquetFileWriter.java:182)
at org.apache.parquet.hadoop.ParquetFileWriter$STATE.startBlock(ParquetFileWriter.java:174)
at org.apache.parquet.hadoop.ParquetFileWriter.startBlock(ParquetFileWriter.java:284)
at org.apache.parquet.hadoop.InternalParquetRecordWriter.flushRowGroupToStore(InternalParquetRecordWriter.java:166)
at org.apache.parquet.hadoop.InternalParquetRecordWriter.close(InternalParquetRecordWriter.java:109)
at org.apache.parquet.hadoop.ParquetRecordWriter.close(ParquetRecordWriter.java:163)
at org.apache.spark.sql.execution.datasources.parquet.ParquetOutputWriter.close(ParquetOutputWriter.scala:42)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$DynamicPartitionWriteTask.releaseResources(FileFormatWriter.scala:577)
at org.apache.spark.sql.execution.datasources.FileFormatWriter$$anonfun$org$apache$spark$sql$execution$datasources$FileFormatWriter$$executeTask$1.apply$mcV$sp(FileFormatWriter.scala:278)
at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1426)
... 9 more
2020-02-17 21:26:35,368 [dispatcher-event-loop-9] INFO org.apache.spark.scheduler.TaskSetManager - Starting task 0.1 in stage 41.0 (TID 48, xxx, executor 1, partition 0, PROCESS_LOCAL, 8657 bytes)

[asfimport]

Emil Kleszcz:
The same issue was observed in spark 3.2.1 when writing a parquet file to HDFS using parquet v. 1.12.2. 

Some of the logs from the server side in HDFS, NN:

<2022-11-14T08:12:31.941+0100> <INFO> <org.apache.hadoop.hdfs.StateChange>: <BLOCK* allocate blk_5112663052_4039631919, replicas=xxx:1004, xxx:1004, xxx:1004 for /project//part-00003-a1dc7884-5736-437f-a573-35c8d0ea922f.c000.snappy.parquet>
<2022-11-14T08:13:43.029+0100> <INFO> <org.apache.hadoop.hdfs.StateChange>: <BLOCK* allocate blk_5112664776_4039633643, replicas=xxx:1004, xxx:1004, xxx:1004 for /project/437f-a573-35c8d0ea922f.c000.snappy.parquet>
<2022-11-14T08:14:51.025+0100> <INFO> <org.apache.hadoop.ipc.Server>: <IPC Server handler 124 on default port 8020, call Call#64349 Retry#0 org.apache.hadoop.hdfs.protocol.ClientProtocol.addBlock from xxx:45140: java.io.FileNotFoundException: File does not exist: /project/part-00003-a1dc7884-5736-437f-a573-35c8d0ea922f.c000.snappy.parquet (inode 7450217277) Holder DFSClient_NONMAPREDUCE_-885269500_210 does not have any open files.>

DN logs for one of the file blocks:
 

<2022-11-14T08:12:31.945+0100> <INFO> <org.apache.hadoop.hdfs.server.datanode.DataNode>: <Receiving BP-1877419787-xxx-1507645851253:blk_5112663052_4039631919 src: /xxx:43856 dest: /xxx:1004> <2022-11-14T08:13:43.028+0100> <INFO> <org.apache.hadoop.hdfs.server.datanode.DataNode.clienttrace>: <src: /xxx:43856, dest: /xxx:1004, bytes: 268435456, op: HDFS_WRITE, cliID: DFSClient_NONMAPREDUCE_-885269500_210, offset: 0, srvID: e20a1db9-5297-41b3-a49a-a6dc6a2169b6, blockid: BP-1877419787-xxx-1507645851253:blk_5112663052_4039631919, duration(ns): 71075002589> <2022-11-14T08:13:43.028+0100> <INFO> <org.apache.hadoop.hdfs.server.datanode.DataNode>: <PacketResponder: BP-1877419787-xxx-1507645851253:blk_5112663052_4039631919, type=HAS_DOWNSTREAM_IN_PIPELINE, downstreams=2:[xxx:1004, xxx:1004] terminating> <2022-11-14T08:15:01.095+0100> <INFO> <org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsDatasetAsyncDiskService>: <Scheduling blk_5112663052_4039631919 replica FinalizedReplica, blk_5112663052_4039631919, FINALIZED <2022-11-14T08:15:01.157+0100> <INFO> <org.apache.hadoop.hdfs.server.datanode.fsdataset.impl.FsDatasetAsyncDiskService>: <Deleted BP-1877419787-xxx-1507645851253 blk_5112663052_4039631919 URI file:/data18/hdfs/data/current/BP-1877419787-xxx-1507645851253/current/finalized/subdir29/subdir12/blk_5112663052>