[#1824] feat(spark): Support map side combine of shuffle writer #1825
Conversation
| if (isCombine) { | ||
| createCombiner = shuffleDependency.aggregator().get().createCombiner(); | ||
| if (RssSparkConfig.toRssConf(sparkConf).get(RSS_CLIENT_MAP_SIDE_COMBINE_ENABLED)) { | ||
| iterator = shuffleDependency.aggregator().get().combineValuesByKey(records, taskContext); |
There was a problem hiding this comment.
Do we need to check the existence of shuffleDependency.aggregator() ?
There was a problem hiding this comment.
BTW, will this map combine cause the disk/memory burden , especially on the tight disk space on k8s?
There was a problem hiding this comment.
Do we need to check the existence of shuffleDependency.aggregator() ?
It seems not, there is a relevant check in ShuffleDependency.
There was a problem hiding this comment.
BTW, will this map combine cause the disk/memory burden , especially on the tight disk space on k8s?
I think it is possible when there are spills. So I added a configuration to control whether it is enabled.
There was a problem hiding this comment.
Yeh. I got your point, just an extra attention.
For having this point implicitly, let's make this shown in this config description.
zuston
changed the title
| WriteAndReadMetricsSparkListener listener = new WriteAndReadMetricsSparkListener(); | ||
| spark.sparkContext().addSparkListener(listener); | ||
|
|
||
| Thread.sleep(4000); |
There was a problem hiding this comment.
hmmm, why so large sleep time?
It's not the problem of this pr, but I'm not a big fan of using sleep directly in the test. Could we replace it with something likes semaphore or similar.
There was a problem hiding this comment.
There was a problem hiding this comment.
Most of the test cases keep it, I guess it's to wait for the shuffle server to be ready, do we still need to remove it?
There was a problem hiding this comment.
You can left it as it is for now. However it will introduce a lot of unnecessary sleep time for integration tests, we should address it in a followup PRs.
| ConfigOptions.key("rss.client.mapSideCombine.enabled") | ||
| .booleanType() | ||
| .defaultValue(false) | ||
| .withDescription("Whether to enable map side combine of shuffle writer."); |
There was a problem hiding this comment.
Could you also update the user documentation about this config?
I believe the reason we didn't perform map side combine when writing is that the map side could be super lightweight and improve write throughput. The documentation should state that implication as well.
| sc.parallelize(data, 10).mapToPair(x -> new Tuple2<>(x % 10, 1)).reduceByKey((x, y) -> x + y) | ||
| .collect().stream() | ||
| .forEach(x -> result.put(x._1 + "-result-value", x._2)); |
There was a problem hiding this comment.
One more thing, could we add more test cases to demonstrate the map-side combine.
There was a problem hiding this comment.
This test uses map side combine by default in vanilla spark implementation. We checked that rss behaves same as vanilla spark.
There was a problem hiding this comment.
I mean more RDD calls that require map side combine other than reduceByKey.
There was a problem hiding this comment.
I mean more RDD calls that require map side combine other than
reduceByKey.
Got it, I will add tests for other operators
| } | ||
|
|
||
| // check map side combine | ||
| assertEquals(100L, result.get("0-write-records")); |
There was a problem hiding this comment.
When map side combine is in effect, the shuffle write records of stage 0 is 100, otherwise it is 1000.
|
Would you mind adding the new config into the markdown documentation? |
|
@advancedxy @rickyma Thank you for your review , I will add doc later. |
…apache#1825) Support map side combine of shuffle write Fix: apache#1824 Yes, support new shuffle writer behavior. Added integration test
What changes were proposed in this pull request?
Support map side combine of shuffle writer
Why are the changes needed?
Fix: #1824
Does this PR introduce any user-facing change?
Yes, support new shuffle writer behavior.
How was this patch tested?
Added integration test