Skip to content

[SPARK-10515] The total of number of executors in driver should be the same with in AM #8815

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
KaiXinXiaoLei wants to merge 2 commits into from
Closed

[SPARK-10515] The total of number of executors in driver should be the same with in AM #8815

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a5bd136
change numExecutorsTarget
Sep 18, 2015
dc035fd
Update ExecutorAllocationManager.scala
Sep 18, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -327,15 +327,15 @@ private[spark] class ExecutorAllocationManager(
* @return the number of additional executors actually requested.
*/
private def addExecutors(maxNumExecutorsNeeded: Int): Int = {
val numExistingExecutors = executorIds.size - executorsPendingToRemove.size
// Do not request more executors if it would put our target over the upper bound
if (numExecutorsTarget >= maxNumExecutors) {
if (numExistingExecutors >= maxNumExecutors) {
logDebug(s"Not adding executors because our current target total " +
s"is already $numExecutorsTarget (limit $maxNumExecutors)")
numExecutorsToAdd = 1
return 0
}

val oldNumExecutorsTarget = numExecutorsTarget
// There's no point in wasting time ramping up to the number of executors we already have, so
// make sure our target is at least as much as our current allocation:
numExecutorsTarget = math.max(numExecutorsTarget, executorIds.size)
Expand All @@ -346,7 +346,7 @@ private[spark] class ExecutorAllocationManager(
// Ensure that our target fits within configured bounds:
numExecutorsTarget = math.max(math.min(numExecutorsTarget, maxNumExecutors), minNumExecutors)

val delta = numExecutorsTarget - oldNumExecutorsTarget
val delta = numExecutorsTarget - numExistingExecutors

// If our target has not changed, do not send a message
// to the cluster manager and reset our exponential growth
Expand Down Expand Up @@ -406,6 +406,7 @@ private[spark] class ExecutorAllocationManager(
logInfo(s"Removing executor $executorId because it has been idle for " +
s"$executorIdleTimeoutS seconds (new desired total will be ${numExistingExecutors - 1})")
executorsPendingToRemove.add(executorId)
numExecutorsTarget = math.min(numExecutorsTarget, numExistingExecutors - 1)
true
} else {
logWarning(s"Unable to reach the cluster manager to kill executor $executorId!")
Expand Down