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

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.

Comments of #8280 provide more details of the symptom of this issue.

This PR changes the default setting of `spark.shuffle.reduceLocality.enabled` to `false` for branch 1.5.

Author: Yin Huai <[email protected]>

Closes #8296 from yhuai/setNumPartitionsCorrectly-branch1.5.

Author: yhuai

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

@@ -138,7 +138,7 @@ class DAGScheduler(
 
   // Flag to control if reduce tasks are assigned preferred locations
   private val shuffleLocalityEnabled =
-    sc.getConf.getBoolean("spark.shuffle.reduceLocality.enabled", true)
+    sc.getConf.getBoolean("spark.shuffle.reduceLocality.enabled", false)
   // Number of map, reduce tasks above which we do not assign preferred locations
   // based on map output sizes. We limit the size of jobs for which assign preferred locations
   // as computing the top locations by size becomes expensive.

core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala

@@ -909,7 +909,7 @@ class DAGSchedulerSuite
     assertDataStructuresEmpty()
   }
 
-  test("reduce tasks should be placed locally with map output") {
+  ignore("reduce tasks should be placed locally with map output") {
     // Create an shuffleMapRdd with 1 partition
     val shuffleMapRdd = new MyRDD(sc, 1, Nil)
     val shuffleDep = new ShuffleDependency(shuffleMapRdd, null)
@@ -929,7 +929,7 @@ class DAGSchedulerSuite
     assertDataStructuresEmpty()
   }
 
-  test("reduce task locality preferences should only include machines with largest map outputs") {
+  ignore("reduce task locality preferences should only include machines with largest map outputs") {
     val numMapTasks = 4
     // Create an shuffleMapRdd with more partitions
     val shuffleMapRdd = new MyRDD(sc, numMapTasks, Nil)