[SPARK-5095] [Mesos] Support launching multiple mesos executors in coarse grained mesos mode. #10993

mgummelt · 2016-01-30T00:00:02Z

This is the next iteration of @tnachen's previous PR: #4027

In that PR, we resolved with @andrewor14 and @pwendell to implement the Mesos scheduler's support of spark.executor.cores to be consistent with YARN and Standalone. This PR implements that resolution.

This PR implements two high-level features. These two features are co-dependent, so they're implemented both here:

Mesos support for spark.executor.cores
Multiple executors per slave

We at Mesosphere have been working with Typesafe on a Spark/Mesos integration test suite: https://github.com/typesafehub/mesos-spark-integration-tests, which passes for this PR.

The contribution is my original work and I license the work to the project under the project's open source license.

mgummelt · 2016-01-30T00:01:22Z

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala

@@ -89,13 +82,11 @@ private[spark] class CoarseMesosSchedulerBackend(
   */
  private[mesos] def executorLimit: Int = executorLimitOption.getOrElse(Int.MaxValue)

-  private val pendingRemovedSlaveIds = new HashSet[String]


AFAICT, this is never used, so I removed it.

SparkQA · 2016-01-30T01:40:30Z

Test build #50419 has finished for PR 10993 at commit f421133.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

dragos · 2016-02-01T11:00:21Z

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala

-  val taskIdToSlaveId: HashBiMap[Int, String] = HashBiMap.create[Int, String]
-  // How many times tasks on each slave failed
-  val failuresBySlaveId: HashMap[String, Int] = new HashMap[String, Int]
+  private val slaves = new HashMap[String, Slave]


Not a big issue, but I wonder if slaveInfo would be a better name. Also, a one-line comment to explain what the key is.

andrewor14 · 2016-02-01T19:44:36Z

Looks like this is failing real tests

andrewor14 · 2016-02-01T20:32:34Z

Also, have you had a chance to test this with cluster mode and/or with dynamic allocation?

mgummelt · 2016-02-01T22:19:40Z

I made an update that should fix the test.

I've tested in cluster mode, but not with dynamic allocation. Though I have added some unit tests that cover dynamic allocation. I'll see about setting up dynamic allocation.

SparkQA · 2016-02-02T00:32:01Z

Test build #50506 has finished for PR 10993 at commit 318486e.

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

dragos · 2016-02-02T10:45:57Z

I didn't have time to look at this in detail, I'll do so this afternoon.

dragos · 2016-02-02T16:29:48Z

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala

+      val id = offer.getId.getValue
+
+      if (tasks.contains(offer.getId)) { // accept
+        val filters = Filters.newBuilder().setRefuseSeconds(5).build()


I know this is not your code, but it would be good to document this. Why do we filter out offers for 5 seconds on the offers we use?

I was thinking about this when I ran into it. The default is actually 5: https://github.com/apache/mesos/blob/master/include/mesos/mesos.proto#L1211

So I'll just remove it.

dragos · 2016-02-02T17:18:52Z

@mgummelt this looks really good! I have a few comments. I still have to run this PR with dynamic allocation and see it in action!

SparkQA · 2016-02-03T01:05:01Z

Test build #50608 has finished for PR 10993 at commit 0a1181a.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

Astralidea · 2016-02-03T03:19:55Z

@mgummelt If I want to scheduler deploy with : Limiting a slave can only start one executor, how to do that? should I can have another config like spark.executor.maxperslave=1?
The scene is like is:
I read Kafka use spark-streaming & I need Receiver to deploy in each slave separately, not all of the them in one slave to occupied bandwidth..

dragos · 2016-02-03T07:22:57Z

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala

+    * @param offers Mesos offers that match attribute constraints
+    * @return A map from OfferID to a list of Mesos tasks to launch on that offer
+    */
+  private def getMesosTasks(offers: Buffer[Offer]): Map[OfferID, List[MesosTaskInfo]] = {


I find it non-productive to quibble over a name. That being said, this method doesn't just get tasks from somewhere. It produces them itself, based on a round-robin scheduling strategy over the given offers. I don't think get is the best verb to describe that action.

Sure. Changed to build

mgummelt · 2016-02-05T19:18:31Z

@andrewor14 Glad to be here! Flaky tests or no

I think all concerns have been addressed except for dynamic allocation testing, which seems to be broken entirely: SPARK-12583

@dragos Any other comments?

SparkQA · 2016-02-05T21:24:12Z

Test build #50833 has finished for PR 10993 at commit 7e3f39d.

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

dragos · 2016-02-06T12:26:54Z

LGTM! Great work, @mgummelt!

tnachen · 2016-02-08T16:10:05Z

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala

-    val slaveId: String = status.getSlaveId.getValue
+    val taskId = status.getTaskId.getValue
+    val slaveId = status.getSlaveId.getValue
+    val slave = slaves(slaveId)


This needs to be moved in the stateLock right?

Moved, but I don't understand why the mesos methods are synchronized in the first place. They should only be called by a single thread (the driver thread).

The ExecutionAllocationManager, used when dynamic allocation is enabled, runs on a different thread. Not sure if this particular method can run on different threads, but there's at least the issue of visibility.

tnachen · 2016-02-08T16:14:22Z

Just one comment, overall LGTM

Support spark.executor.cores on Mesos.

andrewor14 · 2016-02-09T18:44:36Z

retest this please

SparkQA · 2016-02-09T20:48:49Z

Test build #50985 has finished for PR 10993 at commit ecad77a.

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

SparkQA · 2016-02-09T20:56:58Z

Test build #50984 has finished for PR 10993 at commit ecad77a.

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

andrewor14 · 2016-02-10T18:42:29Z

core/src/main/scala/org/apache/spark/scheduler/cluster/mesos/CoarseMesosSchedulerBackend.scala

+  // SlaveID -> Slave
+  // This map accumulates entries for the duration of the job.  Slaves are never deleted, because
+  // we need to maintain e.g. failure state and connection state.
+  private val slaves = new HashMap[String, Slave]


elsewhere in Spark we would call this class SlaveInfo instead of just Slave, so we don't confuse it with the Mesos Slave

andrewor14 · 2016-02-10T18:52:53Z

LGTM merging into master. @mgummelt feel free to address the remainder of the comments in a follow-up patch.

mgummelt · 2016-02-10T21:30:55Z

Thanks for merging. Will this go into 1.6.1, or not until 2.0?

andrewor14 · 2016-02-10T21:36:20Z

This is a big new feature. It will not go into a maintenance release (1.6.1).

andrewor14 · 2016-02-11T00:10:24Z

@mgummelt looks like this caused a flaky test:
https://spark-tests.appspot.com/tests/org.apache.spark.scheduler.cluster.mesos.CoarseMesosSchedulerBackendSuite/mesos%20kills%20an%20executor%20when%20told

Do you have the bandwidth to fix it quickly? If not I'll just revert this patch for now and we can resubmit it later.

mgummelt · 2016-02-11T01:59:45Z

looking into it

Astralidea · 2016-02-11T02:14:34Z

@mgummelt Great Work! I think this feature will allow more people to use mesos.

mgummelt · 2016-02-11T02:18:14Z

@andrewor14 I haven't found the problem, but here's a PR to remove the test in the interim #11164

It's a strange test to be flaky. It's very simple.

mgummelt · 2016-02-11T02:23:10Z

Ah, I see the issue. There's a thread causing a race.

I won't be able to fix until tomorrow, though.

andrewor14 This addressed your style comments from #10993 Author: Michael Gummelt <mgummelt@mesosphere.io> Closes #11187 from mgummelt/fix_mesos_style.

…rse grained mesos mode. This is the next iteration of tnachen's previous PR: apache#4027 In that PR, we resolved with andrewor14 and pwendell to implement the Mesos scheduler's support of `spark.executor.cores` to be consistent with YARN and Standalone. This PR implements that resolution. This PR implements two high-level features. These two features are co-dependent, so they're implemented both here: - Mesos support for spark.executor.cores - Multiple executors per slave We at Mesosphere have been working with Typesafe on a Spark/Mesos integration test suite: https://github.com/typesafehub/mesos-spark-integration-tests, which passes for this PR. The contribution is my original work and I license the work to the project under the project's open source license. Author: Michael Gummelt <mgummelt@mesosphere.io> Closes apache#10993 from mgummelt/executor_sizing.

mgummelt reviewed Jan 30, 2016
View reviewed changes

mgummelt mentioned this pull request Jan 30, 2016

Multiple executors per node, and spark.executor.cores support mesosphere/spark#3

Closed

dragos reviewed Feb 1, 2016
View reviewed changes

mgummelt force-pushed the executor_sizing branch from f421133 to 318486e Compare February 1, 2016 22:13

dragos reviewed Feb 2, 2016
View reviewed changes

mgummelt force-pushed the executor_sizing branch from 318486e to 0a1181a Compare February 2, 2016 23:18

dragos reviewed Feb 3, 2016
View reviewed changes

tnachen reviewed Feb 8, 2016
View reviewed changes

Support multiple executors per node on Mesos.

ecad77a

Support spark.executor.cores on Mesos.

mgummelt force-pushed the executor_sizing branch from 7e3f39d to ecad77a Compare February 9, 2016 18:36

andrewor14 reviewed Feb 10, 2016
View reviewed changes

asfgit closed this in 80cb963 Feb 10, 2016

andrewor14 mentioned this pull request Feb 11, 2016

[SPARK-5095] remove flaky test #11164

Closed

mgummelt deleted the executor_sizing branch February 12, 2016 19:31

mgummelt mentioned this pull request Feb 12, 2016

[SPARK-5095] Fix style in mesos coarse grained scheduler code #11187

Closed

asfgit pushed a commit that referenced this pull request Feb 12, 2016

[SPARK-5095] Fix style in mesos coarse grained scheduler code

38bc601

andrewor14 This addressed your style comments from #10993 Author: Michael Gummelt <mgummelt@mesosphere.io> Closes #11187 from mgummelt/fix_mesos_style.

[SPARK-5095] [Mesos] Support launching multiple mesos executors in coarse grained mesos mode. #10993

[SPARK-5095] [Mesos] Support launching multiple mesos executors in coarse grained mesos mode. #10993

Conversation

mgummelt commented Jan 30, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Jan 30, 2016

Choose a reason for hiding this comment

andrewor14 commented Feb 1, 2016

andrewor14 commented Feb 1, 2016

mgummelt commented Feb 1, 2016

SparkQA commented Feb 2, 2016

dragos commented Feb 2, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dragos commented Feb 2, 2016

SparkQA commented Feb 3, 2016

Astralidea commented Feb 3, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mgummelt commented Feb 5, 2016

SparkQA commented Feb 5, 2016

dragos commented Feb 6, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tnachen commented Feb 8, 2016

andrewor14 commented Feb 9, 2016

SparkQA commented Feb 9, 2016

SparkQA commented Feb 9, 2016

Choose a reason for hiding this comment

andrewor14 commented Feb 10, 2016

mgummelt commented Feb 10, 2016

andrewor14 commented Feb 10, 2016

andrewor14 commented Feb 11, 2016

mgummelt commented Feb 11, 2016

Astralidea commented Feb 11, 2016

mgummelt commented Feb 11, 2016

mgummelt commented Feb 11, 2016