[SPARK-19326] Speculated task attempts do not get launched in few scenarios #18492

janewangfb · 2017-06-30T17:29:53Z

What changes were proposed in this pull request?

Add a new listener event when a speculative task is created and notify it to ExecutorAllocationManager for requesting more executor.

How was this patch tested?

Added Unittests.
For the test snippet in the jira:
val n = 100
val someRDD = sc.parallelize(1 to n, n)
someRDD.mapPartitionsWithIndex( (index: Int, it: Iterator[Int]) => {
if (index == 1) {
Thread.sleep(Long.MaxValue) // fake long running task(s)
}
it.toList.map(x => index + ", " + x).iterator
}).collect
With this code change, spark indicates 101 jobs are running (99 succeeded, 2 running and 1 is speculative job)

jiangxb1987 · 2017-07-07T15:43:27Z

@tejasapatil Do you want to look at this PR? Thanks!

jiangxb1987 · 2017-07-19T11:00:56Z

core/src/main/java/org/apache/spark/SparkFirehoseListener.java

+  }
+
+  @Override
+  public void onExtraExecutorNeeded() {


Maybe I'm missing something obvious, but what's use of this except for adding some log?

When there is one executor left but with multiple cpu cores, the task is running on that executor. But due to locality requirements, speculative jobs cannot launch on the same host. We will have to request for one extra executor. Thats is what this event is for.

jiangxb1987 · 2017-07-19T11:01:23Z

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

@@ -572,20 +572,35 @@ private[spark] class ExecutorAllocationManager(
  }

  /**
+   * Callback invokded when an extra executor is needed (See SPARK-19326)


nit: invokded -> invoked

jiangxb1987 · 2017-07-19T11:01:56Z

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

@@ -281,6 +281,20 @@ class DAGScheduler(
    eventProcessLoop.post(TaskSetFailed(taskSet, reason, exception))
  }

+  /**
+   * Called by the TaskSetManager when it needs a speculative task is needed.


nit: needs -> decides

jiangxb1987 · 2017-07-20T08:22:18Z

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

+   * Callback invoked when an extra executor is needed (See SPARK-19326)
+   */
+  private def onExtraExecutorNeeded(): Unit = synchronized {
+    val maxNeeded = math.max(math.min(maxNumExecutorsNeeded + 1, maxNumExecutors), minNumExecutors)


Perhaps what we need is just to ensure there are more than two active executors left, so we may meet the locality requirement and launch the speculative jobs?

BTW how do we ensure we run new executors on a different host?

we cannot just ensure there are more than two active executors left. it will depends on if there is any speculative jobs not launched.

hostToLocalTaskCount will make the new executors request satisfies the locality requirement. Also, on the other hand, if even the new executor was not guaranteed to be different host, it will idle and then die. and the speculative task will request another executor.

cloud-fan · 2017-07-28T12:48:42Z

ok to test

SparkQA · 2017-07-28T15:59:13Z

Test build #80022 has finished for PR 18492 at commit 0fa5c01.

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

cloud-fan · 2017-08-07T08:50:24Z

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

    private var numRunningTasks: Int = _

+    private val stageIdToNumSpeculativeTasks = new mutable.HashMap[Int, Int]


at the first glance I thought stageIdToNumSpeculativeTasks is just stageIdToSpeculativeTaskIndices.mapValues(_.size), but seems it's not true, can you add some comments to explain these 2 variables?

Added comments

cloud-fan · 2017-08-08T04:38:59Z

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

+case class SparkListenerSpeculativeTaskSubmitted(stageId: Int) extends SparkListenerEvent
+
+@DeveloperApi
+case class SparkListenerExtraExecutorNeeded() extends SparkListenerEvent


case object?

case object wont compile.

cloud-fan · 2017-08-08T04:45:01Z

core/src/main/java/org/apache/spark/SparkFirehoseListener.java

+
+  @Override
+  public void onExtraExecutorNeeded() {
+    onEvent(null);


SparkFirehoseListener is a public API and users may assume onEvent will never accept null, how about we just do nothing here?

ok. updated.

SparkQA · 2017-08-08T07:04:50Z

Test build #80378 has finished for PR 18492 at commit 37cf760.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-08-08T07:04:51Z

Test build #80380 has finished for PR 18492 at commit 77b4729.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

janewangfb · 2017-08-08T07:15:28Z

Jenkins test this again please.

jiangxb1987 · 2017-08-08T11:01:06Z

retest this please

SparkQA · 2017-08-08T14:08:23Z

Test build #80387 has finished for PR 18492 at commit 77b4729.

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

cloud-fan · 2017-08-11T07:24:14Z

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

+  /**
+   * Called when an extra executor is needed
+   */
+  def onExtraExecutorNeeded(): Unit


I'm hesitant to add this. SparkListener is public and should be the interface to listen to various Spark internal events, and do whatever they want. However, onExtraExecutorNeeded sounds like something Spark asks the listener to do, which is against the pattern.

In another word, onExtraExecutorNeeded looks specific to the executor allocation manager, but not a general spark event.

do you know how executor allocation manager adjust the number of executors currently? can we follow it instead of hacking the SparkListener?

@cloud-fan after thoughts, yes, I think we can get rid of extraExecutorNeeded event and handle it in ExecutorAllocationManager.scala.

jiangxb1987 · 2017-08-15T13:16:42Z

ping @janewangfb

SparkQA · 2017-08-16T07:04:49Z

Test build #80714 has finished for PR 18492 at commit 8b8b128.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2017-08-16T07:04:50Z

Test build #80712 has finished for PR 18492 at commit f7cdad9.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

jiangxb1987 · 2017-08-16T15:57:56Z

retest this please

SparkQA · 2017-08-16T19:08:02Z

Test build #80742 has finished for PR 18492 at commit 8b8b128.

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

cloud-fan · 2017-08-17T14:19:05Z

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

@@ -291,6 +294,11 @@ private[spark] trait SparkListenerInterface {
  def onBlockUpdated(blockUpdated: SparkListenerBlockUpdated): Unit

  /**
+   * Called when a speculative task is submitted
+   */
+  def onSpeculativeTaskSubmitted(speculativeTask: SparkListenerSpeculativeTaskSubmitted): Unit


For normal task, we have onTaskStart, onTaskEnd, etc. but don't have onTaskSubmitted. Shall we make the name consistent for speculative task?

I would keep the name onSpeculativeTaskSubmitted. Because when the event happens, it only submit a speculative task to be launched in the future, the task has not started yet.

oh i see. So we don't track the submission of normal tasks?

I grepped, and dont think we have events related to normal task submitted.

cloud-fan · 2017-08-17T14:20:02Z

core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala

@@ -980,10 +1014,12 @@ private object ExecutorAllocationManagerSuite extends PrivateMethodTester {
      taskLocalityPreferences = taskLocalityPreferences)
  }

-  private def createTaskInfo(taskId: Int, taskIndex: Int, executorId: String): TaskInfo = {
-    new TaskInfo(taskId, taskIndex, 0, 0, executorId, "", TaskLocality.ANY, speculative = false)
+  private def createTaskInfo(taskId: Int, taskIndex: Int, executorId: String,


nit:

private def xxxx( para1: XX, para2: XX)

4 spaces indention for the parameter.

cloud-fan · 2017-08-17T14:20:39Z

LGTM except 2 minor comments, thanks for working on it!

SparkQA · 2017-08-17T20:17:47Z

Test build #80791 has finished for PR 18492 at commit 7a8ca2a.

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

jiangxb1987 · 2017-08-19T13:58:07Z

LGTM

cloud-fan · 2017-08-21T14:24:03Z

core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala

+    assert(addExecutors(manager) === 1)
+    sc.listenerBus.postToAll(SparkListenerSpeculativeTaskSubmitted(1))
+    sc.listenerBus.postToAll(SparkListenerSpeculativeTaskSubmitted(1))
+    sc.listenerBus.postToAll(SparkListenerStageSubmitted(createStageInfo(1, 2)))


why the stage submitted event is posted after speculative task submitted event?

In real life, it is possible that a job has multiple stages, one stage is still running some tasks but the next stage starts already. This test tries to micmic.

cloud-fan · 2017-08-21T14:27:31Z

core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala

+    val manager = sc.executorAllocationManager.get
+
+    // Verify that we're capped at number of tasks including the speculative ones in the stage
+    sc.listenerBus.postToAll(SparkListenerSpeculativeTaskSubmitted(1))


is it a possible case? the first event is speculative task submitted?

That is not likely. Speculative job is only submitted when certain percentage of jobs have finished successfully.

cloud-fan · 2017-08-22T05:33:48Z

core/src/test/scala/org/apache/spark/ExecutorAllocationManagerSuite.scala

+    assert(numExecutorsTarget(manager) === 5)
+    assert(numExecutorsToAdd(manager) === 1)
+
+    // Verify that running a task doesn't affect the target


can you explain more about this test? Why the first 3 SparkListenerSpeculativeTaskSubmitted events can trigger to allocate more executors, but here we don't?

It is because we use the sum of (running + appending) jobs to calculate how many executors are needed (maxNumExecutorsNeeded). so wether a task is pending or running, the executors needed are the same.

then why speculative task submission adds running/appending jobs?

speculative task is also a task which needs executor to execute it, so, when we calculate how many executors are needed, we need to count the speculative tasks.

cloud-fan · 2017-08-23T03:32:20Z

thanks, merging to master!

jerryshao · 2017-10-23T08:07:24Z

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

+      // Check if there is any speculative jobs pending
+      if (listener.pendingTasks == 0 && listener.pendingSpeculativeTasks > 0) {
+        numExecutorsTarget =
+          math.max(math.min(maxNumExecutorsNeeded + 1, maxNumExecutors), minNumExecutors)


@janewangfb Would you please explain why here + 1 if there's pending speculativeTasks, should the number of executors be calculated based on the number of pending tasks? Thanks!

Same here.
maxNumExecutorsNeeded + 1 doesn't quite make sense.
@janewangfb could you please post/update some comments here? And I wonder why we didn't take pendingSpeculativeTasks into account when calculating maxNumExecutorsNeeded()

Or @jerryshao, do you know the rationale?

Also confused by +1 here. And I think we have already take pendingSpeculativeTasks into account @advancedxy :

def totalPendingTasks(): Int = { pendingTasks + pendingSpeculativeTasks }

Seems this check is redundant.
And it doesn't sync to CM if numExecutorsTarget change(after +1).

@janewangfb @cloud-fan Sorry, I realize this is very old PR but found it because I was confused by this logic as well, is there a reason we are adding 1 here with speculative?

I can't remember the details as it's too old. But when I look at it again now, this looks a mistake to me: the +1 seems to try to match the numExecutorsToAdd = 1 in the previous code. However, numExecutorsToAdd = 1 doesn't mean we want to allocate one more executor right now.

thanks, filed https://issues.apache.org/jira/browse/SPARK-28403

janewangfb added 8 commits June 29, 2017 11:50

add speculative job for executor calculation

9d4886e

Merge branch 'master' into speculated_task_not_launched

d4dd56c

Add Unittests

210ba70

remove debug lines

cb28404

remove debug

41170df

remove debug lines

9784b4e

Merge branch 'master' into speculated_task_not_launched

5331ab9

Merge branch 'master' into speculated_task_not_launched

32d0c23

janewangfb added 2 commits July 12, 2017 17:40

Merge branch 'master' into speculated_task_not_launched

92c4d09

Added extraExecutorNeeded event

33f5757

jiangxb1987 reviewed Jul 19, 2017

View reviewed changes

janewangfb added 2 commits July 19, 2017 12:04

Merge branch 'master' into speculated_task_not_launched

a1fb8d3

address jiangxb1987's comments

0fa5c01

jiangxb1987 reviewed Jul 20, 2017

View reviewed changes

cloud-fan reviewed Aug 7, 2017

View reviewed changes

janewangfb added 2 commits August 7, 2017 21:18

Merge branch 'master' into speculated_task_not_launched

94c4db6

address cloud-fan's comment

37cf760

cloud-fan reviewed Aug 8, 2017

View reviewed changes

address cloud-fan's comment

77b4729

cloud-fan reviewed Aug 11, 2017

View reviewed changes

janewangfb added 4 commits August 15, 2017 10:24

Merge branch 'master' into speculated_task_not_launched

c7c16a6

remove extraExecutorNeeded event

fa31cd8

Merge branch 'master' into speculated_task_not_launched

f7cdad9

remove SparkListenerExtraExecutorNeeded

8b8b128

cloud-fan reviewed Aug 17, 2017

View reviewed changes

janewangfb added 2 commits August 17, 2017 10:06

Merge branch 'master' into speculated_task_not_launched

35cf6a5

Address cloud-fan's comment

7a8ca2a

cloud-fan reviewed Aug 21, 2017

View reviewed changes

cloud-fan reviewed Aug 22, 2017

View reviewed changes

asfgit closed this in d58a350 Aug 23, 2017

jerryshao reviewed Oct 23, 2017

View reviewed changes

		private var numRunningTasks: Int = _

		private val stageIdToNumSpeculativeTasks = new mutable.HashMap[Int, Int]

[SPARK-19326] Speculated task attempts do not get launched in few scenarios #18492

[SPARK-19326] Speculated task attempts do not get launched in few scenarios #18492

Conversation

janewangfb commented Jun 30, 2017

What changes were proposed in this pull request?

How was this patch tested?

jiangxb1987 commented Jul 7, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Jul 28, 2017

SparkQA commented Jul 28, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Aug 8, 2017

SparkQA commented Aug 8, 2017

janewangfb commented Aug 8, 2017

jiangxb1987 commented Aug 8, 2017

SparkQA commented Aug 8, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jiangxb1987 commented Aug 15, 2017

SparkQA commented Aug 16, 2017

SparkQA commented Aug 16, 2017

jiangxb1987 commented Aug 16, 2017

SparkQA commented Aug 16, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Aug 17, 2017

SparkQA commented Aug 17, 2017

jiangxb1987 commented Aug 19, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cloud-fan commented Aug 23, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Ngone51 Sep 11, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Ngone51 Sep 11, 2018 •

edited

Loading