Bug fix: numExecutors wasn't set & use constant DEFAULT_NUMBER_EXECUTORS

MediaV · Jun 27, 2014 · c6f0522 · c6f0522
1 parent 4d6d847
commit c6f0522
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 7 deletions.
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala
@@ -51,8 +51,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, actorSystem: A
   private val timeout = AkkaUtils.askTimeout(conf)
   private val akkaFrameSize = AkkaUtils.maxFrameSizeBytes(conf)
   // Submit tasks only after (registered executors / total expected executors) 
-  // is equal to at least this value.
+  // is equal to at least this value, that is double between 0 and 1.
   var minRegisteredRatio = conf.getDouble("spark.scheduler.minRegisteredExecutorsRatio", 0)
+  if (minRegisteredRatio > 1) minRegisteredRatio = 1
   // Whatever minRegisteredExecutorsRatio is arrived, submit tasks after the time(milliseconds).
   val maxRegisteredWaitingTime =
     conf.getInt("spark.scheduler.maxRegisteredExecutorsWaitingTime", 30000)

diff --git a/yarn/common/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMasterArguments.scala b/yarn/common/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMasterArguments.scala
@@ -26,7 +26,7 @@ class ApplicationMasterArguments(val args: Array[String]) {
   var userArgs: Seq[String] = Seq[String]()
   var executorMemory = 1024
   var executorCores = 1
-  var numExecutors = 2
+  var numExecutors = ApplicationMasterArguments.DEFAULT_NUMBER_EXECUTORS
 
   parseArgs(args.toList)
 
@@ -93,3 +93,7 @@ class ApplicationMasterArguments(val args: Array[String]) {
     System.exit(exitCode)
   }
 }
+
+object ApplicationMasterArguments {
+  val DEFAULT_NUMBER_EXECUTORS = 2
+}
diff --git a/...ommon/src/main/scala/org/apache/spark/scheduler/cluster/YarnClusterSchedulerBackend.scala b/...ommon/src/main/scala/org/apache/spark/scheduler/cluster/YarnClusterSchedulerBackend.scala
@@ -18,6 +18,7 @@
 package org.apache.spark.scheduler.cluster
 
 import org.apache.spark.SparkContext
+import org.apache.spark.deploy.yarn.ApplicationMasterArguments
 import org.apache.spark.scheduler.TaskSchedulerImpl
 import org.apache.spark.util.IntParam
 
@@ -28,12 +29,12 @@ private[spark] class YarnClusterSchedulerBackend(
 
   override def start() {
     super.start()
-    var numExecutors = 2
-    if (sc.getConf.contains("spark.executor.instances")) {
-      numExecutors = sc.getConf.getInt("spark.executor.instances", 2)
-    } else if (System.getenv("SPARK_EXECUTOR_INSTANCES") != null) {
-      IntParam.unapply(System.getenv("SPARK_EXECUTOR_INSTANCES")).map(_.toInt).getOrElse(2)
+    var numExecutors = ApplicationMasterArguments.DEFAULT_NUMBER_EXECUTORS
+    if (System.getenv("SPARK_EXECUTOR_INSTANCES") != null) {
+      numExecutors = IntParam.unapply(System.getenv("SPARK_EXECUTOR_INSTANCES")).getOrElse(numExecutors)
     }
+    // System property can override environment variable.
+    numExecutors = sc.getConf.getInt("spark.executor.instances", numExecutors)
     totalExpectedExecutors.set(numExecutors)
   }
 }