Skip to content

PARQUET-353: Recycle compressors in parquet write path #282

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
nitin2goyal wants to merge 3 commits into from
Closed

PARQUET-353: Recycle compressors in parquet write path #282

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
382ccb0
PARQUET-353: Recycle compressors in parquet write path
nitindexter Oct 19, 2015
801f18e
PARQUET-353: Recycle compressors in parquet write path
nitindexter Oct 19, 2015
29cc207
Dummy checkin
nitindexter Oct 26, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 13 additions & 0 deletions parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -171,6 +171,8 @@ private final STATE error() throws IOException {

private STATE state = STATE.NOT_STARTED;

private final CodecFactory codecFactory;

@julienledem julienledem Nov 17, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It seems that this is not the right place to put this as this class does not use the codecFactory.
it just creates it and makes it available. I'd rather have private members not accessible to the outside to respect encapsulation and layering of the code.

@rdblue rdblue Dec 1, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1


/**
* @param configuration Hadoop configuration
* @param schema the schema of the data
Expand Down Expand Up @@ -226,6 +228,8 @@ public ParquetFileWriter(Configuration configuration, MessageType schema,
this.alignment = NoAlignment.get(rowGroupSize);
this.out = fs.create(file, overwriteFlag);
}

this.codecFactory = new CodecFactory(configuration);
}

/**
Expand All @@ -246,6 +250,7 @@ public ParquetFileWriter(Configuration configuration, MessageType schema,
rowAndBlockSize, rowAndBlockSize, maxPaddingSize);
this.out = fs.create(file, true, DFS_BUFFER_SIZE_DEFAULT,
fs.getDefaultReplication(file), rowAndBlockSize);
this.codecFactory = new CodecFactory(configuration);
}

/**
Expand Down Expand Up @@ -469,6 +474,7 @@ public void end(Map<String, String> extraMetaData) throws IOException {
ParquetMetadata footer = new ParquetMetadata(new FileMetaData(schema, extraMetaData, Version.FULL_VERSION), blocks);
serializeFooter(footer, out);
out.close();
codecFactory.release();
}

private static void serializeFooter(ParquetMetadata footer, FSDataOutputStream out) throws IOException {
Expand Down Expand Up @@ -588,6 +594,13 @@ public long getPos() throws IOException {
return out.getPos();
}

/**
* @return The codec factory which should be used to get compressors
*/
public CodecFactory getCodecFactory() {
return codecFactory;
}

public long getNextRowGroupSize() throws IOException {
return alignment.nextRowGroupSize(out);
}
Expand Down
3 changes: 1 addition & 2 deletions parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -341,7 +341,6 @@ public RecordWriter<Void, T> getRecordWriter(Configuration conf, Path file, Comp
throws IOException, InterruptedException {
final WriteSupport<T> writeSupport = getWriteSupport(conf);

CodecFactory codecFactory = new CodecFactory(conf);
long blockSize = getLongBlockSize(conf);

@julienledem julienledem Nov 17, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't get this change. the creation has been moved to ParquetFileWriter w = new ParquetFileWriter(... just a few lines bellow. Could you explain why we do this?

if (INFO) LOG.info("Parquet block size to " + blockSize);
int pageSize = getPageSize(conf);
Expand Down Expand Up @@ -379,7 +378,7 @@ public RecordWriter<Void, T> getRecordWriter(Configuration conf, Path file, Comp
init.getSchema(),
init.getExtraMetaData(),
blockSize, pageSize,
codecFactory.getCompressor(codec, pageSize),
w.getCodecFactory().getCompressor(codec, pageSize),
dictionaryPageSize,
enableDictionary,
validating,
Expand Down
3 changes: 1 addition & 2 deletions parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -267,8 +267,7 @@ public ParquetWriter(Path file, Configuration conf, WriteSupport<T> writeSupport
conf, schema, file, mode, blockSize, maxPaddingSize);
fileWriter.start();

CodecFactory codecFactory = new CodecFactory(conf);
CodecFactory.BytesCompressor compressor = codecFactory.getCompressor(compressionCodecName, 0);
CodecFactory.BytesCompressor compressor = fileWriter.getCodecFactory().getCompressor(compressionCodecName, 0);

@julienledem julienledem Nov 17, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

possibly what you really want is pass the codec factory as a parameter to this constructor.

this.writer = new InternalParquetRecordWriter<T>(
fileWriter,
writeSupport,
Expand Down