[SPARK-17365][Core] Remove/Kill multiple executors together to reduce RPC call time. by dhruve · Pull Request #15152 · apache/spark

dhruve · 2016-09-19T19:03:48Z

What changes were proposed in this pull request?

We are killing multiple executors together instead of iterating over expensive RPC calls to kill single executor.

How was this patch tested?

Executed sample spark job to observe executors being killed/removed with dynamic allocation enabled.

… contention [SPARK-17365][Core] Return executorIds which are actually killed while killing multiple executors Fix import wrap and remove unused import Add mima exclude and refactor few methods.

vanzin · 2016-09-19T19:08:29Z

ok to test

vanzin · 2016-09-19T19:08:34Z

add to whitelist

SparkQA · 2016-09-19T21:03:09Z

Test build #65609 has finished for PR 15152 at commit 5a185d2.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class SparkContext(config: SparkConf) extends Logging

SparkQA · 2016-09-20T00:34:09Z

Test build #65618 has finished for PR 15152 at commit d779a88.

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

SparkQA · 2016-09-20T15:27:05Z

Test build #65655 has finished for PR 15152 at commit bef1d1e.

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

dhruve · 2016-09-20T18:44:30Z

PySpark unit tests failure unrelated.

dhruve · 2016-09-20T18:44:52Z

retest this please

SparkQA · 2016-09-20T21:18:47Z

Test build #65675 has finished for PR 15152 at commit bef1d1e.

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

dhruve · 2016-09-21T14:59:42Z

@vanzin, @tgravescs Can you review these changes.

I will resolve the MiMa exclude merge conflicts once the PR is ready to be accepted - as these are being added often. Thanks.

vanzin

Looks ok, a few minor things to fix though.

vanzin · 2016-09-21T17:14:48Z

- def killExecutor(executorId: String): Boolean = killExecutors(Seq(executorId))
+ def killExecutor(executorId: String): Boolean = {
+ val killedExecutors = killExecutors(Seq(executorId))
+ killedExecutors.nonEmpty && killedExecutors(0).equals(executorId)


I seriously hope the backend is not killing some random executor instead of the one that was asked...

It doesn't. Just added an additional check here to be sure. I need to change the short circuit operator here to &.

vanzin · 2016-09-21T17:20:39Z

 .filter { id => force || !scheduler.isExecutorBusy(id) }
 executorsToKill.foreach { id => executorsPendingToRemove(id) = !replace }

+ logInfo(s"Requesting to kill filtered executor(s) ${executorsToKill.mkString(", ")}")


This sounds weird. The following looks slightly better: "Actual list of executors to be killed is ...".

This one sounds better.

vanzin · 2016-09-21T17:20:53Z

+ val killResponse = adjustTotalExecutors.flatMap(killExecutors)(ThreadUtils.sameThread)
+
+ killResponse.flatMap(killSuccessful =>
+ Future.successful ( if (killSuccessful) executorsToKill else Seq.empty[String])


nit: no space before '('

vanzin · 2016-09-21T17:21:53Z

+ assert(executorsPendingToRemove(manager).contains("2"))
+ assert(executorsPendingToRemove(manager).contains("3"))
+ assert(!removeExecutor(manager, "100")) // remove non-existent executors
+ assert(!(removeExecutors(manager, Seq("101", "102")) == Seq("101", "102")))


!= doesn't work?

vanzin · 2016-09-21T17:22:01Z

+
+ // Keep removing until the limit is reached
+ assert(executorsPendingToRemove(manager).isEmpty)
+ assert(removeExecutors(manager, Seq("1")) == Seq("1"))


Use === in all these.

vanzin · 2016-09-21T17:24:03Z

+
+ def canCreate(masterURL: String): Boolean = masterURL == "myDummyLocalExternalClusterManager"
+
+ override def createTaskScheduler(sc: SparkContext,


style: https://cwiki.apache.org/confluence/display/SPARK/Spark+Code+Style+Guide#SparkCodeStyleGuide-Indentation
Each parameter on a separate line when breaking. Happens in a bunch of methods here.

vanzin · 2016-09-21T17:24:52Z

+ ): Boolean = sc.requestTotalExecutors(numExecutors, localityAwareTasks, hostToLocalTaskCount)
+
+ override def requestExecutors(numAdditionalExecutors: Int): Boolean =
+ sc.requestExecutors(numAdditionalExecutors)


indentation

dhruve · 2016-09-21T18:39:27Z

@vanzin - fixed minor nits and indentation.

SparkQA · 2016-09-21T21:14:25Z

Test build #65729 has finished for PR 15152 at commit b76147c.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- class ChiSqSelector @Since(\"2.1.0\") () extends Serializable
- abstract class CompactibleFileStreamLog[T: ClassTag](
- class FileStreamSinkLog(
- case class FileEntry(path: String, timestamp: Timestamp, batchId: Long = NOT_SET)
- class FileStreamSourceLog(

vanzin · 2016-09-21T21:57:17Z

- def killExecutor(executorId: String): Boolean = killExecutors(Seq(executorId))
+ def killExecutor(executorId: String): Boolean = {
+ val killedExecutors = killExecutors(Seq(executorId))
+ killedExecutors.nonEmpty & killedExecutors(0).equals(executorId)


This needs to be short-circuited, otherwise it will throw an exception if the list is empty.

Ah! The earlier one was fine. Missed it.

SparkQA · 2016-09-22T01:22:31Z

Test build #65740 has finished for PR 15152 at commit 3d2fac4.

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

vanzin · 2016-09-22T17:10:01Z

LGTM. Merging to master.

[SPARK-17365][Core] Remove/Kill multiple executors together to reduce…
5a185d2
… contention [SPARK-17365][Core] Return executorIds which are actually killed while killing multiple executors Fix import wrap and remove unused import Add mima exclude and refactor few methods.

Fix streaming test by using mocked cluster manager
d779a88

Resolve merge conflict for MiMa excludes
bef1d1e

vanzin reviewed Sep 21, 2016
View reviewed changes

Dhruve Ashar added 2 commits September 21, 2016 12:56

Fix indentation & log message for killing executors
dfc188c

Resolve merge conflict for MimaExcludes
b76147c

vanzin reviewed Sep 21, 2016
View reviewed changes

Use short circuit operator
3d2fac4

asfgit closed this in 17b72d3Sep 22, 2016


		def canCreate(masterURL: String): Boolean = masterURL == "myDummyLocalExternalClusterManager"

		override def createTaskScheduler(sc: SparkContext,

Uh oh!

Conversation

dhruve commented Sep 19, 2016

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

vanzin commented Sep 19, 2016

Uh oh!

vanzin commented Sep 19, 2016

Uh oh!

SparkQA commented Sep 19, 2016

Uh oh!

SparkQA commented Sep 20, 2016

Uh oh!

SparkQA commented Sep 20, 2016

Uh oh!

dhruve commented Sep 20, 2016

Uh oh!

dhruve commented Sep 20, 2016

Uh oh!

SparkQA commented Sep 20, 2016

Uh oh!

dhruve commented Sep 21, 2016

Uh oh!

vanzin left a comment

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!

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

vanzinSep 21, 2016• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dhruve commented Sep 21, 2016

Uh oh!

SparkQA commented Sep 21, 2016

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Sep 22, 2016

Uh oh!

vanzin commented Sep 22, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

vanzinSep 21, 2016•
edited
Loading