[SPARK-13904][Scheduler]Add support for pluggable cluster manager #11723

hbhanawat · 2016-03-15T09:37:48Z

What changes were proposed in this pull request?

This commit adds support for pluggable cluster manager. And also allows a cluster manager to clean up tasks without taking the parent process down.

To plug a new external cluster manager, ExternalClusterManager trait should be implemented. It returns task scheduler and backend scheduler that will be used by SparkContext to schedule tasks. An external cluster manager is registered using the java.util.ServiceLoader mechanism (This mechanism is also being used to register data sources like parquet, json, jdbc etc.). This allows auto-loading implementations of ExternalClusterManager interface.

Currently, when a driver fails, executors exit using system.exit. This does not bode well for cluster managers that would like to reuse the parent process of an executor. Hence,

Moving system.exit to a function that can be overriden in subclasses of CoarseGrainedExecutorBackend.
Added functionality of killing all the running tasks in an executor.

How was this patch tested?

ExternalClusterManagerSuite.scala was added to test this patch.

…ws a cluster manager to clean up tasks without taking the parent process down. To plug a new external cluster manager, ExternalClusterManager trait should be implemented. It returns task scheduler and backend scheduler that will be used by SparkContext to schedule tasks. An external cluster manager is registered using the java.util.ServiceLoader mechanism (This mechanism is also being used to register data sources like parquet, json, jdbc etc.). This allows auto-loading implementations of ExternalClusterManager interface. Currently, when a driver fails, executors exit using system.exit. This does not bode well for cluster managers that would like to reuse the parent process of an executor. Hence, 1. Moving system.exit to a function that can be overriden in subclasses of CoarseGrainedExecutorBackend. 2. Added functionality of killing all the running tasks in an executor.

sarutak · 2016-03-16T02:48:32Z

ok to test.

SparkQA · 2016-03-16T05:04:01Z

Test build #53264 has finished for PR 11723 at commit 800834f.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- trait ExternalClusterManager

rxin · 2016-03-16T05:30:37Z

Hmm I'm not sure if it makes sense to create a public API for this at this point. This is directly exposing something that is very internal to the current implementation of Spark (SchedulerBackend), and these are by no means stable.

hbhanawat · 2016-03-16T09:09:59Z

@rxin Thanks for commenting.

Spark was designed such that it is agnostic to the underlying cluster manager (as long as it can acquire executor processes, and these communicate with each other). Since Spark is now being used in newer and different use cases, there is a need for allowing other cluster managers to manage spark components. One such use case is - embedding spark components like executor and driver inside another process which may be a datastore. This allows co-location of data and processing. Another use case would be using Spark like an application server (you might have heard about spark-jobserver). Spark's current design allows handling such use cases if the cluster manager supports it. Hence, IMO, it is meaningful to allow plugging in new cluster managers.

From code perspective, I think that even creation of TaskScheduler and SchedulerBackend for Yarn/Mesos/local mode should be done using a similar interface.

rxin · 2016-03-16T17:13:50Z

@hbhanawat I understand that. The problem is not whether you can find a single legitimate use case.

The introduction of every API always benefit something -- there is no argument about it. Otherwise nobody would be adding new APIs. The question is how many it benefits, and how the APIs can be maintained and evolved. You are effectively taking a bunch of private APIs that were never meant to be public and making them public. This approach is not maintainable.

rxin · 2016-03-16T17:46:36Z

@hbhanawat to be clear, I think we might be able to add this as a semi-private API and external resource managers can use, but with the understanding that this is tied to specific versions of Spark and might break from release over release. Exposing this as a stable public API based on my experience seeing how Spark has evolved in the last 5 years is not going to work.

hbhanawat · 2016-03-17T10:21:05Z

@rxin ok, I get it. I would make ExternalClusterManager as private[spark] and mark it as developer API. I think that should suffice.

SparkQA · 2016-03-17T14:16:43Z

Test build #53424 has finished for PR 11723 at commit e87c1e0.

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

…king ExternalClusterManager as DeveloperApi and making it as private[spark].

SparkQA · 2016-03-18T10:39:35Z

Test build #53525 has finished for PR 11723 at commit ae808d7.

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

hbhanawat · 2016-03-18T11:21:25Z

@rxin I have completed the changes. Please review.

hbhanawat · 2016-03-23T12:36:59Z

@rxin Any update? Any changes needed from my side?

tejasapatil · 2016-03-24T05:30:10Z

core/src/main/scala/org/apache/spark/SparkContext.scala

+      // exactly one registered manager
+      case head :: Nil => Some(head)
+      case Nil => None
+      case multipleMgrs => sys.error(s"Multiple Cluster Managers registered " +


Can you include the list of matching cluster managers in the message ?

tejasapatil · 2016-03-24T05:56:39Z

@rxin : I would really like to have this PR in trunk. As things stand, for anyone using their own scheduler, one has to maintain a patch over open source release to have that glue in Spark. Re API breaking over Spark releases : I agree that breaking APIs is bad. But it would be atleast better than the current model of dealing with this: doing a merge for every release.

rxin · 2016-03-24T06:17:37Z

Is this something Facebook needs too?

tejasapatil · 2016-03-24T07:32:32Z

@rxin : Yes !! At Facebook we are using an internal scheduler to run Spark executors. Maintaining an internal patch to have that "glue" and merging it against every Spark release can be avoided by this PR. Ideally we want to get to a place where there are no patches to maintain and everything is just pluggable. If we get there, testing RCs would be easier and we can help flagging issues early on (especially ones related to scalability).

rxin · 2016-03-24T07:35:23Z

OK - once you are done with your own review ping me. I will take a look at it again.

1. Fixed formatting issues 2. Added master url as part of the other functions of ExternalClusterManager

tejasapatil · 2016-03-24T15:43:52Z

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

+        setAppName("testcm").set("spark.driver.allowMultipleContexts", "true")
+    sc = new SparkContext(conf)
+    // check if the scheduler components are created
+    assert(sc.schedulerBackend.isInstanceOf[FakeSchedulerBackend])


Sorry I missed this in the last review comments yesterday. I thought that FakeSchedulerBackend was in this same file and you could rename it but now I see that its from some other place.

While reading, it feels odd to have Fake* and then Dummy* test classes. I am not sure about the whats followed in Spark codebase. Couple options:

rename Dummy* classes => Fake_. Move all the Fake_ classes to a common test utils file for the module.

Instead of re-using FakeSchedulerBackend from another place, create a FakeSchedulerBackend here.

I too missed it completely.
I think it wasn't a great idea in the first place to use FakeSchedulerBackend of some other class from maintenance perspective. I am going ahead with your option 2.

rxin · 2016-04-15T05:43:23Z

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

+    * @param scheduler TaskScheduler that will be used with the scheduler backend.
+    * @return SchedulerBackend that works with a TaskScheduler
+    */
+  def createSchedulerBackend(sc: SparkContext,


the way we indent is

def createSchedulerBackend( sc: SparkContext, masterURL: String, scheduler: TaskScheduler): SchedulerBackend

rxin · 2016-04-15T05:44:27Z

Sorry for the delay. This looks pretty good. Just have some comments about the style to be more consistent with rest of the Spark codebase.

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

rxin · 2016-04-16T07:22:18Z

there are some conflicts with master - can you rebase? Thanks.

rxin · 2016-04-16T07:23:06Z

The latest changes LGTM

SparkQA · 2016-04-16T07:38:53Z

Test build #55997 has finished for PR 11723 at commit 2a81517.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-04-16T07:51:57Z

Test build #55998 has finished for PR 11723 at commit 6747420.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-04-16T09:26:24Z

Test build #55994 has finished for PR 11723 at commit 696cc71.

This patch passes all tests.
This patch does not merge cleanly.
This patch adds the following public classes (experimental):
- trait ExternalClusterManager

hbhanawat · 2016-04-16T09:50:03Z

test this please

hbhanawat · 2016-04-16T17:26:21Z

@rxin how do I get this retested by Jenkins? There were few issues going on with the Jenkins when I checked in my last changes and now it is not retesting it?

srowen · 2016-04-16T17:26:48Z

Jenkins retest this please

srowen · 2016-04-16T17:26:55Z

Jenkins add to whitelist

SparkQA · 2016-04-16T19:20:28Z

Test build #56012 has finished for PR 11723 at commit 6747420.

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

rxin · 2016-04-17T06:43:04Z

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

+ * A cluster manager interface to plugin external scheduler.
+ */
+@DeveloperApi
+trait ExternalClusterManager {


i just realized most of the return types used in this class are private[spark], so your implementation of this interface would need to be in the spark package anyway. I'm going to add private[spark] to this when I merge.

rxin · 2016-04-17T06:43:20Z

Merging in master. Thanks.

rxin · 2016-04-17T06:50:37Z

One thing - can you guys try to see if you can implement one of the existing cluster managers with this, and then we can make sure this is a proper API? Otherwise it is really easy to get removed because it is currently unused by anything in Spark.

hbhanawat · 2016-04-17T07:43:42Z

@rxin I will open another JIRA and a PR to do this. Thanks for the review.

tedyu · 2016-04-17T17:06:40Z

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

+   * executor exits differently. For e.g. when an executor goes down,
+   * back-end may not want to take the parent process down.
+   */
+  protected def exitExecutor(): Unit = System.exit(1)


Should a parameter of exit code (int) be added to this method ?

Created #12457

## What changes were proposed in this pull request? This commit adds support for pluggable cluster manager. And also allows a cluster manager to clean up tasks without taking the parent process down. To plug a new external cluster manager, ExternalClusterManager trait should be implemented. It returns task scheduler and backend scheduler that will be used by SparkContext to schedule tasks. An external cluster manager is registered using the java.util.ServiceLoader mechanism (This mechanism is also being used to register data sources like parquet, json, jdbc etc.). This allows auto-loading implementations of ExternalClusterManager interface. Currently, when a driver fails, executors exit using system.exit. This does not bode well for cluster managers that would like to reuse the parent process of an executor. Hence, 1. Moving system.exit to a function that can be overriden in subclasses of CoarseGrainedExecutorBackend. 2. Added functionality of killing all the running tasks in an executor. ## How was this patch tested? ExternalClusterManagerSuite.scala was added to test this patch. Author: Hemant Bhanawat <[email protected]> Closes apache#11723 from hbhanawat/pluggableScheduler.

…he#11723, any cluster manager can now be integrated with Spark. It was suggested in ExternalClusterManager PR that one of the existing cluster managers should start using the new interface to ensure that the API is correct. Ideally, all the existing cluster managers should eventually use the ECM interface but as a first step yarn will now use the ECM interface. This PR refactors YARN code from SparkContext.createTaskScheduler function into YarnClusterManager that implements ECM interface.

…se newly added ExternalClusterManager ## What changes were proposed in this pull request? With the addition of ExternalClusterManager(ECM) interface in PR #11723, any cluster manager can now be integrated with Spark. It was suggested in ExternalClusterManager PR that one of the existing cluster managers should start using the new interface to ensure that the API is correct. Ideally, all the existing cluster managers should eventually use the ECM interface but as a first step yarn will now use the ECM interface. This PR refactors YARN code from SparkContext.createTaskScheduler function into YarnClusterManager that implements ECM interface. ## How was this patch tested? Since this is refactoring, no new tests has been added. Existing tests have been run. Basic manual testing with YARN was done too. Author: Hemant Bhanawat <[email protected]> Closes #12641 from hbhanawat/yarnClusterMgr.

This commit adds support for pluggable cluster manager. And also allows a cluster manager to clean up tasks without taking the parent process down. To plug a new external cluster manager, ExternalClusterManager trait should be implemented. It returns task scheduler and backend scheduler that will be used by SparkContext to schedule tasks. An external cluster manager is registered using the java.util.ServiceLoader mechanism (This mechanism is also being used to register data sources like parquet, json, jdbc etc.). This allows auto-loading implementations of ExternalClusterManager interface. Currently, when a driver fails, executors exit using system.exit. This does not bode well for cluster managers that would like to reuse the parent process of an executor. Hence, 1. Moving system.exit to a function that can be overriden in subclasses of CoarseGrainedExecutorBackend. 2. Added functionality of killing all the running tasks in an executor. ExternalClusterManagerSuite.scala was added to test this patch. Author: Hemant Bhanawat <[email protected]> Closes apache#11723 from hbhanawat/pluggableScheduler. Conflicts: core/src/main/scala/org/apache/spark/SparkContext.scala core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala core/src/main/scala/org/apache/spark/executor/Executor.scala core/src/main/scala/org/apache/spark/scheduler/ExternalClusterManager.scala core/src/test/resources/META-INF/services/org.apache.spark.scheduler.ExternalClusterManager core/src/test/scala/org/apache/spark/scheduler/ExternalClusterManagerSuite.scala dev/.rat-excludes

…se newly added ExternalClusterManager With the addition of ExternalClusterManager(ECM) interface in PR apache#11723, any cluster manager can now be integrated with Spark. It was suggested in ExternalClusterManager PR that one of the existing cluster managers should start using the new interface to ensure that the API is correct. Ideally, all the existing cluster managers should eventually use the ECM interface but as a first step yarn will now use the ECM interface. This PR refactors YARN code from SparkContext.createTaskScheduler function into YarnClusterManager that implements ECM interface. Since this is refactoring, no new tests has been added. Existing tests have been run. Basic manual testing with YARN was done too. Author: Hemant Bhanawat <[email protected]> Closes apache#12641 from hbhanawat/yarnClusterMgr. Conflicts: core/src/main/scala/org/apache/spark/SparkContext.scala core/src/test/scala/org/apache/spark/SparkContextSchedulerCreationSuite.scala

Fixed issues in the test

e87c1e0

Since the scheduler APIs are not stable and may change in future, mar…

ae808d7

…king ExternalClusterManager as DeveloperApi and making it as private[spark].

tejasapatil reviewed Mar 24, 2016
View reviewed changes

Added review comments.

59990bf

1. Fixed formatting issues 2. Added master url as part of the other functions of ExternalClusterManager

tejasapatil reviewed Mar 24, 2016
View reviewed changes

rxin reviewed Apr 15, 2016
View reviewed changes

Hemant Bhanawat added 2 commits April 16, 2016 12:37

Fixed code style issues.

696cc71

Merge branch 'master' into pluggableScheduler

2a81517

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

Fixed scala style issues

6747420

rxin reviewed Apr 17, 2016
View reviewed changes

asfgit closed this in af1f4da Apr 17, 2016

tedyu reviewed Apr 17, 2016
View reviewed changes

hbhanawat mentioned this pull request Apr 23, 2016

[SPARK-14729][Scheduler] Refactored YARN scheduler creation code to use newly added ExternalClusterManager #12641

Closed

[SPARK-13904][Scheduler]Add support for pluggable cluster manager #11723

[SPARK-13904][Scheduler]Add support for pluggable cluster manager #11723

Uh oh!

Conversation

hbhanawat commented Mar 15, 2016

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

sarutak commented Mar 16, 2016

Uh oh!

SparkQA commented Mar 16, 2016

Uh oh!

rxin commented Mar 16, 2016

Uh oh!

hbhanawat commented Mar 16, 2016

Uh oh!

rxin commented Mar 16, 2016

Uh oh!

rxin commented Mar 16, 2016

Uh oh!

hbhanawat commented Mar 17, 2016

Uh oh!

SparkQA commented Mar 17, 2016

Uh oh!

SparkQA commented Mar 18, 2016

Uh oh!

hbhanawat commented Mar 18, 2016

Uh oh!

hbhanawat commented Mar 23, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tejasapatil commented Mar 24, 2016

Uh oh!

rxin commented Mar 24, 2016

Uh oh!

tejasapatil commented Mar 24, 2016

Uh oh!

rxin commented Mar 24, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rxin commented Apr 15, 2016

Uh oh!

rxin commented Apr 16, 2016

Uh oh!

rxin commented Apr 16, 2016

Uh oh!

SparkQA commented Apr 16, 2016

Uh oh!

SparkQA commented Apr 16, 2016

Uh oh!

SparkQA commented Apr 16, 2016

Uh oh!

hbhanawat commented Apr 16, 2016

Uh oh!

hbhanawat commented Apr 16, 2016

Uh oh!

srowen commented Apr 16, 2016

Uh oh!

srowen commented Apr 16, 2016

Uh oh!

SparkQA commented Apr 16, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rxin commented Apr 17, 2016

Uh oh!

rxin commented Apr 17, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hbhanawat commented Apr 17, 2016

Uh oh!

Choose a reason for hiding this comment

rxin commented Apr 17, 2016 •

edited

Loading