Large rows cause unnecessary OOM exceptions

[asfimport]

If columns contains lots of lengthy string value, it will run into OOM error during writing.

2014-09-22 19:16:11,626 FATAL [main] org.apache.hadoop.mapred.YarnChild: Error running child : java.lang.OutOfMemoryError: Java heap space
at java.util.Arrays.copyOf(Arrays.java:2271)
at java.io.ByteArrayOutputStream.grow(ByteArrayOutputStream.java:113)
at java.io.ByteArrayOutputStream.ensureCapacity(ByteArrayOutputStream.java:93)
at java.io.ByteArrayOutputStream.write(ByteArrayOutputStream.java:140)
at org.apache.hadoop.io.compress.CompressorStream.compress(CompressorStream.java:83)
at org.apache.hadoop.io.compress.CompressorStream.write(CompressorStream.java:76)
at parquet.bytes.CapacityByteArrayOutputStream.writeTo(CapacityByteArrayOutputStream.java:144)
at parquet.bytes.BytesInput$CapacityBAOSBytesInput.writeAllTo(BytesInput.java:308)
at parquet.bytes.BytesInput$SequenceBytesIn.writeAllTo(BytesInput.java:233)
at parquet.hadoop.CodecFactory$BytesCompressor.compress(CodecFactory.java:108)
at parquet.hadoop.ColumnChunkPageWriteStore$ColumnChunkPageWriter.writePage(ColumnChunkPageWriteStore.java:110)
at parquet.column.impl.ColumnWriterImpl.writePage(ColumnWriterImpl.java:147)
at parquet.column.impl.ColumnWriterImpl.flush(ColumnWriterImpl.java:236)
at parquet.column.impl.ColumnWriteStoreImpl.flush(ColumnWriteStoreImpl.java:113)
at parquet.hadoop.InternalParquetRecordWriter.flushStore(InternalParquetRecordWriter.java:151)
at parquet.hadoop.InternalParquetRecordWriter.checkBlockSizeReached(InternalParquetRecordWriter.java:130)
at parquet.hadoop.InternalParquetRecordWriter.write(InternalParquetRecordWriter.java:122)
at parquet.hadoop.ParquetRecordWriter.write(ParquetRecordWriter.java:81)
at parquet.hadoop.ParquetRecordWriter.write(ParquetRecordWriter.java:37)
at org.apache.hadoop.hive.ql.io.parquet.write.ParquetRecordWriterWrapper.write(ParquetRecordWriterWrapper.java:77)
at org.apache.hadoop.hive.ql.io.parquet.write.ParquetRecordWriterWrapper.write(ParquetRecordWriterWrapper.java:90)
at org.apache.hadoop.hive.ql.exec.FileSinkOperator.processOp(FileSinkOperator.java:688)
at org.apache.hadoop.hive.ql.exec.Operator.process(Operator.java:502)
at org.apache.hadoop.hive.ql.exec.Operator.forward(Operator.java:832)
at org.apache.hadoop.hive.ql.exec.SelectOperator.processOp(SelectOperator.java:84)
at org.apache.hadoop.hive.ql.exec.Operator.process(Operator.java:502)
at org.apache.hadoop.hive.ql.exec.Operator.forward(Operator.java:832)
at org.apache.hadoop.hive.ql.exec.FilterOperator.processOp(FilterOperator.java:132)
at org.apache.hadoop.hive.ql.exec.Operator.process(Operator.java:502)
at org.apache.hadoop.hive.ql.exec.Operator.forward(Operator.java:832)
at org.apache.hadoop.hive.ql.exec.TableScanOperator.processOp(TableScanOperator.java:90)
at org.apache.hadoop.hive.ql.exec.Operator.process(Operator.java:502)

Reporter: Tongjie Chen / @tongjiechen
Assignee: Daniel Weeks / @danielcweeks

Related issues:

• Release Parquet-mr 1.9.0 (blocks)
• InternalParquetRecordWriter doesn't use min/max row counts (relates to)

PRs and other links:

• PR #250
• PR #297

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

[asfimport]

Jianshi Huang:
Does disabling dictionary encoding helps?

Jianshi

[asfimport]

Tongjie Chen / @tongjiechen :
I disable dictionary and it still run into the problem.
it keep creating new slabs (after 10 times, grow exponentially) until it exhaust memory.

Sep 24, 2014 7:10:31 PM INFO: parquet.bytes.CapacityByteArrayOutputStream: used 11 slabs, new slab size 55827257
Sep 24, 2014 7:10:31 PM INFO: parquet.bytes.CapacityByteArrayOutputStream: new slab of size 55827257
Sep 24, 2014 7:10:34 PM INFO: parquet.bytes.CapacityByteArrayOutputStream: used 12 slabs, new slab size 102292952
Sep 24, 2014 7:10:34 PM INFO: parquet.bytes.CapacityByteArrayOutputStream: new slab of size 102292952
Sep 24, 2014 7:10:42 PM INFO: parquet.bytes.CapacityByteArrayOutputStream: used 13 slabs, new slab size 212521210
Sep 24, 2014 7:10:42 PM INFO: parquet.bytes.CapacityByteArrayOutputStream: new slab of size 212521210
Sep 24, 2014 7:10:55 PM INFO: parquet.bytes.CapacityByteArrayOutputStream: used 14 slabs, new slab size 426081554
Sep 24, 2014 7:10:55 PM INFO: parquet.bytes.CapacityByteArrayOutputStream: new slab of size 426081554
Sep 24, 2014 7:10:56 PM INFO: parquet.hadoop.InternalParquetRecordWriter: mem size 463,559,051 > 134,217,728: flushing 100 records to disk.
Sep 24, 2014 7:10:56 PM INFO: parquet.hadoop.InternalParquetRecordWriter: Flushing mem store to file. allocated memory: 907,972,142

[asfimport]

Tongjie Chen / @tongjiechen :
This is a more like a data issue. Close it.

[asfimport]

Julien Le Dem / @julienledem:
did you have one huge record?

[asfimport]

Tongjie Chen / @tongjiechen :
That table has a bunch of huge records.

[asfimport]

Jianshi Huang:
It's a data issue, but I think Parquet should take care of the situation, say:

If (string.length > X) won't use dictionary encoding.

Jianshi

[asfimport]

Tongjie Chen / @tongjiechen :
Re-open the jira. We hit the same problem a few more times and unfortunately there are no configuration to work around it.

Basically when there are continuous fat rows, Parquet needs an elegant way to handle it.

[asfimport]

Ryan Blue / @rdblue:
Do we have any idea what the boundary condition is that causes this problem?

Also: is this something we should do by 1.6.0?

[asfimport]

Tongjie Chen / @tongjiechen :
We have one file that contains 1100 rows of data, of which 670 rows are around 2M, and parquet fails to write those 1100 rows to a file.

[asfimport]

Daniel Weeks / @danielcweeks:
This has been affecting us pretty regularly and I have a fix we're using internally and will clean it up and supply a patch.

There are two ways this can be triggered:

1. Rows are so large that OOM occurs before the initial page size check (ColumnWriter). This happens a when you have many very large rows (multiple megabytes per row). Of course, this might indicate that you have issues with your data, but parquet should be able to work around it.
2. The size of the rows vary significantly and the estimation for the next page size check is determined by small rows and is set very high followed by may large rows. This is the case we've been running into recently.

The fix I have is to be able to configure the initial page size check and to force constant checking (bypass the estimation) because the estimate can be problematic with certain datasets.

-Dan

[asfimport]

Daniel Weeks / @danielcweeks:
Pull request: #250

[asfimport]

Daniel Weeks / @danielcweeks:
#250

[asfimport]

Ryan Blue / @rdblue:
Merged Dan's changes in #297. Thanks, Dan!