From ee7cf0e58dced03711dfc8e0a8da7a84375f551f Mon Sep 17 00:00:00 2001 From: Nishkam Ravi Date: Tue, 7 Jul 2015 17:08:28 -0700 Subject: [PATCH 01/13] Improved scheduling algorithm for executors --- .../apache/spark/deploy/master/Master.scala | 90 ++++++++++++------- 1 file changed, 56 insertions(+), 34 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index fccceb3ea528b..5fd59edaeff99 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -544,38 +544,60 @@ private[master] class Master( * has enough cores and memory. Otherwise, each executor grabs all the cores available on the * worker by default, in which case only one executor may be launched on each worker. */ - private def startExecutorsOnWorkers(): Unit = { - // Right now this is a very simple FIFO scheduler. We keep trying to fit in the first app - // in the queue, then the second app, etc. + + private[master] def scheduleExecutorsOnWorkers(app: ApplicationInfo, usableWorkers: Array[WorkerInfo], + spreadOutApps: Boolean): Array[Int] = { + val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1) + val memoryPerExecutor = app.desc.memoryPerExecutorMB + val numUsable = usableWorkers.length + val assignedCores = new Array[Int](numUsable) // Number of cores to give to each worker + val assignedMemory = new Array[Int](numUsable) // Amount of memory to give to each worker + var toAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum) + var pos = 0 if (spreadOutApps) { - // Try to spread out each app among all the workers, until it has all its cores - for (app <- waitingApps if app.coresLeft > 0) { - val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE) - .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB && - worker.coresFree >= app.desc.coresPerExecutor.getOrElse(1)) - .sortBy(_.coresFree).reverse - val numUsable = usableWorkers.length - val assigned = new Array[Int](numUsable) // Number of cores to give on each node - var toAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum) - var pos = 0 - while (toAssign > 0) { - if (usableWorkers(pos).coresFree - assigned(pos) > 0) { - toAssign -= 1 - assigned(pos) += 1 - } - pos = (pos + 1) % numUsable - } - // Now that we've decided how many cores to give on each node, let's actually give them - for (pos <- 0 until numUsable if assigned(pos) > 0) { - allocateWorkerResourceToExecutors(app, assigned(pos), usableWorkers(pos)) + // Try to spread out executors among workers (sparse scheduling) + while (toAssign > 0) { + if (usableWorkers(pos).coresFree - assignedCores(pos) >= coresPerExecutor && + usableWorkers(pos).memoryFree - assignedMemory(pos) >= memoryPerExecutor) { + toAssign -= coresPerExecutor + assignedCores(pos) += coresPerExecutor + assignedMemory(pos) += memoryPerExecutor } + pos = (pos + 1) % numUsable } } else { - // Pack each app into as few workers as possible until we've assigned all its cores - for (worker <- workers if worker.coresFree > 0 && worker.state == WorkerState.ALIVE) { - for (app <- waitingApps if app.coresLeft > 0) { - allocateWorkerResourceToExecutors(app, app.coresLeft, worker) + // Pack executors into as few workers as possible (dense scheduling) + while (toAssign > 0) { + while (usableWorkers(pos).coresFree - assignedCores(pos) >= coresPerExecutor && + usableWorkers(pos).memoryFree - assignedMemory(pos) >= memoryPerExecutor) { + toAssign -= coresPerExecutor + assignedCores(pos) += coresPerExecutor + assignedMemory(pos) += memoryPerExecutor } + pos = (pos + 1) % numUsable + } + } + assignedCores + } + + /** + * Schedule and launch executors on workers + */ + private def startExecutorsOnWorkers(): Unit = { + // Right now this is a very simple FIFO scheduler. We keep trying to fit in the first app + // in the queue, then the second app, etc. + for (app <- waitingApps if app.coresLeft > 0) { + val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1) + val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE) + .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB && + worker.coresFree >= coresPerExecutor) + .sortBy(_.coresFree).reverse + val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps) + + // Now that we've decided how many cores to allocate on each worker, let's allocate them + var pos = 0 + for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) { + allocateWorkerResourceToExecutors(app, assignedCores(pos), coresPerExecutor, usableWorkers(pos)) } } } @@ -583,19 +605,19 @@ private[master] class Master( /** * Allocate a worker's resources to one or more executors. * @param app the info of the application which the executors belong to - * @param coresToAllocate cores on this worker to be allocated to this application + * @param assignedCores number of cores on this worker for this application + * @param coresPerExecutor number of cores per executor * @param worker the worker info */ private def allocateWorkerResourceToExecutors( app: ApplicationInfo, - coresToAllocate: Int, + assignedCores: Int, + coresPerExecutor: Int, worker: WorkerInfo): Unit = { - val memoryPerExecutor = app.desc.memoryPerExecutorMB - val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(coresToAllocate) - var coresLeft = coresToAllocate - while (coresLeft >= coresPerExecutor && worker.memoryFree >= memoryPerExecutor) { + + var numExecutors = assignedCores/coresPerExecutor + for (i <- 1 to numExecutors) { val exec = app.addExecutor(worker, coresPerExecutor) - coresLeft -= coresPerExecutor launchExecutor(worker, exec) app.state = ApplicationState.RUNNING } From 66362d56dee3a1e000d9e60c8aa80585baf29ffe Mon Sep 17 00:00:00 2001 From: nishkamravi2 Date: Wed, 8 Jul 2015 03:35:33 -0700 Subject: [PATCH 02/13] Update Master.scala --- .../src/main/scala/org/apache/spark/deploy/master/Master.scala | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index 5fd59edaeff99..9a333cceba16e 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -569,7 +569,8 @@ private[master] class Master( // Pack executors into as few workers as possible (dense scheduling) while (toAssign > 0) { while (usableWorkers(pos).coresFree - assignedCores(pos) >= coresPerExecutor && - usableWorkers(pos).memoryFree - assignedMemory(pos) >= memoryPerExecutor) { + usableWorkers(pos).memoryFree - assignedMemory(pos) >= memoryPerExecutor && + toAssign > 0) { toAssign -= coresPerExecutor assignedCores(pos) += coresPerExecutor assignedMemory(pos) += memoryPerExecutor From 2d6371c82b74b12d87b6d7ff93a2ec590d11516a Mon Sep 17 00:00:00 2001 From: Nishkam Ravi Date: Wed, 8 Jul 2015 13:36:08 -0700 Subject: [PATCH 03/13] Update Master.scala --- .../scala/org/apache/spark/deploy/master/Master.scala | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index 9a333cceba16e..098f64811e298 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -545,8 +545,8 @@ private[master] class Master( * worker by default, in which case only one executor may be launched on each worker. */ - private[master] def scheduleExecutorsOnWorkers(app: ApplicationInfo, usableWorkers: Array[WorkerInfo], - spreadOutApps: Boolean): Array[Int] = { + private[master] def scheduleExecutorsOnWorkers(app: ApplicationInfo, + usableWorkers: Array[WorkerInfo], spreadOutApps: Boolean): Array[Int] = { val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1) val memoryPerExecutor = app.desc.memoryPerExecutorMB val numUsable = usableWorkers.length @@ -580,9 +580,9 @@ private[master] class Master( } assignedCores } - + /** - * Schedule and launch executors on workers + * Schedule and launch executors on workers */ private def startExecutorsOnWorkers(): Unit = { // Right now this is a very simple FIFO scheduler. We keep trying to fit in the first app @@ -598,7 +598,8 @@ private[master] class Master( // Now that we've decided how many cores to allocate on each worker, let's allocate them var pos = 0 for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) { - allocateWorkerResourceToExecutors(app, assignedCores(pos), coresPerExecutor, usableWorkers(pos)) + allocateWorkerResourceToExecutors(app, assignedCores(pos), coresPerExecutor, + usableWorkers(pos)) } } } From 5d6a19cf71436d827a74e2749eea19394d29d7d7 Mon Sep 17 00:00:00 2001 From: nishkamravi2 Date: Thu, 9 Jul 2015 03:20:55 -0700 Subject: [PATCH 04/13] Update Master.scala (for the purpose of issuing a retest) --- core/src/main/scala/org/apache/spark/deploy/master/Master.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index 098f64811e298..546b010cc473a 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -605,7 +605,7 @@ private[master] class Master( } /** - * Allocate a worker's resources to one or more executors. + * Allocate a worker's resources to one or more executors * @param app the info of the application which the executors belong to * @param assignedCores number of cores on this worker for this application * @param coresPerExecutor number of cores per executor From c11c68993a0c8b41eac6d6c5b3e50a1d42b8116f Mon Sep 17 00:00:00 2001 From: nishkamravi2 Date: Thu, 9 Jul 2015 20:57:30 -0700 Subject: [PATCH 05/13] Update EventLoggingListenerSuite.scala Bug in the test --- .../org/apache/spark/scheduler/EventLoggingListenerSuite.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala index f681f21b6205e..5cb2d4225d281 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/EventLoggingListenerSuite.scala @@ -180,7 +180,7 @@ class EventLoggingListenerSuite extends SparkFunSuite with LocalSparkContext wit // into SPARK-6688. val conf = getLoggingConf(testDirPath, compressionCodec) .set("spark.hadoop.fs.defaultFS", "unsupported://example.com") - val sc = new SparkContext("local-cluster[2,2,512]", "test", conf) + val sc = new SparkContext("local-cluster[2,2,1024]", "test", conf) assert(sc.eventLogger.isDefined) val eventLogger = sc.eventLogger.get val eventLogPath = eventLogger.logPath From 40c8f9f3a58ac979bf8d476d4174b7c510d147c4 Mon Sep 17 00:00:00 2001 From: nishkamravi2 Date: Thu, 9 Jul 2015 21:02:33 -0700 Subject: [PATCH 06/13] Update Master.scala (to trigger retest) --- core/src/main/scala/org/apache/spark/deploy/master/Master.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index 546b010cc473a..098f64811e298 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -605,7 +605,7 @@ private[master] class Master( } /** - * Allocate a worker's resources to one or more executors + * Allocate a worker's resources to one or more executors. * @param app the info of the application which the executors belong to * @param assignedCores number of cores on this worker for this application * @param coresPerExecutor number of cores per executor From a06da760cbbe329a614bf58fbc02af49de353735 Mon Sep 17 00:00:00 2001 From: nishkamravi2 Date: Fri, 10 Jul 2015 02:36:33 -0700 Subject: [PATCH 07/13] Update Master.scala --- core/src/main/scala/org/apache/spark/deploy/master/Master.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index 098f64811e298..546b010cc473a 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -605,7 +605,7 @@ private[master] class Master( } /** - * Allocate a worker's resources to one or more executors. + * Allocate a worker's resources to one or more executors * @param app the info of the application which the executors belong to * @param assignedCores number of cores on this worker for this application * @param coresPerExecutor number of cores per executor From adec84be85e1a8a6d9a56fff044d1397a23370ba Mon Sep 17 00:00:00 2001 From: Nishkam Ravi Date: Fri, 17 Jul 2015 00:22:59 -0700 Subject: [PATCH 08/13] Update Master.scala --- .../apache/spark/deploy/master/Master.scala | 28 +++++++++++++------ 1 file changed, 19 insertions(+), 9 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index 546b010cc473a..f443dc672659f 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -533,6 +533,7 @@ private[master] class Master( /** * Schedule executors to be launched on the workers. + * Returns an array containing number of cores assigned to each worker. * * There are two modes of launching executors. The first attempts to spread out an application's * executors on as many workers as possible, while the second does the opposite (i.e. launch them @@ -543,10 +544,18 @@ private[master] class Master( * multiple executors from the same application may be launched on the same worker if the worker * has enough cores and memory. Otherwise, each executor grabs all the cores available on the * worker by default, in which case only one executor may be launched on each worker. + * + * It is important to allocate coresPerExecutor on each worker at a time (instead of 1 core + * at a time). Consider the following example: cluster has 4 workers with 16 cores each. + * User requests 3 executors (spark.cores.max = 48, spark.executor.cores = 16). If 1 core is + * allocated at a time, 12 cores from each worker would be assigned to each executor. + * Since 12 < 16, no executors would launch [SPARK-8881]. */ - - private[master] def scheduleExecutorsOnWorkers(app: ApplicationInfo, - usableWorkers: Array[WorkerInfo], spreadOutApps: Boolean): Array[Int] = { + private[master] def coresToAssign( + app: ApplicationInfo, + usableWorkers: Array[WorkerInfo], + spreadOutApps: Boolean): Array[Int] = { + // Default value for number of cores per executor is 1 val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1) val memoryPerExecutor = app.desc.memoryPerExecutorMB val numUsable = usableWorkers.length @@ -588,24 +597,25 @@ private[master] class Master( // Right now this is a very simple FIFO scheduler. We keep trying to fit in the first app // in the queue, then the second app, etc. for (app <- waitingApps if app.coresLeft > 0) { + // Default value for number of cores per executor is 1 val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1) val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE) .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB && worker.coresFree >= coresPerExecutor) .sortBy(_.coresFree).reverse - val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps) + val assignedCores = coresToAssign(app, usableWorkers, spreadOutApps) // Now that we've decided how many cores to allocate on each worker, let's allocate them var pos = 0 for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) { - allocateWorkerResourceToExecutors(app, assignedCores(pos), coresPerExecutor, - usableWorkers(pos)) + allocateWorkerResourceToExecutors( + app, assignedCores(pos), coresPerExecutor, usableWorkers(pos)) } } } /** - * Allocate a worker's resources to one or more executors + * Allocate a worker's resources to one or more executors. * @param app the info of the application which the executors belong to * @param assignedCores number of cores on this worker for this application * @param coresPerExecutor number of cores per executor @@ -616,8 +626,8 @@ private[master] class Master( assignedCores: Int, coresPerExecutor: Int, worker: WorkerInfo): Unit = { - - var numExecutors = assignedCores/coresPerExecutor + // If cores per executor is specified, then this division should have a remainder of zero + val numExecutors = assignedCores / coresPerExecutor for (i <- 1 to numExecutors) { val exec = app.addExecutor(worker, coresPerExecutor) launchExecutor(worker, exec) From f279cdf9d8af539febcc9df26af17cf5186adaa4 Mon Sep 17 00:00:00 2001 From: Nishkam Ravi Date: Fri, 17 Jul 2015 18:47:57 -0700 Subject: [PATCH 09/13] Update Master.scala --- .../apache/spark/deploy/master/Master.scala | 35 ++++++++++--------- 1 file changed, 19 insertions(+), 16 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index f443dc672659f..06b44c02f367a 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -551,24 +551,25 @@ private[master] class Master( * allocated at a time, 12 cores from each worker would be assigned to each executor. * Since 12 < 16, no executors would launch [SPARK-8881]. */ - private[master] def coresToAssign( + private[master] def scheduleExecutorsOnWorkers( app: ApplicationInfo, usableWorkers: Array[WorkerInfo], spreadOutApps: Boolean): Array[Int] = { - // Default value for number of cores per executor is 1 + // If the number of cores per executor is not specified, then we can just schedule + // 1 core at a time since we expect a single executor to be launched on each worker val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1) val memoryPerExecutor = app.desc.memoryPerExecutorMB val numUsable = usableWorkers.length val assignedCores = new Array[Int](numUsable) // Number of cores to give to each worker val assignedMemory = new Array[Int](numUsable) // Amount of memory to give to each worker - var toAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum) + var coresToAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum) var pos = 0 if (spreadOutApps) { // Try to spread out executors among workers (sparse scheduling) - while (toAssign > 0) { + while (coresToAssign > 0) { if (usableWorkers(pos).coresFree - assignedCores(pos) >= coresPerExecutor && usableWorkers(pos).memoryFree - assignedMemory(pos) >= memoryPerExecutor) { - toAssign -= coresPerExecutor + coresToAssign -= coresPerExecutor assignedCores(pos) += coresPerExecutor assignedMemory(pos) += memoryPerExecutor } @@ -576,11 +577,11 @@ private[master] class Master( } } else { // Pack executors into as few workers as possible (dense scheduling) - while (toAssign > 0) { + while (coresToAssign > 0) { while (usableWorkers(pos).coresFree - assignedCores(pos) >= coresPerExecutor && usableWorkers(pos).memoryFree - assignedMemory(pos) >= memoryPerExecutor && - toAssign > 0) { - toAssign -= coresPerExecutor + coresToAssign > 0) { + coresToAssign -= coresPerExecutor assignedCores(pos) += coresPerExecutor assignedMemory(pos) += memoryPerExecutor } @@ -597,13 +598,12 @@ private[master] class Master( // Right now this is a very simple FIFO scheduler. We keep trying to fit in the first app // in the queue, then the second app, etc. for (app <- waitingApps if app.coresLeft > 0) { - // Default value for number of cores per executor is 1 - val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1) + val coresPerExecutor: Option[Int] = app.desc.coresPerExecutor val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE) .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB && - worker.coresFree >= coresPerExecutor) + worker.coresFree >= coresPerExecutor.getOrElse(1)) .sortBy(_.coresFree).reverse - val assignedCores = coresToAssign(app, usableWorkers, spreadOutApps) + val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps) // Now that we've decided how many cores to allocate on each worker, let's allocate them var pos = 0 @@ -624,12 +624,15 @@ private[master] class Master( private def allocateWorkerResourceToExecutors( app: ApplicationInfo, assignedCores: Int, - coresPerExecutor: Int, + coresPerExecutor: Option[Int], worker: WorkerInfo): Unit = { - // If cores per executor is specified, then this division should have a remainder of zero - val numExecutors = assignedCores / coresPerExecutor + // If the number of cores per executor is specified, we divide the cores assigned + // to this worker evenly among the executors with no remainder. + // Otherwise, we launch a single executor that grabs all the assignedCores on this worker. + val numExecutors = coresPerExecutor.map { assignedCores / _ }.getOrElse(1) + val coresToAssign = coresPerExecutor.getOrElse(assignedCores) for (i <- 1 to numExecutors) { - val exec = app.addExecutor(worker, coresPerExecutor) + val exec = app.addExecutor(worker, coresToAssign) launchExecutor(worker, exec) app.state = ApplicationState.RUNNING } From 1daf25f07eed67be205993cec9b60ddf93b6fc59 Mon Sep 17 00:00:00 2001 From: Nishkam Ravi Date: Mon, 20 Jul 2015 23:26:50 -0700 Subject: [PATCH 10/13] Update Master.scala --- .../apache/spark/deploy/master/Master.scala | 34 ++++++++++++++----- 1 file changed, 26 insertions(+), 8 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index 06b44c02f367a..4e775583f9149 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -533,7 +533,7 @@ private[master] class Master( /** * Schedule executors to be launched on the workers. - * Returns an array containing number of cores assigned to each worker. + * Returns an array containing number of cores assigned to each worker (None if scheduling fails) * * There are two modes of launching executors. The first attempts to spread out an application's * executors on as many workers as possible, while the second does the opposite (i.e. launch them @@ -554,7 +554,7 @@ private[master] class Master( private[master] def scheduleExecutorsOnWorkers( app: ApplicationInfo, usableWorkers: Array[WorkerInfo], - spreadOutApps: Boolean): Array[Int] = { + spreadOutApps: Boolean): Option[Array[Int]] = { // If the number of cores per executor is not specified, then we can just schedule // 1 core at a time since we expect a single executor to be launched on each worker val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1) @@ -564,6 +564,7 @@ private[master] class Master( val assignedMemory = new Array[Int](numUsable) // Amount of memory to give to each worker var coresToAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum) var pos = 0 + var lastCoresToAssign = coresToAssign if (spreadOutApps) { // Try to spread out executors among workers (sparse scheduling) while (coresToAssign > 0) { @@ -574,6 +575,13 @@ private[master] class Master( assignedMemory(pos) += memoryPerExecutor } pos = (pos + 1) % numUsable + if (pos == 0) { + if (lastCoresToAssign == coresToAssign) { + logError("Not enough resources to schedule executors, please check configuration") + return None + } + lastCoresToAssign = coresToAssign + } } } else { // Pack executors into as few workers as possible (dense scheduling) @@ -586,9 +594,16 @@ private[master] class Master( assignedMemory(pos) += memoryPerExecutor } pos = (pos + 1) % numUsable + if (pos == 0) { + if (lastCoresToAssign == coresToAssign) { + logError("Not enough resources to schedule executors, please check configuration") + return None + } + lastCoresToAssign = coresToAssign + } } } - assignedCores + Some(assignedCores) } /** @@ -599,17 +614,20 @@ private[master] class Master( // in the queue, then the second app, etc. for (app <- waitingApps if app.coresLeft > 0) { val coresPerExecutor: Option[Int] = app.desc.coresPerExecutor + // Filter out workers that don't have enough resources to launch an executor val usableWorkers = workers.toArray.filter(_.state == WorkerState.ALIVE) .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB && worker.coresFree >= coresPerExecutor.getOrElse(1)) .sortBy(_.coresFree).reverse - val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps) + val assignedCores_ = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps) // Now that we've decided how many cores to allocate on each worker, let's allocate them - var pos = 0 - for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) { - allocateWorkerResourceToExecutors( - app, assignedCores(pos), coresPerExecutor, usableWorkers(pos)) + if (assignedCores_ != None) { + val assignedCores = assignedCores_.get + for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) { + allocateWorkerResourceToExecutors( + app, assignedCores(pos), coresPerExecutor, usableWorkers(pos)) + } } } } From 79084e83e3111f53657808532fc61b54cce0f39d Mon Sep 17 00:00:00 2001 From: Nishkam Ravi Date: Tue, 21 Jul 2015 01:04:16 -0700 Subject: [PATCH 11/13] Update Master.scala --- .../apache/spark/deploy/master/Master.scala | 21 +++++++------------ 1 file changed, 8 insertions(+), 13 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index 4e775583f9149..1c986defd693c 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -554,7 +554,7 @@ private[master] class Master( private[master] def scheduleExecutorsOnWorkers( app: ApplicationInfo, usableWorkers: Array[WorkerInfo], - spreadOutApps: Boolean): Option[Array[Int]] = { + spreadOutApps: Boolean): Array[Int] = { // If the number of cores per executor is not specified, then we can just schedule // 1 core at a time since we expect a single executor to be launched on each worker val coresPerExecutor = app.desc.coresPerExecutor.getOrElse(1) @@ -577,8 +577,7 @@ private[master] class Master( pos = (pos + 1) % numUsable if (pos == 0) { if (lastCoresToAssign == coresToAssign) { - logError("Not enough resources to schedule executors, please check configuration") - return None + return assignedCores } lastCoresToAssign = coresToAssign } @@ -596,14 +595,13 @@ private[master] class Master( pos = (pos + 1) % numUsable if (pos == 0) { if (lastCoresToAssign == coresToAssign) { - logError("Not enough resources to schedule executors, please check configuration") - return None + return assignedCores } lastCoresToAssign = coresToAssign } } } - Some(assignedCores) + assignedCores } /** @@ -619,15 +617,12 @@ private[master] class Master( .filter(worker => worker.memoryFree >= app.desc.memoryPerExecutorMB && worker.coresFree >= coresPerExecutor.getOrElse(1)) .sortBy(_.coresFree).reverse - val assignedCores_ = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps) + val assignedCores = scheduleExecutorsOnWorkers(app, usableWorkers, spreadOutApps) // Now that we've decided how many cores to allocate on each worker, let's allocate them - if (assignedCores_ != None) { - val assignedCores = assignedCores_.get - for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) { - allocateWorkerResourceToExecutors( - app, assignedCores(pos), coresPerExecutor, usableWorkers(pos)) - } + for (pos <- 0 until usableWorkers.length if assignedCores(pos) > 0) { + allocateWorkerResourceToExecutors( + app, assignedCores(pos), coresPerExecutor, usableWorkers(pos)) } } } From da0f491a930e8b7f7a761ff666afaac5cbb13aaa Mon Sep 17 00:00:00 2001 From: Nishkam Ravi Date: Tue, 21 Jul 2015 19:50:35 -0700 Subject: [PATCH 12/13] Update Master.scala --- .../apache/spark/deploy/master/Master.scala | 52 +++++++------------ 1 file changed, 20 insertions(+), 32 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index 1c986defd693c..192eb7844be94 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -533,7 +533,7 @@ private[master] class Master( /** * Schedule executors to be launched on the workers. - * Returns an array containing number of cores assigned to each worker (None if scheduling fails) + * Returns an array containing number of cores assigned to each worker. * * There are two modes of launching executors. The first attempts to spread out an application's * executors on as many workers as possible, while the second does the opposite (i.e. launch them @@ -563,41 +563,29 @@ private[master] class Master( val assignedCores = new Array[Int](numUsable) // Number of cores to give to each worker val assignedMemory = new Array[Int](numUsable) // Amount of memory to give to each worker var coresToAssign = math.min(app.coresLeft, usableWorkers.map(_.coresFree).sum) - var pos = 0 - var lastCoresToAssign = coresToAssign - if (spreadOutApps) { - // Try to spread out executors among workers (sparse scheduling) - while (coresToAssign > 0) { - if (usableWorkers(pos).coresFree - assignedCores(pos) >= coresPerExecutor && - usableWorkers(pos).memoryFree - assignedMemory(pos) >= memoryPerExecutor) { - coresToAssign -= coresPerExecutor - assignedCores(pos) += coresPerExecutor - assignedMemory(pos) += memoryPerExecutor - } - pos = (pos + 1) % numUsable - if (pos == 0) { - if (lastCoresToAssign == coresToAssign) { - return assignedCores - } - lastCoresToAssign = coresToAssign - } - } - } else { - // Pack executors into as few workers as possible (dense scheduling) - while (coresToAssign > 0) { - while (usableWorkers(pos).coresFree - assignedCores(pos) >= coresPerExecutor && - usableWorkers(pos).memoryFree - assignedMemory(pos) >= memoryPerExecutor && - coresToAssign > 0) { + var freeWorkers = (0 until numUsable).toIndexedSeq + + def canLaunchExecutor(pos: Int): Boolean = { + usableWorkers(pos).coresFree - assignedCores(pos) >= coresPerExecutor && + usableWorkers(pos).memoryFree - assignedMemory(pos) >= memoryPerExecutor + } + + while (coresToAssign > 0 && freeWorkers.nonEmpty) { + freeWorkers = freeWorkers.filter(canLaunchExecutor) + freeWorkers.foreach { pos => + var keepScheduling = true + while (keepScheduling && canLaunchExecutor(pos) && coresToAssign > 0) { coresToAssign -= coresPerExecutor assignedCores(pos) += coresPerExecutor assignedMemory(pos) += memoryPerExecutor - } - pos = (pos + 1) % numUsable - if (pos == 0) { - if (lastCoresToAssign == coresToAssign) { - return assignedCores + + // Spreading out an application means spreading out its executors across as + // many workers as possible. If we are not spreading out, then we should keep + // scheduling executors on this worker until we use all of its resources. + // Otherwise, just move on to the next worker. + if (spreadOutApps) { + keepScheduling = false } - lastCoresToAssign = coresToAssign } } } From b99809762f1c7863ff1e13250a77d0e8b65404ca Mon Sep 17 00:00:00 2001 From: nishkamravi2 Date: Wed, 22 Jul 2015 15:23:49 -0700 Subject: [PATCH 13/13] Update Master.scala --- .../main/scala/org/apache/spark/deploy/master/Master.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala index 192eb7844be94..ac74ebaa4d076 100644 --- a/core/src/main/scala/org/apache/spark/deploy/master/Master.scala +++ b/core/src/main/scala/org/apache/spark/deploy/master/Master.scala @@ -570,11 +570,11 @@ private[master] class Master( usableWorkers(pos).memoryFree - assignedMemory(pos) >= memoryPerExecutor } - while (coresToAssign > 0 && freeWorkers.nonEmpty) { + while (coresToAssign >= coresPerExecutor && freeWorkers.nonEmpty) { freeWorkers = freeWorkers.filter(canLaunchExecutor) freeWorkers.foreach { pos => var keepScheduling = true - while (keepScheduling && canLaunchExecutor(pos) && coresToAssign > 0) { + while (keepScheduling && canLaunchExecutor(pos) && coresToAssign >= coresPerExecutor) { coresToAssign -= coresPerExecutor assignedCores(pos) += coresPerExecutor assignedMemory(pos) += memoryPerExecutor