[SPARK-11063][Streaming]Change preferredLocations of Receiver's RDD to hosts rather than hostports #9075

zsxwing · 2015-10-12T08:50:30Z

The format of RDD's preferredLocations must be hostname but the format of Streaming Receiver's scheduling executors is hostport. So it doesn't work.

This PR converts schedulerExecutors to hosts before creating Receiver's RDD.

…ports

SparkQA · 2015-10-12T09:50:28Z

Test build #43564 has finished for PR 9075 at commit 4706ec0.

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

jerryshao · 2015-10-13T12:37:20Z

The fix looks good to me.

tdas · 2015-10-16T08:46:35Z

What is the implication of this bug in case for the scheduling? We are trying to evenly distribute receivers to executors. But if we set the preferredLocations to only the granularity of hosts, then the following could happen. Two executors can be in the same host (through YARN). We set the preferred locations for two receivers for that same host even though we want them to be in two different executors. It may so happen that both executors get scheduled at the same executor. Isnt it?

zsxwing · 2015-10-16T08:54:32Z

What is the implication of this bug in case for the scheduling? We are trying to evenly distribute receivers to executors. But if we set the preferredLocations to only the granularity of hosts, then the following could happen. Two executors can be in the same host (through YARN). We set the preferred locations for two receivers for that same host even though we want them to be in two different executors. It may so happen that both executors get scheduled at the same executor. Isnt it?

Right. However, preferredLocations only supports hosts. So at least for branch 1.5, this is the best we can do. We can enhance preferredLocations to support host and port in master branch.

jerryshao · 2015-10-16T08:55:15Z

From my understanding, Spark's scheduler does not support scheduling with port (https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/scheduler/TaskLocation.scala)

tdas · 2015-10-16T09:00:59Z

Yeah, we have to fix it for branch 1.5 with a small fix like this.
Is it possible to test this with unit tests? Could you check if there are unit tests of the DAGScheduler / TaskScheduler that we can use to unit test so that something like this does not happen ever again.

jerryshao · 2015-10-16T09:13:39Z

I think we could add another TaskLocation, like ExecutorTaskLocation and pass the preferredLocations with special tag prefix like HDFS in memory tag, to let TaskSetManager know we want to schedule the task with process local.

zsxwing · 2015-10-16T16:08:21Z

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

@@ -388,7 +388,7 @@ private[spark] class TaskSetManager(
    if (TaskLocality.isAllowed(maxLocality, TaskLocality.NO_PREF)) {
      // Look for noPref tasks after NODE_LOCAL for minimize cross-rack traffic
      for (index <- dequeueTaskFromList(execId, pendingTasksWithNoPrefs)) {
-        return Some((index, TaskLocality.PROCESS_LOCAL, false))
+        return Some((index, TaskLocality.NO_PREF, false))


@CodingCat could you take a look at this change?

sure, will do tomorrow morning

This looks like a significant bug in the core. If so probably go into a separate patch.

And how is this related to the main bug we are considering in this PR?

This is not related. I will move it in a separate PR.

hmmm....it's a typo but would not take any effect

see discussions here: #3816 (only look at the comments after Dec 30, 2014 is enough)

Didn't notice this PR. Thanks.

zsxwing · 2015-10-16T16:09:02Z

Added a unit test and fixed a minor issue in TaskSetManager.

zsxwing · 2015-10-16T17:28:12Z

Tested this patch in a 5 nodes cluster. Each node has an executor, the executor core is 1 and the receiver number is 5.

Before this patch, there were several Receiver restarting logs. After applying this patch, all restarting logs disappeared.

SparkQA · 2015-10-16T18:30:27Z

Test build #43840 has finished for PR 9075 at commit 2d7c030.

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

SparkQA · 2015-10-17T04:43:08Z

Test build #43877 has finished for PR 9075 at commit 49b7792.

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

tdas · 2015-10-17T09:09:30Z

streaming/src/test/scala/org/apache/spark/streaming/scheduler/ReceiverTrackerSuite.scala

+      ssc.start()
+      eventually(timeout(10 seconds), interval(10 millis)) {
+        // If preferredLocations is set correctly, receiverTaskLocality should be NODE_LOCAL
+        assert(receiverTaskLocality === TaskLocality.NODE_LOCAL)


What will it be if it is not set correctly? That is, without the fix above, what would locality be

It's TaskLocality.ANY without this patch.

tdas · 2015-10-17T09:10:51Z

LGTM. Just a clarifying question.

tdas · 2015-10-19T09:30:25Z

Merging this to master and branch 1.5

… to hosts rather than hostports The format of RDD's preferredLocations must be hostname but the format of Streaming Receiver's scheduling executors is hostport. So it doesn't work. This PR converts `schedulerExecutors` to `hosts` before creating Receiver's RDD. Author: zsxwing <zsxwing@gmail.com> Closes #9075 from zsxwing/SPARK-11063. (cherry picked from commit 6758213) Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>

Change preferredLocations of Receiver's RDD to hosts rather than host…

4706ec0

…ports

Add a unit test

2d7c030

zsxwing reviewed Oct 16, 2015
View reviewed changes

Revert the change to core

49b7792

tdas reviewed Oct 17, 2015
View reviewed changes

asfgit closed this in 6758213 Oct 19, 2015

zsxwing deleted the SPARK-11063 branch October 20, 2015 00:40

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-11063][Streaming]Change preferredLocations of Receiver's RDD to hosts rather than hostports #9075

[SPARK-11063][Streaming]Change preferredLocations of Receiver's RDD to hosts rather than hostports #9075

zsxwing commented Oct 12, 2015

SparkQA commented Oct 12, 2015

jerryshao commented Oct 13, 2015

tdas commented Oct 16, 2015

zsxwing commented Oct 16, 2015

jerryshao commented Oct 16, 2015

tdas commented Oct 16, 2015

jerryshao commented Oct 16, 2015

zsxwing Oct 16, 2015

CodingCat Oct 16, 2015

tdas Oct 16, 2015

tdas Oct 16, 2015

zsxwing Oct 17, 2015

CodingCat Oct 17, 2015

zsxwing Oct 17, 2015

zsxwing commented Oct 16, 2015

zsxwing commented Oct 16, 2015

SparkQA commented Oct 16, 2015

SparkQA commented Oct 17, 2015

tdas Oct 17, 2015

zsxwing Oct 17, 2015

tdas commented Oct 17, 2015

tdas commented Oct 19, 2015

[SPARK-11063][Streaming]Change preferredLocations of Receiver's RDD to hosts rather than hostports #9075

[SPARK-11063][Streaming]Change preferredLocations of Receiver's RDD to hosts rather than hostports #9075

Conversation

zsxwing commented Oct 12, 2015

SparkQA commented Oct 12, 2015

jerryshao commented Oct 13, 2015

tdas commented Oct 16, 2015

zsxwing commented Oct 16, 2015

jerryshao commented Oct 16, 2015

tdas commented Oct 16, 2015

jerryshao commented Oct 16, 2015

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

zsxwing commented Oct 16, 2015

zsxwing commented Oct 16, 2015

SparkQA commented Oct 16, 2015

SparkQA commented Oct 17, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tdas commented Oct 17, 2015

tdas commented Oct 19, 2015