diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java index 8979dba3f6..31cc96ba10 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java @@ -323,9 +323,7 @@ private static int getMaxPaddingSize(Configuration conf) { /** * constructor used when this OutputFormat in wrapped in another one (In Pig for example) - * @param writeSupportClass the class used to convert the incoming records - * @param schema the schema of the records - * @param extraMetaData extra meta data to be stored in the footer of the file + * @param writeSupport the class used to convert the incoming records */ public > ParquetOutputFormat(S writeSupport) { this.writeSupport = writeSupport; @@ -387,8 +385,6 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp if (INFO) LOG.info("Min row count for page size check is: " + props.getMinRowCountForPageSizeCheck()); if (INFO) LOG.info("Min row count for page size check is: " + props.getMaxRowCountForPageSizeCheck()); - CodecFactory codecFactory = new CodecFactory(conf, props.getPageSizeThreshold()); - WriteContext init = writeSupport.init(conf); ParquetFileWriter w = new ParquetFileWriter( conf, init.getSchema(), file, Mode.CREATE, blockSize, maxPaddingSize); @@ -411,10 +407,11 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp init.getSchema(), init.getExtraMetaData(), blockSize, - codecFactory.getCompressor(codec), + codec, validating, props, - memoryManager); + memoryManager, + conf); } /** diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java index 6c94fac5c6..a9ade96b00 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetRecordWriter.java @@ -21,6 +21,7 @@ import java.io.IOException; import java.util.Map; +import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.mapreduce.RecordWriter; import org.apache.hadoop.mapreduce.TaskAttemptContext; @@ -28,6 +29,7 @@ import org.apache.parquet.column.ParquetProperties.WriterVersion; import org.apache.parquet.hadoop.CodecFactory.BytesCompressor; import org.apache.parquet.hadoop.api.WriteSupport; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; import org.apache.parquet.schema.MessageType; import static org.apache.parquet.Preconditions.checkNotNull; @@ -43,8 +45,9 @@ */ public class ParquetRecordWriter extends RecordWriter { - private InternalParquetRecordWriter internalWriter; - private MemoryManager memoryManager; + private final InternalParquetRecordWriter internalWriter; + private final MemoryManager memoryManager; + private final CodecFactory codecFactory; /** * @@ -79,6 +82,7 @@ public ParquetRecordWriter( internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, extraMetaData, blockSize, compressor, validating, props); this.memoryManager = null; + this.codecFactory = null; } /** @@ -106,14 +110,17 @@ public ParquetRecordWriter( boolean validating, WriterVersion writerVersion, MemoryManager memoryManager) { - this(w, writeSupport, schema, extraMetaData, blockSize, compressor, - validating, ParquetProperties.builder() - .withPageSize(pageSize) - .withDictionaryPageSize(dictionaryPageSize) - .withDictionaryEncoding(enableDictionary) - .withWriterVersion(writerVersion) - .build(), - memoryManager); + ParquetProperties props = ParquetProperties.builder() + .withPageSize(pageSize) + .withDictionaryPageSize(dictionaryPageSize) + .withDictionaryEncoding(enableDictionary) + .withWriterVersion(writerVersion) + .build(); + internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, + extraMetaData, blockSize, compressor, validating, props); + this.memoryManager = checkNotNull(memoryManager, "memoryManager"); + memoryManager.addWriter(internalWriter, blockSize); + this.codecFactory = null; } /** @@ -123,7 +130,7 @@ public ParquetRecordWriter( * @param schema the schema of the records * @param extraMetaData extra meta data to write in the footer of the file * @param blockSize the size of a block in the file (this will be approximate) - * @param compressor the compressor used to compress the pages + * @param codec the compression codec used to compress the pages * @param validating if schema validation should be turned on * @param props parquet encoding properties */ @@ -133,12 +140,15 @@ public ParquetRecordWriter( MessageType schema, Map extraMetaData, long blockSize, - BytesCompressor compressor, + CompressionCodecName codec, boolean validating, ParquetProperties props, - MemoryManager memoryManager) { + MemoryManager memoryManager, + Configuration conf) { + this.codecFactory = new CodecFactory(conf, props.getPageSizeThreshold()); internalWriter = new InternalParquetRecordWriter(w, writeSupport, schema, - extraMetaData, blockSize, compressor, validating, props); + extraMetaData, blockSize, codecFactory.getCompressor(codec), validating, + props); this.memoryManager = checkNotNull(memoryManager, "memoryManager"); memoryManager.addWriter(internalWriter, blockSize); } @@ -148,9 +158,16 @@ public ParquetRecordWriter( */ @Override public void close(TaskAttemptContext context) throws IOException, InterruptedException { - internalWriter.close(); - if (memoryManager != null) { - memoryManager.removeWriter(internalWriter); + try { + internalWriter.close(); + // release after the writer closes in case it is used for a last flush + } finally { + if (codecFactory != null) { + codecFactory.release(); + } + if (memoryManager != null) { + memoryManager.removeWriter(internalWriter); + } } } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java index f58dda4288..58cbe957d0 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java @@ -52,6 +52,7 @@ public class ParquetWriter implements Closeable { public static final int MAX_PADDING_SIZE_DEFAULT = 0; private final InternalParquetRecordWriter writer; + private final CodecFactory codecFactory; /** * Create a new ParquetWriter. @@ -273,7 +274,7 @@ public ParquetWriter(Path file, Configuration conf, WriteSupport writeSupport conf, schema, file, mode, blockSize, maxPaddingSize); fileWriter.start(); - CodecFactory codecFactory = new CodecFactory(conf, encodingProps.getPageSizeThreshold()); + this.codecFactory = new CodecFactory(conf, encodingProps.getPageSizeThreshold()); CodecFactory.BytesCompressor compressor = codecFactory.getCompressor(compressionCodecName); this.writer = new InternalParquetRecordWriter( fileWriter, @@ -300,6 +301,9 @@ public void close() throws IOException { writer.close(); } catch (InterruptedException e) { throw new IOException(e); + } finally { + // release after the writer closes in case it is used for a last flush + codecFactory.release(); } }