[SPARK-32217] Plumb whether a worker would also be decommissioned along with executor #29032

agrawaldevesh · 2020-07-08T00:19:17Z

What changes were proposed in this pull request?

This PR is a giant plumbing PR that plumbs an ExecutorDecommissionInfo along
with the DecommissionExecutor message.

Why are the changes needed?

The primary motivation is to know whether a decommissioned executor
would also be loosing shuffle files -- and thus it is important to know
whether the host would also be decommissioned.

In the absence of this PR, the existing code assumes that decommissioning an executor does not loose the whole host with it, and thus does not clear the shuffle state if external shuffle service is enabled. While this may hold in some cases (like K8s decommissioning an executor pod, or YARN container preemption), it does not hold in others like when the cluster is managed by a Standalone Scheduler (Master). This is similar to the existing workerLost field in the ExecutorProcessLost message.

In the future, this ExecutorDecommissionInfo can be embellished for
knowing how long the executor has to live for scenarios like Cloud spot
kills (or Yarn preemption) and the like.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Tweaked an existing unit test in AppClientSuite

agrawaldevesh · 2020-07-10T04:24:47Z

@holdenk, @jiangxb1987 @cloud-fan @Ngone51 -- This PR is ready for your review please. Thanks !

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

Ngone51 · 2020-07-13T12:31:42Z

core/src/main/scala/org/apache/spark/scheduler/DecommissionInfo.scala

+ *                                 Used to know if the shuffle data might be lost too.
+ */
+private[spark]
+case class DecommissionInfo(message: String, workerAlsoDecommissioned: Boolean)


Maybe, workerDecommissioned?

We probably need a base class for decom or just extends ExecutorLossReason... Then, we could use the base class or ExecutorLossReason in TaskScheduler.executorDecommission. Therefore, TaskScheduler.executorDecommission could support other cluster managers as well.

DecommissionInfo is just an info struct. It's not the loss reason itself, so it shouldn't overload ExecutorLossReason.

I didn't get the part about TaskScheduler.executorDecommission supporting other cluster managers: TaskScheduler is about task scheduling not coarse grained executor scheduling. I think it should be totally unaware of whether the decommission signal came via Yarn or K8s or whatever other cluster manager we have.

But now, it looks like DecommissionInfo is specific to Standalone because of the terminology “worker”.

I agree that TaskScheduler should be unaware of where the decom signal comes. That's why I asked for the base decom class. However, it's still ok if DecommissionInfo could be applied to other clusters as well.

Fair enough. I can see why the "Worker" nomenclature is confusing. Although ExecutorProcessLost has a workerLost field that can (in theory) be set by any scheduler backend. I will rename/reword to make this independance clearer.

agrawaldevesh · 2020-07-13T20:03:43Z

Retest this please.

agrawaldevesh · 2020-07-13T20:07:25Z

Jenkins, retest this please

Ngone51 · 2020-07-15T10:15:49Z

ok to test

cloud-fan · 2020-07-15T17:51:50Z

ok to test

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala

core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala

core/src/test/scala/org/apache/spark/scheduler/ExternalClusterManagerSuite.scala

core/src/main/scala/org/apache/spark/scheduler/DecommissionInfo.scala

SparkQA · 2020-07-16T01:00:13Z

Test build #125909 has finished for PR 29032 at commit 090eecd.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class DecommissionInfo(message: String, isWorkerDecommissioned: Boolean)
case class DecommissionExecutor(executorId: String, decommissionInfo: DecommissionInfo)

SparkQA · 2020-07-16T05:28:31Z

Test build #125926 has finished for PR 29032 at commit cbd92ea.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-07-16T05:40:30Z

Test build #125925 has finished for PR 29032 at commit aa6aae9.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

core/src/main/scala/org/apache/spark/scheduler/TaskScheduler.scala

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

core/src/main/scala/org/apache/spark/deploy/client/StandaloneAppClient.scala

SparkQA · 2020-07-16T07:05:02Z

Test build #125930 has finished for PR 29032 at commit 514b5a0.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorDecommissionInfo(message: String, isHostDecommissioned: Boolean)
case class DecommissionExecutor(executorId: String, decommissionInfo: ExecutorDecommissionInfo)

cloud-fan · 2020-07-16T10:10:17Z

retest this please

SparkQA · 2020-07-16T13:32:27Z

Test build #125964 has finished for PR 29032 at commit 514b5a0.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorDecommissionInfo(message: String, isHostDecommissioned: Boolean)
case class DecommissionExecutor(executorId: String, decommissionInfo: ExecutorDecommissionInfo)

SparkQA · 2020-07-17T04:08:53Z

Test build #126011 has finished for PR 29032 at commit 9356fac.

This patch fails PySpark pip packaging tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorDecommissionInfo(message: String, isHostDecommissioned: Boolean)
case class DecommissionExecutor(executorId: String, decommissionInfo: ExecutorDecommissionInfo)

cloud-fan · 2020-07-17T04:46:41Z

retest this please

SparkQA · 2020-07-17T07:05:02Z

Test build #126029 has finished for PR 29032 at commit 9356fac.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorDecommissionInfo(message: String, isHostDecommissioned: Boolean)
case class DecommissionExecutor(executorId: String, decommissionInfo: ExecutorDecommissionInfo)

agrawaldevesh · 2020-07-18T05:54:11Z

@holdenk ... It would be great for you to review this PR. This PR stems from your suggestion of plumbing a "isWorkerLost" to the executor decommission message, as you suggested in #29014 (comment). It does not introduce any other semantic changes.

I am keeping this as a separate PR instead of folding it into #29014, to keep the latter easy to revert. This PR (#29032) shouldn't be introducing any semantic changes and thus should be less risky.

holdenk

Thanks for working on this. A few minor questions and nits for cleanup but I think it is really close :) Thanks for factoring this out into a seperate PR for simpler reviewing as well, that's always appreciated :)

core/src/main/scala/org/apache/spark/deploy/master/Master.scala

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

core/src/main/scala/org/apache/spark/scheduler/ExecutorDecommissionInfo.scala

SparkQA · 2020-07-21T05:49:03Z

Test build #126212 has finished for PR 29032 at commit 71633f0.

This patch fails PySpark pip packaging tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorDecommissionInfo(message: String, isHostDecommissioned: Boolean)
case class DecommissionExecutor(executorId: String, decommissionInfo: ExecutorDecommissionInfo)

agrawaldevesh · 2020-07-21T23:07:32Z

@holdenk, this is ready again for your review. Thanks !

holdenk · 2020-07-22T00:12:11Z

The pyspark packaging tests have generally been flaky so i think we don't have to worry about them given that the GH actions all passed. Let me do another pass through reveiw, although it looks like it's already got one approval :)

holdenk · 2020-07-22T00:21:21Z

Thanks for the updates @agrawaldevesh. This looks good to me. I don't see any pending conversations that are ongoing, so if no one has any concerns that they aren't ok carrying through to the follow on PR that you've already got open, I plan to merge this tomorrow.

This PR is a giant plumbing PR that plumbs an `ExecutorDecommissionInfo` along with the DecommissionExecutor message. The primary motivation is to know whether a decommissioned executor would also be loosing shuffle files -- and thus it is important to know whether the host would also be decommissioned. This is similar to the existing `workerLost` field in the `ExecutorProcessLost` message. In the future, this `ExecutorDecommissionInfo` can be embellished for knowing how long the executor has to live for scenarios like Cloud spot kills (or Yarn preemption) and the like.

SparkQA · 2020-07-22T07:05:01Z

Test build #126301 has finished for PR 29032 at commit 88060be.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorDecommissionInfo(message: String, isHostDecommissioned: Boolean)
case class DecommissionExecutor(executorId: String, decommissionInfo: ExecutorDecommissionInfo)

Ngone51 · 2020-07-22T14:00:03Z

retest this please.

core/src/main/scala/org/apache/spark/scheduler/ExecutorDecommissionInfo.scala

Ngone51 · 2020-07-22T14:08:38Z

LGTM

SparkQA · 2020-07-22T17:26:15Z

Test build #126338 has finished for PR 29032 at commit 88060be.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorDecommissionInfo(message: String, isHostDecommissioned: Boolean)
case class DecommissionExecutor(executorId: String, decommissionInfo: ExecutorDecommissionInfo)

agrawaldevesh · 2020-07-22T22:04:14Z

retest this please

SparkQA · 2020-07-23T00:42:27Z

Test build #126363 has finished for PR 29032 at commit 88060be.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ExecutorDecommissionInfo(message: String, isHostDecommissioned: Boolean)
case class DecommissionExecutor(executorId: String, decommissionInfo: ExecutorDecommissionInfo)

agrawaldevesh · 2020-07-23T02:29:13Z

@holdenk... this is ready to go in. Thanks !

holdenk · 2020-07-23T02:31:55Z

Cool I’ll try merge this after dinner, thanks everyone for working on and reviewing this 👍

cloud-fan · 2020-07-23T03:47:37Z

I have a general question (not related to this PR): We have deprecated the usage of "multi-workers on one host", so host and worker should be the same thing. However, in the codebase sometimes we use host and sometimes worker, do we have a general rule for the naming?

This PR LGTM, I just ask this question for curiosity.

holdenk · 2020-07-23T04:08:03Z

Merged, thanks everyone.

I don't know if we have a good principle around naming host v. worker with the deprecation.

agrawaldevesh · 2020-07-23T05:44:05Z

Merged, thanks everyone.

I don't know if we have a good principle around naming host v. worker with the deprecation.

Thanks @holdenk for championing and shepherding this change !!

@cloud-fan, I agree that this host <=> worker naming is confusing. It was especially confusing for a newcomer to me: ExecutorProcessLost has a field workerLost (renamed from the erstwhile slaveLost). I had to pick b/w Worker and Host in ExecutorDecommissionInfo.isHostDecommissioned and I went with Worker initially but later changed Host.

But in my mind, they are both the same: Basically there is one external shuffle service (if any) per worker (or host). The word worker might draw some unintended connotation to the Standalone mode so I refrained from using it. But I think its usage is quite pervasive already.

It would be nice to have consistent terminology around worker vs host vs node :-P. In the absence of any, I went with host here.

cloud-fan · 2020-07-23T05:48:27Z

I agree host is better as it's general to any deployment model. I think we should only use worker when it refers to the standalone worker. Maybe we can start following this principle from now on.

[SPARK-21040][CORE] Speculate tasks which are running on decommission executors This PR adds functionality to consider the running tasks on decommission executors based on some config. In spark-on-cloud , we sometimes already know that an executor won't be alive for more than fix amount of time. Ex- In AWS Spot nodes, once we get the notification, we know that a node will be gone in 120 seconds. So if the running tasks on the decommissioning executors may run beyond currentTime+120 seconds, then they are candidate for speculation. Currently when an executor is decommission, we stop scheduling new tasks on those executors but the already running tasks keeps on running on them. Based on the cloud, we might know beforehand that an executor won't be alive for more than a preconfigured time. Different cloud providers gives different timeouts before they take away the nodes. For Ex- In case of AWS spot nodes, an executor won't be alive for more than 120 seconds. We can utilize this information in cloud environments and take better decisions about speculating the already running tasks on decommission executors. Yes. This PR adds a new config "spark.executor.decommission.killInterval" which they can explicitly set based on the cloud environment where they are running. Added UT. Closes apache#28619 from prakharjain09/SPARK-21040-speculate-decommission-exec-tasks. Authored-by: Prakhar Jain <prakharjain09@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-20629][CORE][K8S] Copy shuffle data when nodes are being shutdown This pull request adds the ability to migrate shuffle files during Spark's decommissioning. The design document associated with this change is at https://docs.google.com/document/d/1xVO1b6KAwdUhjEJBolVPl9C6sLj7oOveErwDSYdT-pE . To allow this change the `MapOutputTracker` has been extended to allow the location of shuffle files to be updated with `updateMapOutput`. When a shuffle block is put, a block update message will be sent which triggers the `updateMapOutput`. Instead of rejecting remote puts of shuffle blocks `BlockManager` delegates the storage of shuffle blocks to it's shufflemanager's resolver (if supported). A new, experimental, trait is added for shuffle resolvers to indicate they handle remote putting of blocks. The existing block migration code is moved out into a separate file, and a producer/consumer model is introduced for migrating shuffle files from the host as quickly as possible while not overwhelming other executors. Recomputting shuffle blocks can be expensive, we should take advantage of our decommissioning time to migrate these blocks. This PR introduces two new configs parameters, `spark.storage.decommission.shuffleBlocks.enabled` & `spark.storage.decommission.rddBlocks.enabled` that control which blocks should be migrated during storage decommissioning. New unit test & expansion of the Spark on K8s decom test to assert that decommisioning with shuffle block migration means that the results are not recomputed even when the original executor is terminated. This PR is a cleaned-up version of the previous WIP PR I made apache#28331 (thanks to attilapiros for his very helpful reviewing on it :)). Closes apache#28708 from holdenk/SPARK-20629-copy-shuffle-data-when-nodes-are-being-shutdown-cleaned-up. Lead-authored-by: Holden Karau <hkarau@apple.com> Co-authored-by: Holden Karau <holden@pigscanfly.ca> Co-authored-by: “attilapiros” <piros.attila.zsolt@gmail.com> Co-authored-by: Attila Zsolt Piros <attilazsoltpiros@apiros-mbp16.lan> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-24266][K8S] Restart the watcher when we receive a version changed from k8s Restart the watcher when it failed with a HTTP_GONE code from the kubernetes api. Which means a resource version has changed. For more relevant information see here: fabric8io/kubernetes-client#1075 No Running spark-submit to a k8s cluster. Not sure how to make an automated test for this. If someone can help me out that would be great. Closes apache#28423 from stijndehaes/bugfix/k8s-submit-resource-version-change. Authored-by: Stijn De Haes <stijndehaes@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32217] Plumb whether a worker would also be decommissioned along with executor This PR is a giant plumbing PR that plumbs an `ExecutorDecommissionInfo` along with the DecommissionExecutor message. The primary motivation is to know whether a decommissioned executor would also be loosing shuffle files -- and thus it is important to know whether the host would also be decommissioned. In the absence of this PR, the existing code assumes that decommissioning an executor does not loose the whole host with it, and thus does not clear the shuffle state if external shuffle service is enabled. While this may hold in some cases (like K8s decommissioning an executor pod, or YARN container preemption), it does not hold in others like when the cluster is managed by a Standalone Scheduler (Master). This is similar to the existing `workerLost` field in the `ExecutorProcessLost` message. In the future, this `ExecutorDecommissionInfo` can be embellished for knowing how long the executor has to live for scenarios like Cloud spot kills (or Yarn preemption) and the like. No Tweaked an existing unit test in `AppClientSuite` Closes apache#29032 from agrawaldevesh/plumb_decom_info. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32199][SPARK-32198] Reduce job failures during decommissioning This PR reduces the prospect of a job loss during decommissioning. It fixes two holes in the current decommissioning framework: - (a) Loss of decommissioned executors is not treated as a job failure: We know that the decommissioned executor would be dying soon, so its death is clearly not caused by the application. - (b) Shuffle files on the decommissioned host are cleared when the first fetch failure is detected from a decommissioned host: This is a bit tricky in terms of when to clear the shuffle state ? Ideally you want to clear it the millisecond before the shuffle service on the node dies (or the executor dies when there is no external shuffle service) -- too soon and it could lead to some wastage and too late would lead to fetch failures. The approach here is to do this clearing when the very first fetch failure is observed on the decommissioned block manager, without waiting for other blocks to also signal a failure. Without them decommissioning a lot of executors at a time leads to job failures. The task scheduler tracks the executors that were decommissioned along with their `ExecutorDecommissionInfo`. This information is used by: (a) For handling a `ExecutorProcessLost` error, or (b) by the `DAGScheduler` when handling a fetch failure. No Added a new unit test `DecommissionWorkerSuite` to test the new behavior by exercising the Master-Worker decommissioning. I chose to add a new test since the setup logic was quite different from the existing `WorkerDecommissionSuite`. I am open to changing the name of the newly added test suite :-) - Should I add a feature flag to guard these two behaviors ? They seem safe to me that they should only get triggered by decommissioning, but you never know :-). Closes apache#29014 from agrawaldevesh/decom_harden. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32417] Fix flakyness of BlockManagerDecommissionIntegrationSuite This test tries to fix the flakyness of BlockManagerDecommissionIntegrationSuite. Make the block manager decommissioning test be less flaky An interesting failure happens when migrateDuring = true (and persist or shuffle is true): - We schedule the job with tasks on executors 0, 1, 2. - We wait 300 ms and decommission executor 0. - If the task is not yet done on executor 0, it will now fail because the block manager won't be able to save the block. This condition is easy to trigger on a loaded machine where the github checks run. - The task with retry on a different executor (1 or 2) and its shuffle blocks will land there. - No actual block migration happens here because the decommissioned executor technically failed before it could even produce a block. To remove the above race, this change replaces the fixed wait for 300 ms to wait for an actual task to succeed. When a task has succeeded, we know its blocks would have been written for sure and thus its executor would certainly be forced to migrate those blocks when it is decommissioned. The change always decommissions an executor on which a real task finished successfully instead of picking the first executor. Because the system may choose to schedule nothing on the first executor and instead run the two tasks on one executor. I have had bad luck with BlockManagerDecommissionIntegrationSuite and it has failed several times on my PRs. So fixing it. No, unit test only change. Github checks. Ran this test 100 times, 10 at a time in parallel in a script. Closes apache#29226 from agrawaldevesh/block-manager-decom-flaky. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-31197][CORE] Shutdown executor once we are done decommissioning Exit the executor when it has been asked to decommission and there is nothing left for it to do. This is a rebase of apache#28817 If we want to use decommissioning in Spark's own scale down we should terminate the executor once finished. Furthermore, in graceful shutdown it makes sense to release resources we no longer need if we've been asked to shutdown by the cluster manager instead of always holding the resources as long as possible. The decommissioned executors will exit and the end of decommissioning. This is sort of a user facing change, however decommissioning hasn't been in any releases yet. I changed the unit test to not send the executor exit message and still wait on the executor exited message. Closes apache#29211 from holdenk/SPARK-31197-exit-execs-redone. Authored-by: Holden Karau <hkarau@apple.com> Signed-off-by: Holden Karau <hkarau@apple.com> Connect decommissioning to dynamic scaling Because the mock always says there is an RDD we may replicate more than once, and now that there are independent threads Make Spark's dynamic allocation use decommissioning Track the decommissioning executors in the core dynamic scheduler so we don't scale down too low, update the streaming ExecutorAllocationManager to also delegate to decommission Fix up executor add for resource profile Fix our exiting and cleanup thread for better debugging next time. Cleanup the locks we use in decommissioning and clarify some more bits. Verify executors decommissioned, then killed by external external cluster manager are re-launched Verify some additional calls are not occuring in the executor allocation manager suite. Dont' close the watcher until the end of the test Use decommissionExecutors and set adjustTargetNumExecutors to false so that we can match the pattern for killExecutor/killExecutors bump numparts up to 6 Revert "bump numparts up to 6" This reverts commit daf96dd. Small coment & visibility cleanup CR feedback/cleanup Cleanup the merge [SPARK-21040][CORE] Speculate tasks which are running on decommission executors This PR adds functionality to consider the running tasks on decommission executors based on some config. In spark-on-cloud , we sometimes already know that an executor won't be alive for more than fix amount of time. Ex- In AWS Spot nodes, once we get the notification, we know that a node will be gone in 120 seconds. So if the running tasks on the decommissioning executors may run beyond currentTime+120 seconds, then they are candidate for speculation. Currently when an executor is decommission, we stop scheduling new tasks on those executors but the already running tasks keeps on running on them. Based on the cloud, we might know beforehand that an executor won't be alive for more than a preconfigured time. Different cloud providers gives different timeouts before they take away the nodes. For Ex- In case of AWS spot nodes, an executor won't be alive for more than 120 seconds. We can utilize this information in cloud environments and take better decisions about speculating the already running tasks on decommission executors. Yes. This PR adds a new config "spark.executor.decommission.killInterval" which they can explicitly set based on the cloud environment where they are running. Added UT. Closes apache#28619 from prakharjain09/SPARK-21040-speculate-decommission-exec-tasks. Authored-by: Prakhar Jain <prakharjain09@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-20629][CORE][K8S] Copy shuffle data when nodes are being shutdown This pull request adds the ability to migrate shuffle files during Spark's decommissioning. The design document associated with this change is at https://docs.google.com/document/d/1xVO1b6KAwdUhjEJBolVPl9C6sLj7oOveErwDSYdT-pE . To allow this change the `MapOutputTracker` has been extended to allow the location of shuffle files to be updated with `updateMapOutput`. When a shuffle block is put, a block update message will be sent which triggers the `updateMapOutput`. Instead of rejecting remote puts of shuffle blocks `BlockManager` delegates the storage of shuffle blocks to it's shufflemanager's resolver (if supported). A new, experimental, trait is added for shuffle resolvers to indicate they handle remote putting of blocks. The existing block migration code is moved out into a separate file, and a producer/consumer model is introduced for migrating shuffle files from the host as quickly as possible while not overwhelming other executors. Recomputting shuffle blocks can be expensive, we should take advantage of our decommissioning time to migrate these blocks. This PR introduces two new configs parameters, `spark.storage.decommission.shuffleBlocks.enabled` & `spark.storage.decommission.rddBlocks.enabled` that control which blocks should be migrated during storage decommissioning. New unit test & expansion of the Spark on K8s decom test to assert that decommisioning with shuffle block migration means that the results are not recomputed even when the original executor is terminated. This PR is a cleaned-up version of the previous WIP PR I made apache#28331 (thanks to attilapiros for his very helpful reviewing on it :)). Closes apache#28708 from holdenk/SPARK-20629-copy-shuffle-data-when-nodes-are-being-shutdown-cleaned-up. Lead-authored-by: Holden Karau <hkarau@apple.com> Co-authored-by: Holden Karau <holden@pigscanfly.ca> Co-authored-by: “attilapiros” <piros.attila.zsolt@gmail.com> Co-authored-by: Attila Zsolt Piros <attilazsoltpiros@apiros-mbp16.lan> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-24266][K8S] Restart the watcher when we receive a version changed from k8s Restart the watcher when it failed with a HTTP_GONE code from the kubernetes api. Which means a resource version has changed. For more relevant information see here: fabric8io/kubernetes-client#1075 No Running spark-submit to a k8s cluster. Not sure how to make an automated test for this. If someone can help me out that would be great. Closes apache#28423 from stijndehaes/bugfix/k8s-submit-resource-version-change. Authored-by: Stijn De Haes <stijndehaes@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32217] Plumb whether a worker would also be decommissioned along with executor This PR is a giant plumbing PR that plumbs an `ExecutorDecommissionInfo` along with the DecommissionExecutor message. The primary motivation is to know whether a decommissioned executor would also be loosing shuffle files -- and thus it is important to know whether the host would also be decommissioned. In the absence of this PR, the existing code assumes that decommissioning an executor does not loose the whole host with it, and thus does not clear the shuffle state if external shuffle service is enabled. While this may hold in some cases (like K8s decommissioning an executor pod, or YARN container preemption), it does not hold in others like when the cluster is managed by a Standalone Scheduler (Master). This is similar to the existing `workerLost` field in the `ExecutorProcessLost` message. In the future, this `ExecutorDecommissionInfo` can be embellished for knowing how long the executor has to live for scenarios like Cloud spot kills (or Yarn preemption) and the like. No Tweaked an existing unit test in `AppClientSuite` Closes apache#29032 from agrawaldevesh/plumb_decom_info. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32199][SPARK-32198] Reduce job failures during decommissioning This PR reduces the prospect of a job loss during decommissioning. It fixes two holes in the current decommissioning framework: - (a) Loss of decommissioned executors is not treated as a job failure: We know that the decommissioned executor would be dying soon, so its death is clearly not caused by the application. - (b) Shuffle files on the decommissioned host are cleared when the first fetch failure is detected from a decommissioned host: This is a bit tricky in terms of when to clear the shuffle state ? Ideally you want to clear it the millisecond before the shuffle service on the node dies (or the executor dies when there is no external shuffle service) -- too soon and it could lead to some wastage and too late would lead to fetch failures. The approach here is to do this clearing when the very first fetch failure is observed on the decommissioned block manager, without waiting for other blocks to also signal a failure. Without them decommissioning a lot of executors at a time leads to job failures. The task scheduler tracks the executors that were decommissioned along with their `ExecutorDecommissionInfo`. This information is used by: (a) For handling a `ExecutorProcessLost` error, or (b) by the `DAGScheduler` when handling a fetch failure. No Added a new unit test `DecommissionWorkerSuite` to test the new behavior by exercising the Master-Worker decommissioning. I chose to add a new test since the setup logic was quite different from the existing `WorkerDecommissionSuite`. I am open to changing the name of the newly added test suite :-) - Should I add a feature flag to guard these two behaviors ? They seem safe to me that they should only get triggered by decommissioning, but you never know :-). Closes apache#29014 from agrawaldevesh/decom_harden. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-32417] Fix flakyness of BlockManagerDecommissionIntegrationSuite This test tries to fix the flakyness of BlockManagerDecommissionIntegrationSuite. Make the block manager decommissioning test be less flaky An interesting failure happens when migrateDuring = true (and persist or shuffle is true): - We schedule the job with tasks on executors 0, 1, 2. - We wait 300 ms and decommission executor 0. - If the task is not yet done on executor 0, it will now fail because the block manager won't be able to save the block. This condition is easy to trigger on a loaded machine where the github checks run. - The task with retry on a different executor (1 or 2) and its shuffle blocks will land there. - No actual block migration happens here because the decommissioned executor technically failed before it could even produce a block. To remove the above race, this change replaces the fixed wait for 300 ms to wait for an actual task to succeed. When a task has succeeded, we know its blocks would have been written for sure and thus its executor would certainly be forced to migrate those blocks when it is decommissioned. The change always decommissions an executor on which a real task finished successfully instead of picking the first executor. Because the system may choose to schedule nothing on the first executor and instead run the two tasks on one executor. I have had bad luck with BlockManagerDecommissionIntegrationSuite and it has failed several times on my PRs. So fixing it. No, unit test only change. Github checks. Ran this test 100 times, 10 at a time in parallel in a script. Closes apache#29226 from agrawaldevesh/block-manager-decom-flaky. Authored-by: Devesh Agrawal <devesh.agrawal@gmail.com> Signed-off-by: Holden Karau <hkarau@apple.com> [SPARK-31197][CORE] Shutdown executor once we are done decommissioning Exit the executor when it has been asked to decommission and there is nothing left for it to do. This is a rebase of apache#28817 If we want to use decommissioning in Spark's own scale down we should terminate the executor once finished. Furthermore, in graceful shutdown it makes sense to release resources we no longer need if we've been asked to shutdown by the cluster manager instead of always holding the resources as long as possible. The decommissioned executors will exit and the end of decommissioning. This is sort of a user facing change, however decommissioning hasn't been in any releases yet. I changed the unit test to not send the executor exit message and still wait on the executor exited message. Closes apache#29211 from holdenk/SPARK-31197-exit-execs-redone. Authored-by: Holden Karau <hkarau@apple.com> Signed-off-by: Holden Karau <hkarau@apple.com> Connect decommissioning to dynamic scaling Because the mock always says there is an RDD we may replicate more than once, and now that there are independent threads Make Spark's dynamic allocation use decommissioning Track the decommissioning executors in the core dynamic scheduler so we don't scale down too low, update the streaming ExecutorAllocationManager to also delegate to decommission Fix up executor add for resource profile Fix our exiting and cleanup thread for better debugging next time. Cleanup the locks we use in decommissioning and clarify some more bits. Verify executors decommissioned, then killed by external external cluster manager are re-launched Verify some additional calls are not occuring in the executor allocation manager suite. Dont' close the watcher until the end of the test Use decommissionExecutors and set adjustTargetNumExecutors to false so that we can match the pattern for killExecutor/killExecutors bump numparts up to 6 Revert "bump numparts up to 6" This reverts commit daf96dd. Small coment & visibility cleanup CR feedback/cleanup Fix up the merge CR feedback, move adjustExecutors to a common utility function Exclude some non-public APIs Remove junk More CR feedback Fix adjustExecutors backport This test fails for me locally and from what I recall it's because we use a different method of resolving the bind address than upstream so disabling the test This test fails for me locally and from what I recall it's because we use a different method of resolving the bind address than upstream so disabling the test Cleanup and drop watcher changes from the backport

probot-autolabeler bot added the CORE label Jul 8, 2020

agrawaldevesh force-pushed the plumb_decom_info branch 2 times, most recently from d68c4ac to 429bc0c Compare July 8, 2020 00:55

agrawaldevesh changed the title ~~[WIP] Plumb whether a worker would also be decommissioned along with executor~~ [WIP][SPARK-32217] Plumb whether a worker would also be decommissioned along with executor Jul 8, 2020

agrawaldevesh mentioned this pull request Jul 8, 2020

[SPARK-32199][SPARK-32198] Reduce job failures during decommissioning #29014

Closed

agrawaldevesh changed the title ~~[WIP][SPARK-32217] Plumb whether a worker would also be decommissioned along with executor~~ [SPARK-32217] Plumb whether a worker would also be decommissioned along with executor Jul 8, 2020

Ngone51 reviewed Jul 13, 2020

View reviewed changes

agrawaldevesh force-pushed the plumb_decom_info branch 2 times, most recently from 5121009 to 090eecd Compare July 13, 2020 23:10

cloud-fan reviewed Jul 15, 2020

View reviewed changes

core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Jul 15, 2020

View reviewed changes

core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Jul 15, 2020

View reviewed changes

core/src/test/scala/org/apache/spark/scheduler/ExternalClusterManagerSuite.scala Outdated Show resolved Hide resolved

cloud-fan reviewed Jul 15, 2020

View reviewed changes

core/src/main/scala/org/apache/spark/scheduler/DecommissionInfo.scala Outdated Show resolved Hide resolved

agrawaldevesh force-pushed the plumb_decom_info branch 3 times, most recently from cbd92ea to 514b5a0 Compare July 15, 2020 23:50

jiangxb1987 reviewed Jul 16, 2020

View reviewed changes

agrawaldevesh force-pushed the plumb_decom_info branch from 597b5ca to 9356fac Compare July 17, 2020 01:21

cloud-fan approved these changes Jul 17, 2020

View reviewed changes

holdenk reviewed Jul 20, 2020

View reviewed changes

agrawaldevesh force-pushed the plumb_decom_info branch from 9356fac to 71633f0 Compare July 21, 2020 02:55

agrawaldevesh force-pushed the plumb_decom_info branch from 71633f0 to 88060be Compare July 22, 2020 05:41

Ngone51 reviewed Jul 22, 2020

View reviewed changes

core/src/main/scala/org/apache/spark/scheduler/ExecutorDecommissionInfo.scala Show resolved Hide resolved

asfgit closed this in f8d29d3 Jul 23, 2020

agrawaldevesh mentioned this pull request Aug 12, 2020

[SPARK-31198][CORE] Use graceful decommissioning as part of dynamic scaling #29367

Closed

[SPARK-32217] Plumb whether a worker would also be decommissioned along with executor #29032

[SPARK-32217] Plumb whether a worker would also be decommissioned along with executor #29032

Conversation

agrawaldevesh commented Jul 8, 2020 • edited Loading

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

agrawaldevesh commented Jul 10, 2020

Ngone51 Jul 13, 2020

Choose a reason for hiding this comment

Ngone51 Jul 13, 2020 • edited Loading

Choose a reason for hiding this comment

agrawaldevesh Jul 13, 2020 • edited Loading

Choose a reason for hiding this comment

Ngone51 Jul 15, 2020

Choose a reason for hiding this comment

agrawaldevesh Jul 15, 2020

Choose a reason for hiding this comment

agrawaldevesh commented Jul 13, 2020

agrawaldevesh commented Jul 13, 2020

Ngone51 commented Jul 15, 2020

cloud-fan commented Jul 15, 2020

SparkQA commented Jul 16, 2020

SparkQA commented Jul 16, 2020

SparkQA commented Jul 16, 2020

SparkQA commented Jul 16, 2020

cloud-fan commented Jul 16, 2020

SparkQA commented Jul 16, 2020

SparkQA commented Jul 17, 2020

cloud-fan commented Jul 17, 2020

SparkQA commented Jul 17, 2020

agrawaldevesh commented Jul 18, 2020

holdenk left a comment

Choose a reason for hiding this comment

SparkQA commented Jul 21, 2020

agrawaldevesh commented Jul 21, 2020

holdenk commented Jul 22, 2020

holdenk commented Jul 22, 2020

SparkQA commented Jul 22, 2020

Ngone51 commented Jul 22, 2020

Ngone51 commented Jul 22, 2020

SparkQA commented Jul 22, 2020

agrawaldevesh commented Jul 22, 2020

SparkQA commented Jul 23, 2020

agrawaldevesh commented Jul 23, 2020

holdenk commented Jul 23, 2020

cloud-fan commented Jul 23, 2020

holdenk commented Jul 23, 2020

agrawaldevesh commented Jul 23, 2020

cloud-fan commented Jul 23, 2020

agrawaldevesh commented Jul 8, 2020 •

edited

Loading

Ngone51 Jul 13, 2020 •

edited

Loading

agrawaldevesh Jul 13, 2020 •

edited

Loading