[
https://issues.apache.org/jira/browse/FLINK-14845?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16978153#comment-16978153
]
Yingjie Cao commented on FLINK-14845:
-------------------------------------
To verify how many improvements we could gain from compression. I give a POC
implementation which dose not make much abstraction and modifies the existing
InputGate and ResultPartition directly (the final result shows that the change
is modest). The main changes are summarized as follows.
# Copied the block compression utils from flink-table (should be shared by
table and runtime).
# Implement a BufferCompressor and a BufferDecompressor util which take in
Buffer and return Buffer based on the compression until from 1.
# Add a flag to Buffer to identify if it is compressed.
# Each ResultPartition contains a BufferCompressor and each SingleInputGate
contains a BufferDecompressor.
# In BoundedBlockingSubpartition, when writing a buffer out, the buffer will
be compressed (call BufferCompressor#compress) if the compression is enabled.
When written to disk, the compression flag is also written out.
# In SingleInputGate if a received Buffer is compressed (identified by flag
introduced by 3), then it will be decompressed (call
BufferDecompressor#decompress).
# Add a compression flag to BufferResponse to let the Buffer receiver known if
the Buffer is compressed (This additional flag can be avoid by adding a new
message type like CompressedBufferResponse which directly extends
BufferResponse and implements a isCompressed interface which always returns
true).
# There is a bad case in which the data size become even bigger after
compression, the BufferCompressor will handle that and the original
uncompressed buffer will be returned, which means that the receiver will
receive a mixture of compressed and uncompressed Buffer.
[~sewen] [~ykt836] [~lzljs3620320] What do you think?
> Introduce data compression to blocking shuffle.
> -----------------------------------------------
>
> Key: FLINK-14845
> URL: https://issues.apache.org/jira/browse/FLINK-14845
> Project: Flink
> Issue Type: Sub-task
> Components: Runtime / Network
> Reporter: Yingjie Cao
> Priority: Major
>
> Currently, blocking shuffle writer writes raw output data to disk without
> compression. For IO bounded scenario, this can be optimized by compressing
> the output data. It is better to introduce a compression mechanism and offer
> users a config option to let the user decide whether to compress the shuffle
> data. Actually, we hava implemented compression in our inner Flink version
> and here are some key points:
> 1. Where to compress/decompress?
> Compressing at upstream and decompressing at downstream.
> 2. Which thread do compress/decompress?
> Task threads do compress/decompress.
> 3. Data compression granularity.
> Per buffer.
> 4. How to handle that when data size become even bigger after compression?
> Give up compression in this case and introduce an extra flag to identify if
> the data was compressed, that is, the output may be a mixture of compressed
> and uncompressed data.
>
> We'd like to introduce blocking shuffle data compression to Flink if there
> are interests.
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
