Make Spark jobs on Kubernetes more resilient when they encounter pod/executor failures.

[varunkatta]

This issue mainly covers the desired behavior of Spark jobs when executors fail during the job run.
The current behavior of a Spark application on Yarn when executors inadvertently fail is to

• first identify that executors failed
• taking corrective actions on executor failures. Example: Relaunching failed tasks on other healthy executors, launching new executors.

Link to the document which describes spark behavior on k8s and yarn when executors get killed during a spark job run. https://docs.google.com/document/d/1GX__jsCbeCw4RrUpHLqtpAzHwV82NQrgjz1dCCqqRes/edit?usp=sharing

This is an umbrella issue for issues
#134
#136

[foxish]

@varunkatta, Thanks for carrying out these experiments. In case of dynamic allocation, I think we all agree that we should detect loss of an executor and Spark should attempt to restart it by requesting a new executor in its place. As I understand it, the problem we're discussing here is what we should do in the static allocation case if we lose executors.

By specifying a RestartPolicy on the pod, we can ensure that as long as a pod exists in the APIServer, we will always attempt to restart it and recover from errors.
There are 3 relevant cases wherein the pod may get deleted completely from the API and thus never restart in k8s:

1. User intentionally deletes the pod from the API
2. Eviction occurs due to resource constraints
3. Node failure

In the first 2 cases, in my view, there is little value in relaunching said pod. In the third case of Node Failure, I think we could benefit from an understanding of how YARN behaves. While we do not need to match YARN/Mesos in this regard, it would be useful to understand it anyway. The test for this I imagine would be to knock down the node on which an executor is running in YARN.

[erikerlandson]

@foxish I'm wondering about a variation of "case 1" where some user deletes an executor pod of another user's submission. Possibly that just comes down to trusting authorization policies and namespaces, but in a multitenancy situation a "clumsy user" scenario seems conceivable.

[ash211]

1. Defending against "clumsy user" is something I would value for my deployments.
2. For the second item regarding eviction, I'd want to get a message in the driver logs that the executor was evicted at minimum (separate issue) and then I think I'd want the driver to attempt to start another pod. That executor pod may have been evicted from that node, but it might fit on another pod so should be tried to be rescheduled.
3. Node failure definitely needs reschedule.

One concern I have with specifying a RestartPolicy on the executor pod is that it will relaunch the executor with the same ID as the previous pod instance. See --executor-id $SPARK_EXECUTOR_ID in the executor Dockerfile. I think when YARN restarts a failed executor, it gets a new ID (as verified in the Spark UI). So we may not be able to use RestartPolicy on a pod unless we're sure we can either re-use executor IDs or update the executor ID during executor launch.

At first glance I lean towards avoiding the RestartPolicy and instead letting the driver launch new executors as they fail.

Curious what your thoughts are @mccheah

[mccheah]

Executor IDs is certainly a concern; if I recall correctly the driver expects every executor registration request to be with a unique ID that has never registered before. Therefore we may need to have the Spark code itself either handle duplicate executor IDs well or re-create the executors with the new IDs ourselves.

[varunkatta]

+1 Defending against "clumsy user" which is a variation of pod getting killed inadvertently. When pods disappear, driver needs to discover this and recover.
+1 on launching pods on a different node on pod eviction due to resource constraints on an existing node. This is the current Spark behavior on Yarn and a reasonable thing to expect from Spark on K8s.

On a node failure on Yarn, all executors which get killed typically get re-launched on a different node based on the application type. It happens in MR and Spark. Again, I expect K8s to support the same behavior.

All the above cases are not uniquely different, in my opinion. Executors might appear dead/unresponsive to a Driver for a variety of reasons - inadvertent kills, node going unhealthy, executor node hardware/network failures etc.. Driver should be able to reasonably recover from such failures by trying to schedule more executor pods in the cluster and finish the job without taking a huge performance hit. Spark does this reasonably well on Yarn and it is reasonable for users to expect the same on K8s.

Executor ids seen to be unique throughout the lifetime of a spark job. I didn't look at Spark code but this is what I observed for Spark on Yarn. As I killed executors, new executors were launched with Ids different from the lost ones (typically monotonically increasing). This also means any K8s restart policy should account for this.

[foxish]

On a node failure on Yarn, all executors which get killed typically get re-launched on a different node based on the application type. It happens in MR and Spark. Again, I expect K8s to support the same behavior.

Sounds reasonable if that is the expectation in general.
What's the expected Spark behavior when a "lost" executor (thinking intermittent network issues) re-establishes contact with the driver after we've created a replacement? Do we kill that executor?

[varunkatta]

I was curious about this myself. When the lost executor reestablishes contact, the Driver has an option

1. It could always consider it as a new executor, reset it and give it new tasks
2. consider it an orphaned executor and kill it, if it has already asked for more executors
3. if it hasn't already asked for new executors but in the process of asking new executors and found the lost executor then it could simply use the lost-n-found executor and abandon any ideas of summoning new executors.

It is difficult to test all scenarios manually due to the timing issues involved but it seems like 2) is what happens in practice on Yarn.

If you look at the logs closely

17:07:17 to 17:07:24 - driver and executor 1 realize they can't reach each other
17:10:21 - driver got a new executor 3
17:10:37 - lost executor 1 comes back to register with driver
17:10:38 - driver sends a kill to executor 1
17:11:41 - driver registered a new executor 4
17:12:23 - driver finished all tasks and the job finishes.

Job Client logs

.....
17/02/23 17:10:31 INFO Client: Application report for application_1487631764444_0042 (state: RUNNING)
17/02/23 17:10:32 INFO Client: Application report for application_1487631764444_0042 (state: 
....
17/02/23 17:12:20 INFO Client: Application report for application_1487631764444_0042 (state: RUNNING)
17/02/23 17:12:21 INFO Client: Application report for application_1487631764444_0042 (state: RUNNING)
17/02/23 17:12:23 INFO Client: Application report for application_1487631764444_0042 (state: FINISHED)
....

Driver logs

...
17/02/23 17:04:33 INFO BlockManagerMasterEndpoint: **Registering block manager cloudera-mgr-testing-2-n2.pepperdata.com:38077 with 530.3 MB RAM, BlockManagerId(1, cloudera-mgr-testing-2-n2.pepperdata.com, 38077)**
17/02/23 17:04:54 INFO YarnClusterSchedulerBackend: SchedulerBackend is ready for scheduling beginning after waiting maxRegisteredResourcesWaitingTime: 30000(ms)
....
17/02/23 17:07:24 WARN HeartbeatReceiver: Removing executor 1 with no recent heartbeats: 136839 ms exceeds timeout 120000 ms
17/02/23 17:07:24 ERROR YarnClusterScheduler: Lost executor 1 on cloudera-mgr-testing-2-n2.pepperdata.com: Executor heartbeat timed out after 136839 ms
17/02/23 17:07:24 WARN TaskSetManager: Lost task 2437.0 in stage 0.0 (TID 2437, cloudera-mgr-testing-2-n2.pepperdata.com): ExecutorLostFailure (executor 1 lost)
17/02/23 17:07:24 INFO BlockManagerMasterEndpoint: Trying to remove executor 1 from BlockManagerMaster.
....
17/02/23 17:10:21 INFO YarnClusterSchedulerBackend: Registered executor: AkkaRpcEndpointRef(Actor[akka.tcp://sparkExecutor@cloudera-mgr-testing-2-n2.pepperdata.com:37379/user/Executor#398187060]) with ID 3
17/02/23 17:10:21 INFO BlockManagerMasterEndpoint: Registering block manager cloudera-mgr-testing-2-n2.pepperdata.com:43213 with 530.3 MB RAM, BlockManagerId(3, cloudera-mgr-testing-2-n2.pepperdata.com, 43213)
17/02/23 17:10:37 INFO BlockManagerMasterEndpoint: Registering block manager cloudera-mgr-testing-2-n2.pepperdata.com:38077 with 530.3 MB RAM, BlockManagerId(1, cloudera-mgr-testing-2-n2.pepperdata.com, 38077)
....
17/02/23 17:10:39 ERROR YarnClusterScheduler: Lost executor 1 on cloudera-mgr-testing-2-n2.pepperdata.com: remote Rpc client disassociated
17/02/23 17:10:39 INFO BlockManagerMasterEndpoint: Trying to remove executor 1 from BlockManagerMaster.
....
17/02/23 17:11:41 INFO YarnClusterSchedulerBackend: Registered executor: AkkaRpcEndpointRef(Actor[akka.tcp://sparkExecutor@cloudera-mgr-testing-2-n3.pepperdata.com:43904/user/Executor#1402904646]) with ID 4
17/02/23 17:11:41 INFO BlockManagerMasterEndpoint: Registering block manager cloudera-mgr-testing-2-n3.pepperdata.com:43064 with 530.3 MB RAM, BlockManagerId(4, cloudera-mgr-testing-2-n3.pepperdata.com, 43064)
17/02/23 17:12:22 INFO ApplicationMaster: Unregistering ApplicationMaster with SUCCEEDED
17/02/23 17:12:22 INFO AMRMClientImpl: Waiting for application to be successfully unregistered.
....

Executor 1 Logs

17/02/23 17:04:30 INFO CoarseGrainedExecutorBackend: Registered signal handlers for [TERM, HUP, INT]
17/02/23 17:04:33 INFO CoarseGrainedExecutorBackend: Successfully registered with driver
17/02/23 17:04:33 INFO BlockManagerMaster: Trying to register BlockManager
17/02/23 17:04:33 INFO BlockManagerMaster: Registered BlockManager
17/02/23 17:04:33 INFO BlockManager: Registering executor with local external shuffle service.
....
17/02/23 17:05:10 INFO Executor: Running task 2434.0 in stage 0.0 (TID 2434)
17/02/23 17:05:10 INFO Executor: Finished task 2434.0 in stage 0.0 (TID 2434). 908 bytes result sent to driver
17/02/23 17:05:10 INFO CoarseGrainedExecutorBackend: Got assigned task 2437
17/02/23 17:05:10 INFO Executor: Running task 2437.0 in stage 0.0 (TID 2437)
17/02/23 17:05:10 INFO Executor: Finished task 2437.0 in stage 0.0 (TID 2437). 908 bytes result sent to driver
....
7/02/23 17:07:17 WARN AkkaRpcEndpointRef: Error sending message [message = Heartbeat(1,[Lscala.Tuple2;@54dc6a8,BlockManagerId(1, cloudera-mgr-testing-2-n2.pepperdata.com, 38077))] in 1 attempts
org.apache.spark.rpc.RpcTimeoutException: Futures timed out after [120 seconds]. This timeout is controlled by spark.rpc.askTimeout
        at org.apache.spark.rpc.RpcTimeout.org$apache$spark$rpc$RpcTimeout$$createRpcTimeoutException(RpcEnv.scala:214)
        at org.apache.spark.rpc.RpcTimeout$$anonfun$addMessageIfTimeout$1.applyOrElse(RpcEnv.scala:229)
....
17/02/23 17:10:37 INFO Executor: Told to re-register on heartbeat
17/02/23 17:10:37 INFO BlockManager: BlockManager re-registering with master
17/02/23 17:10:37 INFO BlockManagerMaster: Trying to register BlockManager
17/02/23 17:10:37 INFO BlockManagerMaster: Registered BlockManager
/02/23 17:10:37 INFO BlockManager: Reporting 2 blocks to the master.
17/02/23 17:10:38 ERROR CoarseGrainedExecutorBackend: RECEIVED SIGNAL 15: SIGTERM
17/02/23 17:10:38 INFO DiskBlockManager: Shutdown hook called
17/02/23 17:10:38 INFO ShutdownHookManager: Shutdown hook called
(END)
....

[foxish]

I created #151 which should have similar behavior to YARN that you described above.

[iyanuobidele]

My thoughts on this.

Sorry to get to this discussion late, but I think its non trivial to guarantee the number of executors in static allocation due to points mentioned by Anirudh.

Another thing I'll like to add to Anirudh's points above is getting (*) no of requested executors at the start of the job. In the case where a user has specified an unrealistic (constrained by resources available in the cluster) number of executors, the sparkJob gets some executors or none depending on the current state of things in the cluster. We would need to define the behavior to this in whatever logic we add.

But recall, Spark is designed to be resilient with job execution in the sense that, as long as one executor is registered to the master work will continue. It's also possible that this one resource could become a victim to 1, 2, or 3. I'm thinking we might need some kind of coordination between cause for executor loss/inexistence and state of available resources in kubernetes (might be a long shot). Because, in some cases, we really cannot and shouldn't do anything. I think, If we're going to do this (for static allocation), we would have to identify all these scenarios and have logic of the right/acceptable behavior.

[foxish]

Another thing I'll like to add to Anirudh's points above is getting (*) no of requested executors at the start of the job.

That is a good point. I was discussing this recently and I feel that it is more of a scheduling problem. Currently, we schedule pods in k8s, but in the eventual state we want to get to, we should be able to reserve resources and schedule entire SparkJobs (where the resource requirement of the SparkJob is the sum of the minimum resources requested by its individual components). The first step to explore this idea would be to write a custom scheduler for SparkJobs which exhibits this particular behavior. The scheduler would just be another pod. Eventually, we could generalize our learnings and have a batch scheduler living in upstream k8s.

[kimoonkim]

My understanding of Spark's expectation on YARN is that it will continuously do its best effort to run as many of the requested executors as possible.

Note this is different from expecting a guarantee. If a job can't get all the requested executors at the start or any other time point, it will still proceed with the smaller number of executors. It's just that it assumes YARN to launch more executors later if there's resource.

In the same line of thought, YARN should launch replacements of failed executors, because not doing so will effectively mean the maximum number of executors a job possibly gets is decreased by YARN.

This behavior is simple but handles some edge cases well. For example, if a driver got all of its executors preempted, it can simply wait around expecting some of them will come back. (Although, this is a source of potential deadlock. See #127) Otherwise, the driver has to shut itself down after waiting some configured time (as there is no point in continuing when you know all executors are dead and none coming back), which is tricky to implement.

Another example is that the driver does not need to establish the number of live executors at the job start thinking it will try to recover only them. It can simply replace only failed ones regardless of when they started, as long as the total number is at most the max.

[varunkatta]

Thanks Iyanu for chiming in. I am thinking we could start with something simple where Driver is making a best-case effort of procuring the said # of executors specified through the --num-executors option of spark-submit. Currently, the Driver doesn't seem to make any effort to get new executors when running executors are lost. Loosely speaking in K8s parlance, there should be a logical control loop somewhere which ensures the # running executors is always inching towards the # executors specified through --num-executors option. Spark jobs can make progress with reduced number of executors in case of executor failures but the job scheduler should strive to get to the user specified maximum. This should cover the 80% use-case. In situations where running executors are way below expected maximum, there are two choices - Let the job progress indefinitely with existing executors hoping for new executors to get scheduled. This behavior may not be the desired end user experience always. One pathological example here is; the job asks for 100 executors, gets a couple of executors and makes progress at a snail pace. Many users would simply kill the job at this point. - Let the job progress as long as # allocated executors is always greater than a specified user minimum (with best case efforts to reach the user-specified maximum). If the # allocated executors remains below the specified user-minimum longer than a time threshold, the job could terminate itself. We could probably come up with more policies than simply terminating itself.

…

On Fri, Feb 24, 2017 at 10:47 AM, Anirudh Ramanathan < ***@***.***> wrote: Another thing I'll like to add to Anirudh's points above is getting (*) no of requested executors at the start of the job. That is a good point. I was discussing this recently and I feel that it is more of a scheduling problem. Currently, we schedule *pods* in k8s, but in the eventual state we want to get to, we should be able to reserve resources and schedule entire SparkJobs (where the resource requirement of the SparkJob is the sum of the minimum resources requested by its individual components). The first step to explore this idea would be to write a custom scheduler for SparkJobs which exhibits this particular behavior. The scheduler would just be another pod <https://github.com/kubernetes/community/blob/master/contributors/design-proposals/multiple-schedulers.md>. Eventually, we could generalize our learnings and have a batch scheduler living in upstream k8s. — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#133 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAPRN-i6w3EvVCJ_g2Zh-kEauFtpX9T4ks5rfyXegaJpZM4MIVPQ> .

[duyanghao]

I am thinking we could start with something simple where Driver is making a best-case effort of procuring the said # of executors specified through the --num-executors option of spark-submit. Currently, the Driver doesn't seem to make any effort to get new executors when running executors are lost.

Loosely speaking in K8s parlance, there should be a logical control loop somewhere which ensures the # running executors is always inching towards the # executors specified through --num-executors option. Spark jobs can make progress with reduced number of executors in case of executor failures but the job scheduler should strive to get to the user specified maximum.
This should cover the 80% use-case.

@foxish @varunkatta that what user really needs

The executor aborts easily for lots of reasons in practice,and causing driver abnormally shut down,which is intolerant.

[foxish]

@duyanghao It should already be doing that. The executor allocation thread added as part of dynamic allocation does this already.