Code style, revert default minRegisteredRatio of yarn to 0, driver ge…

…t --num-executors in yarn/alpha
MediaV · Jun 20, 2014 · 6cfb9ec · 6cfb9ec
1 parent 812c33c
commit 6cfb9ec
Show file tree

Hide file tree

Showing 7 changed files with 10 additions and 18 deletions.
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskScheduler.scala
@@ -54,5 +54,5 @@ private[spark] trait TaskScheduler {
 
   // Get the default level of parallelism to use in the cluster, as a hint for sizing jobs.
   def defaultParallelism(): Int
-  def waitBackendReady(): Unit = {return}
+  def waitBackendReady(): Unit = { return }
 }
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala
@@ -431,12 +431,13 @@ private[spark] class TaskSchedulerImpl(
 
   // By default, rack is unknown
   def getRackForHost(value: String): Option[String] = None
-  override def waitBackendReady():Unit={
-    if(backend.isReady){
+
+  override def waitBackendReady(): Unit = {
+    if (backend.isReady) {
       return
     }
-    while(!backend.isReady){
-      synchronized{
+    while (!backend.isReady) {
+      synchronized {
         this.wait(100)
       }
     }

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
@@ -254,14 +254,14 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, actorSystem: A
   }
 
   override def isReady(): Boolean = {
-    if (ready){
+    if (ready) {
       return true
     }
     if ((System.currentTimeMillis() - createTime) >= maxRegisteredWaitingTime) {
       ready = true
       return true
     }
-    return false
+    false
   }
 }
 

diff --git a/yarn/alpha/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala b/yarn/alpha/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
@@ -184,6 +184,7 @@ class ApplicationMaster(args: ApplicationMasterArguments, conf: Configuration,
 
   private def startUserClass(): Thread = {
     logInfo("Starting the user JAR in a separate Thread")
+    System.setProperty("spark.executor.instances", args.numExecutors.toString)
     val mainMethod = Class.forName(
       args.userClass,
       false /* initialize */ ,

diff --git a/...common/src/main/scala/org/apache/spark/scheduler/cluster/YarnClientSchedulerBackend.scala b/...common/src/main/scala/org/apache/spark/scheduler/cluster/YarnClientSchedulerBackend.scala
@@ -78,11 +78,6 @@ private[spark] class YarnClientSchedulerBackend(
     logDebug("ClientArguments called with: " + argsArrayBuf)
     val args = new ClientArguments(argsArrayBuf.toArray, conf)
     totalExecutors.set(args.numExecutors)
-    // reset default minRegisteredRatio for yarn mode
-    if (minRegisteredRatio == 0) {
-      minRegisteredRatio = 0.9
-      ready = false
-    }
     client = new Client(args, conf)
     appId = client.runApp()
     waitForApp()

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
@@ -47,10 +47,5 @@ private[spark] class YarnClusterSchedulerBackend(
       .foreach { case (optName, envVar, sysProp) => addArg(optName, envVar, sysProp, argsArrayBuf) }
     val args = new ApplicationMasterArguments(argsArrayBuf.toArray)
     totalExecutors.set(args.numExecutors)
-    // reset default minRegisteredRatio for yarn mode
-    if (minRegisteredRatio == 0) {
-      minRegisteredRatio = 0.9
-      ready = false
-    }
   }
 }
diff --git a/yarn/stable/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala b/yarn/stable/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
@@ -164,7 +164,7 @@ class ApplicationMaster(args: ApplicationMasterArguments, conf: Configuration,
 
   private def startUserClass(): Thread = {
     logInfo("Starting the user JAR in a separate Thread")
-    System.setProperty("spark.executor.instances",args.numExecutors.toString)
+    System.setProperty("spark.executor.instances", args.numExecutors.toString)
     val mainMethod = Class.forName(
       args.userClass,
       false,