[SPARK-10087] [CORE] Disable spark.shuffle.reduceLocality.enabled by default.

[yhuai]

https://issues.apache.org/jira/browse/SPARK-10087

In some cases, when spark.shuffle.reduceLocality.enabled is enabled, we are scheduling all reducers to the same executor (the cluster has plenty of resources). Changing spark.shuffle.reduceLocality.enabled to false resolve the problem.

Here is a little bit more information. For one of my query, all 200 reducers were scheduled to the same reducer and every reducer has about 800 KB input.

[rxin]

cc @shivaram

[shivaram]

Could you provide some more information about the map output ? The reducer locality should not kick in unless a certain map output location has more than 20% of the output data. How many map tasks were run and what were their output sizes ?

[shivaram]

cc @mateiz who has also been looking at this code recently

[yhuai]

The reduce stage had a 2-way join in it. The two map stages had 30 and 1 tasks, respectively. For the stage having 30 tasks, here is the screenshot of task info

For the stage having 1 task, here is the screenshot of task info

[shivaram]

Thanks for the info -- And just to confirm, is everything getting assigned to Executor ID 23 (10.0.145.27) in the reduce stage ?

[yhuai]

ah, sorry i missed the reducer stage's screenshot. Yes, executor 23 was the one got all reduce tasks.

[shivaram]

So my hypothesis right now is that the RDD in the reduce stage has two Shuffle dependencies and the first shuffle dependency happens to be the single map task stage -- so the locality preference ends up giving all the tasks to the single host.

Hmm so my guess is that we need to be able to differentiate among different shuffle dependencies ideally. Here is another suggestion: Can we turn this off if we have more than one shuffle dependency ? it should be pretty cheap to count that

[shivaram]

The diff I'm proposing is something like

+    val numShuffleDeps = rdd.dependencies.filter(_.isInstanceOf[ShuffleDependency[_, _, _]]).length
+
     // If the RDD has shuffle dependencies and shuffle locality is enabled, pick locations that
     // have at least REDUCER_PREF_LOCS_FRACTION of data as preferred locations
-    if (shuffleLocalityEnabled && rdd.partitions.length < SHUFFLE_PREF_REDUCE_THRESHOLD) {
+    if (numShuffleDeps == 1 && shuffleLocalityEnabled &&
+        rdd.partitions.length < SHUFFLE_PREF_REDUCE_THRESHOLD) {

[SparkQA]

Test build #41149 has finished for PR 8280 at commit f77e574.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[mateiz]

It does sound good to turn it off if there are multiple dependencies. However, an even better solution may be to move this into ShuffledRDD, so that we control where exactly it occurs.

BTW to make this robust, I'd also make it affect locality only if the amount of data sent to that task is substantial (say over 100 MB). Otherwise scheduling for locality based in 1-2 MB is unnecessary.

[mateiz]

BTW it may also be fine to turn it off by default for 1.5, but in general, with these things, there's not much point having them in the code if they're off by default. We get very little feedback on them and they increase the number of configurations we have to worry about. I'm generally more inclined to turn these on and expand their scope gradually.

[rxin]

Why don't we turn it on in master but off in 1.5? At this point in the 1.5 cycle, I'm worry about potential bugs this would cause after more fixes.

[shivaram]

But to Matei's point we don't get feedback if its on in the master branch as I guess many more people use a release. I think turning it off for the multiple dependency makes it strictly narrower than what we have now, so I'm not sure it will cause new bugs.

[rxin]

Sorry just too risky right now for 1.5.

[yhuai]

I created #8296 to change the default setting to false for branch 1.5.

[rxin]

Let's close this one.

@shivaram can you submit a proper fix for master?

[shivaram]

Ok - lets leave it on in master and I'll work with @mateiz on changes to move this to ShuffleRDD and capture more use cases. @yhuai could you put in the query you ran in the JIRA so we can test / track this ?

[yhuai]

@shivaram Sure. Just updated the JIRA description.

[mateiz]

@shivaram did you create a JIRA for making this affect only ShuffledRDD? I might do it as part of https://issues.apache.org/jira/browse/SPARK-9852, which I'm working on a patch on (just haven't sent it yet because it depends on another in-review PR).

[shivaram]

Not yet - I was hoping to keep SPARK-10087 open, but I guess thats closed now. Doing it as a part of SPARK-9852 sounds good to me. Let me know if you want me to review the other PR and unblock this etc.