[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r169420133 --- Diff: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala --- @@ -334,6 +336,10 @@ private[spark] class ExecutorAllocationManager( // If the new target has not changed, avoid sending a message to the cluster manager if (numExecutorsTarget < oldNumExecutorsTarget) { +// We lower the target number of executors but don't actively kill any yet. We do this --- End diff -- I'm not sure I follow this comment. From my reading of it, it's saying that you don't want to kill executors because you don't want to immediately get a new one to replace it. But how can that happen, if you're also lowering the target number? --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r169420544 --- Diff: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala --- @@ -455,7 +461,12 @@ private[spark] class ExecutorAllocationManager( val executorsRemoved = if (testing) { executorIdsToBeRemoved } else { - client.killExecutors(executorIdsToBeRemoved) + // We don't want to change our target number of executors, because we already did that + // when the task backlog decreased. Normally there wouldn't be any tasks running on these + // executors, but maybe the scheduler *just* decided to run a task there -- in that case, --- End diff -- Also don't follow this part. If `force = false`, doesn't that mean the executor won't be killed if tasks are running on it? So wouldn't `countFailures` be meaningless in that context? --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r169421040 --- Diff: core/src/main/scala/org/apache/spark/SparkContext.scala --- @@ -1643,7 +1646,10 @@ class SparkContext(config: SparkConf) extends Logging { def killExecutors(executorIds: Seq[String]): Boolean = { schedulerBackend match { case b: ExecutorAllocationClient => -b.killExecutors(executorIds, replace = false, force = true).nonEmpty +require(executorAllocationManager.isEmpty, --- End diff -- This is a developer api, so probably ok, but this is a change in behavior. Is it just not possible to support this with dynamic allocation? --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user squito commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r169436456 --- Diff: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala --- @@ -334,6 +336,10 @@ private[spark] class ExecutorAllocationManager( // If the new target has not changed, avoid sending a message to the cluster manager if (numExecutorsTarget < oldNumExecutorsTarget) { +// We lower the target number of executors but don't actively kill any yet. We do this --- End diff -- I was trying to answer a different question -- if we don't kill the executor now, why even bother lowering the target number? as that would be an alternative solution -- don't adjust the target number here at all, just wait until you kill the executors for being idle. (and really I'm just guessing at the logic.) lemme try to reword this some ... --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user squito commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r169437530 --- Diff: core/src/main/scala/org/apache/spark/SparkContext.scala --- @@ -1643,7 +1646,10 @@ class SparkContext(config: SparkConf) extends Logging { def killExecutors(executorIds: Seq[String]): Boolean = { schedulerBackend match { case b: ExecutorAllocationClient => -b.killExecutors(executorIds, replace = false, force = true).nonEmpty +require(executorAllocationManager.isEmpty, --- End diff -- What would calling this mean with dynamic allocation on? Note this api explicitly says its meant to adjust resource usage downwards. If you've got just one executor, and then you kill it, should your app sit with 0 executors? Or even if you've got 10 executors, and you kill one -- when is dynamic allocation allowed to bump the total back up? I can't think of useful clear semantics for this (though this is not necessary to fix the bug, I could pull this out and move to a discussion in a new jira) --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user squito commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r169438419 --- Diff: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala --- @@ -455,7 +461,12 @@ private[spark] class ExecutorAllocationManager( val executorsRemoved = if (testing) { executorIdsToBeRemoved } else { - client.killExecutors(executorIdsToBeRemoved) + // We don't want to change our target number of executors, because we already did that + // when the task backlog decreased. Normally there wouldn't be any tasks running on these + // executors, but maybe the scheduler *just* decided to run a task there -- in that case, --- End diff -- good point, I didn't look closely enough at the semantics of `force` --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r169452326 --- Diff: core/src/main/scala/org/apache/spark/SparkContext.scala --- @@ -1643,7 +1646,10 @@ class SparkContext(config: SparkConf) extends Logging { def killExecutors(executorIds: Seq[String]): Boolean = { schedulerBackend match { case b: ExecutorAllocationClient => -b.killExecutors(executorIds, replace = false, force = true).nonEmpty +require(executorAllocationManager.isEmpty, --- End diff -- I'm not sure why you'd use this with dynamic allocation, but it's been possible in the past. It's probably ok to change this though. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r170102582 --- Diff: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala --- @@ -334,6 +336,11 @@ private[spark] class ExecutorAllocationManager( // If the new target has not changed, avoid sending a message to the cluster manager if (numExecutorsTarget < oldNumExecutorsTarget) { +// We lower the target number of executors but don't actively kill any yet. Killing is +// controlled separately by an idle timeout. Its still helpful to reduce the target number +// in case an executor just happens to get lost (eg., bad hardware, or the cluster manager +// preempts it) -- in that case, there is no point in trying to immediately get a new +// executor, since we couldn't even use it yet. --- End diff -- s/couldn't/wouldn't --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r170102363 --- Diff: core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala --- @@ -55,18 +55,18 @@ private[spark] trait ExecutorAllocationClient { /** * Request that the cluster manager kill the specified executors. * - * When asking the executor to be replaced, the executor loss is considered a failure, and - * killed tasks that are running on the executor will count towards the failure limits. If no - * replacement is being requested, then the tasks will not count towards the limit. - * * @param executorIds identifiers of executors to kill - * @param replace whether to replace the killed executors with new ones, default false + * @param adjustTargetNumExecutors whether the target number of executors will be adjusted down + * after these executors have been killed + * @param countFailures if there are tasks running on the executors when they are killed, whether + * those failures be counted to task failure limits? --- End diff -- nit: "whether to count those failures toward task failure limits" --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r170103841 --- Diff: core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala --- @@ -55,18 +55,18 @@ private[spark] trait ExecutorAllocationClient { /** * Request that the cluster manager kill the specified executors. * - * When asking the executor to be replaced, the executor loss is considered a failure, and - * killed tasks that are running on the executor will count towards the failure limits. If no - * replacement is being requested, then the tasks will not count towards the limit. - * * @param executorIds identifiers of executors to kill - * @param replace whether to replace the killed executors with new ones, default false + * @param adjustTargetNumExecutors whether the target number of executors will be adjusted down + * after these executors have been killed + * @param countFailures if there are tasks running on the executors when they are killed, whether --- End diff -- I'm still a little confused about this parameter. If `force = false`, it's a no op. And all call sites I've seen seem to set this parameter to `false`. So is there something I'm missing? --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r170102063 --- Diff: core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala --- @@ -334,6 +336,11 @@ private[spark] class ExecutorAllocationManager( // If the new target has not changed, avoid sending a message to the cluster manager if (numExecutorsTarget < oldNumExecutorsTarget) { +// We lower the target number of executors but don't actively kill any yet. Killing is +// controlled separately by an idle timeout. Its still helpful to reduce the target number --- End diff -- nit: it's --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user squito commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r170383918 --- Diff: core/src/main/scala/org/apache/spark/ExecutorAllocationClient.scala --- @@ -55,18 +55,18 @@ private[spark] trait ExecutorAllocationClient { /** * Request that the cluster manager kill the specified executors. * - * When asking the executor to be replaced, the executor loss is considered a failure, and - * killed tasks that are running on the executor will count towards the failure limits. If no - * replacement is being requested, then the tasks will not count towards the limit. - * * @param executorIds identifiers of executors to kill - * @param replace whether to replace the killed executors with new ones, default false + * @param adjustTargetNumExecutors whether the target number of executors will be adjusted down + * after these executors have been killed + * @param countFailures if there are tasks running on the executors when they are killed, whether --- End diff -- whoops, I was supposed to set `countFailures = true` in `sc.killAndReplaceExecutors`, thanks for catching that. --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r170650225 --- Diff: core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala --- @@ -610,20 +611,21 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp val executorsToKill = knownExecutors .filter { id => !executorsPendingToRemove.contains(id) } .filter { id => force || !scheduler.isExecutorBusy(id) } - executorsToKill.foreach { id => executorsPendingToRemove(id) = !replace } + executorsToKill.foreach { id => executorsPendingToRemove(id) = !countFailures} --- End diff -- super nit: space before `}` --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user vanzin commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r170650217 --- Diff: core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala --- @@ -610,20 +611,21 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp val executorsToKill = knownExecutors .filter { id => !executorsPendingToRemove.contains(id) } .filter { id => force || !scheduler.isExecutorBusy(id) } - executorsToKill.foreach { id => executorsPendingToRemove(id) = !replace } + executorsToKill.foreach { id => executorsPendingToRemove(id) = !countFailures} --- End diff -- super nit: space before `}` --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user asfgit closed the pull request at: https://github.com/apache/spark/pull/20604 --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user squito commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r185005638 --- Diff: core/src/main/scala/org/apache/spark/SparkContext.scala --- @@ -1643,7 +1646,10 @@ class SparkContext(config: SparkConf) extends Logging { def killExecutors(executorIds: Seq[String]): Boolean = { schedulerBackend match { case b: ExecutorAllocationClient => -b.killExecutors(executorIds, replace = false, force = true).nonEmpty +require(executorAllocationManager.isEmpty, --- End diff -- My point in general is that the semantics of combining `SparkContext.killExecutors()` (which is a publicly visible function, which the end user can call) with dynamic allocation aren't well defined, and I have no idea what the behavior really should be. I was giving some examples of weird behavior. >> If you've got just one executor, and then you kill it, should your app sit with 0 executors? > if app sit with 0 executors, then pending tasks increase, which lead to ExecutorAllocationManager increases target number of executors. So, app will not always sit with 0 executors. Thats true -- but only *when* pending tasks increase. But if you've got 0 executors, how do you expect pending tasks to increase? That would only happen when another taskset gets submitted, but with no executors your spark program will probably just be blocked. In the other case, I'm just trying to point out strange interactions between user control and dynamic allocation control. Imagine this sequence: Dynamic Allocation: 1000 tasks, so 1000 executors User: I only want 10 executors, so let me tell spark to kill 990 of them ... ... another taskset is submitted to add 1 more task ... Dynamic Allocation: 1001 tasks, so 1001 executors User: ??? I set the target to 10 executors, what happened? > So, if we are not using ExecutorAllocationManager, allocation client will request requestedTotalExecutors = 0 number of executors to cluster manager (this is really terrible) hmm, from a quick look, I think you're right. it doesn't seem that using `sc.killExecutors()` doesn't make sense even with dynamic allocation off. I think `CoarseGrainedSchedulerBackend` should actually initiliaze `requestedTotalExecutors` with `SchedulerBackendUtils.getInitialTargetExecutorNumber` --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user Ngone51 commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r185159109 --- Diff: core/src/main/scala/org/apache/spark/SparkContext.scala --- @@ -1643,7 +1646,10 @@ class SparkContext(config: SparkConf) extends Logging { def killExecutors(executorIds: Seq[String]): Boolean = { schedulerBackend match { case b: ExecutorAllocationClient => -b.killExecutors(executorIds, replace = false, force = true).nonEmpty +require(executorAllocationManager.isEmpty, --- End diff -- Hi @squito , thanks for your reply. > but only *when* pending tasks increase. `ExecutorAllocationManager ` will check pending (or backlog) tasks periodically. So, we do not have to wait for *increment* actually. And for `Dynamic Allocation` & `User` case, yeah, that's hard to define. And I checked `SchedulerBackendUtils.getInitialTargetExecutorNumbe`, it set `DEFAULT_NUMBER_EXECUTORS` = 2. But, this is not consistent with `Master`, which set `executorLimit` to `Int.MaxValue` if we are not under dynamic allocation mode. Maybe we can just init `requestedTotalExecutors ` with `Int.MaxValue`(only when we are not under dynamic allocation mode). Or, we do not call `doRequestTotalExecutors` if we call `requestExecutors` or `killExecutors`, except `requestTotalExecutors`(only when we are not under dynamic allocation mode). --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user Ngone51 commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r186425765 --- Diff: core/src/main/scala/org/apache/spark/SparkContext.scala --- @@ -1643,7 +1646,10 @@ class SparkContext(config: SparkConf) extends Logging { def killExecutors(executorIds: Seq[String]): Boolean = { schedulerBackend match { case b: ExecutorAllocationClient => -b.killExecutors(executorIds, replace = false, force = true).nonEmpty +require(executorAllocationManager.isEmpty, --- End diff -- @squito any thoughts? --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] spark pull request #20604: [SPARK-23365][CORE] Do not adjust num executors w...
Github user Ngone51 commented on a diff in the pull request: https://github.com/apache/spark/pull/20604#discussion_r182086337 --- Diff: core/src/main/scala/org/apache/spark/SparkContext.scala --- @@ -1643,7 +1646,10 @@ class SparkContext(config: SparkConf) extends Logging { def killExecutors(executorIds: Seq[String]): Boolean = { schedulerBackend match { case b: ExecutorAllocationClient => -b.killExecutors(executorIds, replace = false, force = true).nonEmpty +require(executorAllocationManager.isEmpty, --- End diff -- Hi, @squito , I'm quite questioned about the cases: > If you've got just one executor, and then you kill it, should your app sit with 0 executors? if app sit with 0 executors, then pending tasks increase, which lead to `ExecutorAllocationManager` increases target number of executors. So, app will not always sit with 0 executors. > Or even if you've got 10 executors, and you kill one -- when is dynamic allocation allowed to bump the total back up? for this case, to be honest, I really do not get your point. But, it must blame my poor English. And, what will happens if we use this method without `ExecutorAllocationManager `? Or do we really need adjust TargetNumExecutors (set `adjustTargetNumExecutors = true` below) if we are not using `ExecutorAllocationManager `? see these several lines in `killExecutors()`: ``` if (adjustTargetNumExecutors) { requestedTotalExecutors = math.max(requestedTotalExecutors - executorsToKill.size, 0) ... doRequestTotalExecutors(requestedTotalExecutors) } ``` Set `adjustTargetNumExecutors = true` will change `requestedTotalExecutors` . And IIUC, `requestedTotalExecutors ` is only used in dynamic allocation mode. So, if we are not using `ExecutorAllocationManager `, allocation client will request `requestedTotalExecutors = 0` number of executors to cluster manager (this is really terrible). But, actually, app without `ExecutorAllocationManager ` do not have a limit requesting executors (in default). Actually, I think this series methods, including `killAndReplaceExecutor `, `requestExecutors`, etc, are designed with dynamic allocation mode. And if we still want use these methods while app do not use `ExecutorAllocationManager`, we should not change `requestedTotalExecutors `, or even not request cluster manager with a specific number. WDYT? --- - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org