Uh oh!
There was an error while loading. Please reload this page.
[SPARK-24917][CORE] make chunk size configurable - #21933
[SPARK-24917][CORE] make chunk size configurable#21933vincent-grosbois wants to merge 2 commits into
Conversation
Add an option in Spark configuration to change the chunk size (which is by default 4 Mb). This would allow to bypass the issue mentionned in SPARK-24917 when fetching large partitions (a bit less than 2 Gb)
holdensmagicalunicorn
commented
Jul 31, 2018
@vincent-grosbois, thanks! I am a bot who has found some folks who might be able to help with the review:@JoshRosen, @rxin and @vanzin |
HyukjinKwon
commented
Aug 1, 2018
ok to test |
SparkQA
commented
Aug 1, 2018
Test build #93867 has finished for PR 21933 at commit
|
maropu
commented
Aug 2, 2018
Can you add |
vincent-grosbois
commented
Aug 2, 2018
Hello, I updated the description and title |
maropu
commented
Aug 3, 2018
retest this please |
SparkQA
commented
Aug 3, 2018
Test build #94080 has finished for PR 21933 at commit
|
kiszk
commented
Aug 3, 2018
retest this please |
SparkQA
commented
Aug 3, 2018
Test build #94100 has finished for PR 21933 at commit
|
SparkQA
commented
Aug 3, 2018
Test build #94098 has finished for PR 21933 at commit
|
kiszk
commented
Aug 3, 2018
retest this please |
SparkQA
commented
Aug 3, 2018
Test build #94114 has finished for PR 21933 at commit
|
| // Whether to compress shuffle output temporarily spilled to disk | ||
| private[this] val compressShuffleSpill = conf.getBoolean("spark.shuffle.spill.compress", true) | ||
| // Size of the chunks to be used in the ChunkedByteBuffer | ||
| private[this] val chunkSizeMb = conf.getSizeAsMb("spark.memory.chunkSize", "4m").toInt |
There was a problem hiding this comment.
The name spark.memory.chunkSize looks too generic.
How about spark.memory.serializer.chunkSize or others?
There was a problem hiding this comment.
I renamed it to spark.memory.serializer.chunkSize
rename rename spark.memory.chunkSize to spark.memory.serializer.chunkSize
SparkQA
commented
Aug 3, 2018
Test build #94138 has finished for PR 21933 at commit
|
kiszk
commented
Aug 7, 2018
retest this please |
SparkQA
commented
Aug 7, 2018
Test build #94346 has finished for PR 21933 at commit
|
HyukjinKwon
commented
Aug 7, 2018
retest this please |
SparkQA
commented
Aug 7, 2018
Test build #94358 has finished for PR 21933 at commit
|
kiszk
commented
Aug 8, 2018
LGTM |
| // Whether to compress shuffle output temporarily spilled to disk | ||
| private[this] val compressShuffleSpill = conf.getBoolean("spark.shuffle.spill.compress", true) | ||
| // Size of the chunks to be used in the ChunkedByteBuffer | ||
| private[this] val chunkSizeMb = conf.getSizeAsMb("spark.memory.serializer.chunkSize", "4m").toInt |
There was a problem hiding this comment.
Why don't we do byteStringAsBytes and remove 1024 * 1024 below?
HyukjinKwon
commented
Aug 10, 2018
cc @squito too. |
HyukjinKwon
left a comment
There was a problem hiding this comment.
The issue and fix looks coherent and good to me.
squito
commented
Aug 10, 2018
Thanks for detailed analysis @vincent-grosbois . I agree with everything, but as you noted you won't hit this particular issue anymore with |
vincent-grosbois
commented
Aug 11, 2018
Not really! This would be a useful features to backport in all spark branches that will never benefit from an upgrade to netty 4.1.28. However I think for spark 2.4 and the master branch, the netty dependency will eventually be bumped to 4.1.28 or more, which would make this commit useless... |
srowen
commented
Oct 25, 2018
Yeah this can be closed; we updated to 4.1.30 |
Closesapache#22567Closesapache#18457Closesapache#21517Closesapache#21858Closesapache#22383Closesapache#19219Closesapache#22401Closesapache#22811Closesapache#20405Closesapache#21933Closesapache#22819 from srowen/ClosePRs. Authored-by: Sean Owen <sean.owen@databricks.com> Signed-off-by: Sean Owen <sean.owen@databricks.com>
Add an option in Spark configuration to change the chunk size (which is by default 4 Mb).
This would allow to bypass the issue mentionned in SPARK-24917 by allowing users to define larger chunks.
Explanation:
currently, using netty < 4.1.28 (before this patch netty/netty@9b95b8e), sending a ChunkedByteBuffer with more than 16 chunks over the network will trigger a "merge" of all the blocks into one big transient array that is then sent over the network. This is problematic as the total memory for all chunks can be high (2GB) and this would then trigger an allocation of 2GB to merge everything, which will create OOM errors.
A possibility to bypass this netty behavior is to make sure that they data is never split into more than 16 chunks. One way to do this is to create bigger chunks, which is currently fixed to 4MB. In this commit I'm allowing users to define bigger chunk sizes for their job, which allowed us to bypass this OOM error.
What changes were proposed in this pull request?
I'm introducing a configuration parameter to define the chunk size
How was this patch tested?
Tested on several spark jobs, the changes actually work and generates "chunks" of the indicated size