xkrogen commented on a change in pull request #28287:
URL: https://github.com/apache/spark/pull/28287#discussion_r447136222
##########
File path: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
##########
@@ -289,13 +290,23 @@ private[spark] class ExecutorAllocationManager(
s" tasksperexecutor: $tasksPerExecutor")
val maxNeeded = math.ceil(numRunningOrPendingTasks *
executorAllocationRatio /
tasksPerExecutor).toInt
- if (tasksPerExecutor > 1 && maxNeeded == 1 && pendingSpeculative > 0) {
+ val totalNeed = if (tasksPerExecutor > 1 && maxNeeded == 1 &&
pendingSpeculative > 0) {
// If we have pending speculative tasks and only need a single executor,
allocate one more
// to satisfy the locality requirements of speculation
maxNeeded + 1
} else {
maxNeeded
}
+
+ // Request additional executors to schedule the unschedulable tasks as well
+ if (numUnschedulables > 0) {
+ val maxNeededForUnschedulables = math.ceil(numUnschedulables *
executorAllocationRatio /
+ tasksPerExecutor).toInt
+ math.max(totalNeed,
executorMonitor.executorCountWithResourceProfile(rpId)) +
Review comment:
We discussed offline that using the current executor count can result in
too many executors getting allocated, if the scheduling loop runs between when
an executor is allocated and when tasks are scheduled on it (thus clearing out
`unschedulableTaskSets`). Did you come up with any solution for this? If not
and we think it's a small enough issue to ignore, let's at least add a comment
about this caveat.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
