You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2020/05/27 15:17:22 UTC
[GitHub] [spark] Ngone51 opened a new pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Ngone51 opened a new pull request #28656:
URL: https://github.com/apache/spark/pull/28656
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
This PR proposes to shift to the new highest locality level if there're any new higher locality levels are added during `TaskSetManager.recomputeLocality`.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
In the new version of delay scheduling, if a task set is submitted before any executors are added to TaskScheduler, the task set will always schedule all tasks at ANY locality level. It means the delay scheduling can not work in this case, and this PR fixes the problem.
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
No.
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
-->
Updated and added new test.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r431537862
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,16 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
Review comment:
> I think we should change this to only happen when executorAdded. this is also called on lost and decommission and it doesn't make sense to go to "lower" level.
I think it's impossible to go to "lower" or more local level in case of lost and decommission. Lost and decommission would remove executors, so the locality levels can only be less compared to the previous locality levels. It also means, lost and decommission will not add new more local levels.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636073919
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/123290/
Test FAILed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-637230641
thanks all!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636419309
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan edited a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan edited a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634774628
This reminds me of https://github.com/apache/spark/pull/27207#discussion_r400237091
It's not a big problem, as we only violate delay scheduling in the first resource offer. But still good to fix it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636419232
**[Test build #123325 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123325/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r431897015
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,16 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
+ // SPARK-31837: there's new higher locality level, so shift to
Review comment:
how about `If the new level is more local, shift to the most local level`?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636073908
Merged build finished. Test FAILed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r432318441
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,19 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
+ val previousMyLocalityLevels = myLocalityLevels
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
+ // SPARK-31837: If the new level is more local, shift to the new most local locality
+ // level in terms of better data locality. For example, say the previous locality
+ // levels are [PROCESS, NODE, ANY] and current level is ANY. After recompute, the
+ // locality levels are [PROCESS, NODE, RACK, ANY]. Then, we'll shift to RACK level.
+ currentLocalityIndex = getLocalityIndex(myLocalityLevels.diff(previousMyLocalityLevels).head)
Review comment:
Hi all, there's a defect in the previous implement(always reset `currentLocalityIndex` to 0). Think about such a case, say we have locality levels [PROCESS, NODE, ANY] and current locality level is ANY. After recompute, we might have locality levels [PROCESS, NODE, RACK, ANY]. In this case, I think we'd better shit to RACK level instead of PROCESS level, since the TaskSetManager has been already delayed for a while on known levels(PROCESS, NODE). So with this update, I think it could also ease our concern on the possible perf regression introduced by aggressive locality level resetting. @bmarcott @tgravescs @cloud-fan
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 edited a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 edited a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634735188
ping @bmarcott @cloud-fan @tgravescs
also cc @jiangxb1987 @dongjoon-hyun since this's the root cause for the flaky of #28584
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635821738
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636419232
**[Test build #123325 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123325/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634781230
**[Test build #123195 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123195/testReport)** for PR 28656 at commit [`cdc9706`](https://github.com/apache/spark/commit/cdc97066a0896dac4f10d0dc2c0b3c65032f768f).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634875195
Merged build finished. Test FAILed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635377056
ah, sorry I missed that, it is covered. This looks good to me then. I can file another jlira to investigate possibly handling better.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636010492
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r431230286
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,16 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
+ // SPARK-31837: there's new higher level locality, so shift to
+ // the highest locality level in terms of better data locality
+ currentLocalityIndex = 0
Review comment:
My only concern of this change is, for general cases, whether the task set would take more time on delay scheduling and thus reduce the throughput of the cluster. But considering the improvement that we gain(to increase the throughput of the cluster) from the new version of the delay scheduling, I think it should not be a big problem.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634847978
that isn't true if you are on something like YARN and HDFS where data can be on the node.
Ideally I think we reset it when executor added if there aren't any or if it all executors were used and we added one, I just need time to look through the code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635400642
can we also remove the hack mentioned in https://github.com/apache/spark/pull/27207#discussion_r400237091 ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636432446
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/123325/
Test FAILed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635821157
**[Test build #123273 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123273/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636009835
**[Test build #123290 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123290/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635654676
I filed a followup jira https://issues.apache.org/jira/browse/SPARK-31856?filter=-2 to look at handling the executor added better, this fix works for now.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r432524274
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,19 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
+ val previousMyLocalityLevels = myLocalityLevels
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
+ // SPARK-31837: If the new level is more local, shift to the new most local locality
+ // level in terms of better data locality. For example, say the previous locality
+ // levels are [PROCESS, NODE, ANY] and current level is ANY. After recompute, the
+ // locality levels are [PROCESS, NODE, RACK, ANY]. Then, we'll shift to RACK level.
+ currentLocalityIndex = getLocalityIndex(myLocalityLevels.diff(previousMyLocalityLevels).head)
Review comment:
> We can certainly handle some here if you want
What do you mean by "handle some here"? I read your JIRA and don't find the specific solution that could be added to this PR. Could you please elaborate more?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bmarcott commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
bmarcott commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635651858
@Ngone51
Thanks for looking into this.
Let's make a few corrections to the description:
This happened with previous delay scheduling as well (although likely less noticeable as @tgravescs pointed out)
It won't "always" use ANY for "all" tasks in the taskset. For example if a single executor is added that is NODE_LOCAL and is used for scheduling, the very first resourceOffer would switch the taskset to NODE_LOCAL.
I do see how this handles the race condition on startup, although a more specific handling of that case may be preferable (@tgravescs also mentioned).
It seems questionable whether executor added should instantly reset the locality levels.
Deferring to @tgravescs and @cloud-fan to make the call here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635193973
**[Test build #123225 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123225/testReport)** for PR 28656 at commit [`312700a`](https://github.com/apache/spark/commit/312700a7ba51d4a04b24cfdceda9c5d018767db9).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636008321
retest this please.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634735785
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] bmarcott edited a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
bmarcott edited a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635651858
@Ngone51
Thanks for looking into this.
Let's make a few corrections to the description:
- This happened with previous delay scheduling as well (although likely less noticeable as @tgravescs pointed out)
- It won't "always" use ANY for "all" tasks in the taskset. For example if a single executor is added that is NODE_LOCAL and is used for scheduling, the very first resourceOffer would switch the taskset to NODE_LOCAL.
I do see how this handles the race condition on startup, although a more specific handling of that case may be preferable (@tgravescs also mentioned).
It seems questionable whether executor added should instantly reset the locality levels.
Deferring to @tgravescs and @cloud-fan to make the call here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636073366
**[Test build #123290 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123290/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636432445
Merged build finished. Test FAILed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636495436
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634774628
This reminds me of https://github.com/apache/spark/pull/27207#discussion_r400237091
It's not a big problem, as we only violate delay scheduling in the first resource offset. But still good to fix it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
tgravescs commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r432515370
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,19 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
+ val previousMyLocalityLevels = myLocalityLevels
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
+ // SPARK-31837: If the new level is more local, shift to the new most local locality
+ // level in terms of better data locality. For example, say the previous locality
+ // levels are [PROCESS, NODE, ANY] and current level is ANY. After recompute, the
+ // locality levels are [PROCESS, NODE, RACK, ANY]. Then, we'll shift to RACK level.
+ currentLocalityIndex = getLocalityIndex(myLocalityLevels.diff(previousMyLocalityLevels).head)
Review comment:
yes this is one of the cases I was referring to. Ideally you would never run into this case because a host is on a rack so you would always have it. Unfortunately Spark defaults the rack to None so you can. I was going to improve upon it in the jira I filed. We can certainly handle some here if you want
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634781230
**[Test build #123195 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123195/testReport)** for PR 28656 at commit [`cdc9706`](https://github.com/apache/spark/commit/cdc97066a0896dac4f10d0dc2c0b3c65032f768f).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636469213
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636469088
**[Test build #123346 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123346/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635903864
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634734951
**[Test build #123193 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123193/testReport)** for PR 28656 at commit [`c38ace4`](https://github.com/apache/spark/commit/c38ace4d52265160ced2574b28447c93c1ed8f91).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636495436
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 edited a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 edited a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634776960
@tgravescs
The problem is caused by the race condition between `resourcesOffer` and `submitTasks`. If `submitTasks` happens before `resourcesOffer`, especially when there are no executors added to TaskScheduler at all, the TaskSetManager 's `myLocalityLevels` will only have ANY locality level, see:
https://github.com/apache/spark/blob/8b4862953a879a9b3ba6f57e669efc383df68b7c/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L218
And then, `resourcesOffer` is called with new executors added to TaskScheduler. Then, `recomputeLocality` will be called because of `executorAdded`. During `recomputeLocality`, the TaskSetManager's `myLocalityLevels` may have PROCESS_LOCAL, NODE_LOCAL, ANY. But it will stick to the previous locality level, which is ANY, see:
https://github.com/apache/spark/blob/8b4862953a879a9b3ba6f57e669efc383df68b7c/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L1107-L1114
As a result, the TaskSetManager can only schedule tasks at ANY level.
(I'll update the PR description with more detail explanation later)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636009835
**[Test build #123290 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123290/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636468642
retest this please
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636432445
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635903864
Merged build finished. Test FAILed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r431538125
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,16 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
Review comment:
> so perhaps don't say higher or lower but say more local or less local.
Yea, good point!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634875202
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/123195/
Test FAILed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635185856
> can we fix https://github.com/apache/spark/pull/28584/files#diff-aa84904ad2d5693a375c82005b66824dR284 in this PR?
Sure.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636469213
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634875195
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634735785
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634877071
**[Test build #123193 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123193/testReport)** for PR 28656 at commit [`c38ace4`](https://github.com/apache/spark/commit/c38ace4d52265160ced2574b28447c93c1ed8f91).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
tgravescs commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r431477308
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,16 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
Review comment:
I think we should change this to only happen when executorAdded. this is also called on lost and decommission and it doesn't make sense to go to "lower" level. Note we may want to stay away from saying higher in the comment below. The code values, lower is actually more strict - meaning process is lowest value. so perhaps don't say higher or lower but say more local or less local. Perhaps pass in parameter from executorAdded.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636073908
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634734951
**[Test build #123193 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123193/testReport)** for PR 28656 at commit [`c38ace4`](https://github.com/apache/spark/commit/c38ace4d52265160ced2574b28447c93c1ed8f91).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636419089
retest this please
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634764770
can you describe the problem in more detail. Is it these tasks are coming in with preferred locality but by the time you get an executor it has already waited the timeout for each of process, node, rack, so you get to ANY?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 edited a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 edited a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634776960
@tgravescs
The problem is caused by the race condition between `resourcesOffer` and `submitTasks`. If `submitTasks` happens before `resourcesOffer`, especially when there are no executors added to TaskScheduler at all, the TaskSetManager 's `myLocalityLevels` will only have ANY locality level, see:
https://github.com/apache/spark/blob/8b4862953a879a9b3ba6f57e669efc383df68b7c/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L218
And then, `resourcesOffer` is called with new executors added to TaskScheduler. Then, `recomputeLocality` will be called because of `executorAdded`. During `recomputeLocality`, the TaskSetManager's `myLocalityLevels` may have PROCESS_LOCAL, NODE_LOCAL, ANY. But the TaskSetManager will stick to the previous locality level, which is ANY, see:
https://github.com/apache/spark/blob/8b4862953a879a9b3ba6f57e669efc383df68b7c/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L1107-L1114
As a result, the TaskSetManager can only schedule tasks at ANY level.
(I'll update the PR description with more detail explanation later)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636850009
thanks, merging to master!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634782076
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636495133
**[Test build #123346 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123346/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] okimcoder removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
okimcoder removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635650972
@Ngone51
Thanks for looking into this.
Let's make a few corrections to the description:
1. This happened with previous delay scheduling as well (although likely less noticeable as @tgravescs pointed out)
2. It won't "always" use ANY for "all" tasks in the taskset. For example if a single executor is added that is NODE_LOCAL and is used for scheduling, the very first resourceOffer would switch the taskset to NODE_LOCAL.
I do see how this handles the race condition on startup, although a more specific handling of that case may be preferable (@tgravescs also mentioned).
It seems questionable whether executor added should instantly reset the locality levels.
Deferring to @tgravescs and @cloud-fan to make the call here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 edited a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 edited a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634776960
@tgravescs
The problem is caused by the race condition between `resourcesOffer` and `submitTasks`. If `submitTasks` happens before `resourcesOffer`, especially when there are no executors added to TaskScheduler at all, the TaskSetManager 's `myLocalityLevels` will only have ANY locality level, see:
https://github.com/apache/spark/blob/8b4862953a879a9b3ba6f57e669efc383df68b7c/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L218
And then, `resourcesOffer` is called with new executors added to TaskScheduler. Then, `recomputeLocality` will be called because of `executorAdded`. During `recomputeLocality`, the TaskSetManager's `myLocalityLevels` may have PROCESS_LOCAL, NODE_LOCAL, ANY(because at this time we have living executors). But the TaskSetManager will stick to the previous locality level, which is ANY, see:
https://github.com/apache/spark/blob/8b4862953a879a9b3ba6f57e669efc383df68b7c/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L1107-L1114
As a result, the TaskSetManager can only schedule tasks at ANY level.
(I'll update the PR description with more detail explanation later)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634878627
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r432318441
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,19 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
+ val previousMyLocalityLevels = myLocalityLevels
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
+ // SPARK-31837: If the new level is more local, shift to the new most local locality
+ // level in terms of better data locality. For example, say the previous locality
+ // levels are [PROCESS, NODE, ANY] and current level is ANY. After recompute, the
+ // locality levels are [PROCESS, NODE, RACK, ANY]. Then, we'll shift to RACK level.
+ currentLocalityIndex = getLocalityIndex(myLocalityLevels.diff(previousMyLocalityLevels).head)
Review comment:
Hi all, there's a defect in the previous implement(always reset `currentLocalityIndex` to 0). Think about such a case, say we have locality levels [PROCESS, NODE, ANY] and current locality level is ANY. After recompute, we might have locality levels [PROCESS, NODE, RACK, ANY]. In this case, I think we'd better shift to RACK level instead of PROCESS level, since the TaskSetManager has been already delayed for a while on known levels(PROCESS, NODE). So with this update, I think it could also ease our concern on the possible perf regression introduced by aggressive locality level resetting. @bmarcott @tgravescs @cloud-fan
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636469088
**[Test build #123346 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123346/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634823664
> when you have dynamic allocation and you are adding a bunch of executors.
For newly allocated executors, mostly they don't have any local data to trigger the level reset. To me, this is mostly for the first resource offer to a task set manager.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634878627
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635191530
> So another thought here is to only reset this when the locality levels actually changed.
I think the current condition(`if (currentLocalityIndex > previousLocalityIndex)`) could satisfy the requirement. If the `currentLocalityIndex` is changed to be bigger compared to the `previousLocalityIndex`, it means there're new locality levels added before the current locality level.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635821157
**[Test build #123273 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123273/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635903870
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/123273/
Test FAILed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635903144
**[Test build #123273 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123273/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
tgravescs commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r431253949
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,16 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
+ // SPARK-31837: there's new higher level locality, so shift to
+ // the highest locality level in terms of better data locality
+ currentLocalityIndex = 0
Review comment:
I have concerns with this. This is going to reset it back to highest level all the time and then you will have to work your way back through the levels, waiting at each level. This could really delay things a lot if you are down to the ANY level and you have to wait 3 seconds between each one. The problem you are describing is only on startup when you have no executors, correct? Perhaps we can look at something more specific for that.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634776960
@tgravescs
The problem is caused by the race condition between `resourcesOffer` and `submitTasks`. If `submitTasks` happens before `resourcesOffer`, especially when there are no executors added to TaskScheduler at all, the TaskSetManager 's `myLocalityLevels` will have ANY locality level, see:
https://github.com/apache/spark/blob/8b4862953a879a9b3ba6f57e669efc383df68b7c/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L218
And then, `resourcesOffer` is called with new executors added to TaskScheduler. Then, `recomputeLocality` will be called because of `executorAdded`. During `recomputeLocality`, the TaskSetManager's `myLocalityLevels` may have PROCESS_LOCAL, NODE_LOCAL, ANY. But it will stick to the previous locality level, which is ANY, see:
https://github.com/apache/spark/blob/8b4862953a879a9b3ba6f57e669efc383df68b7c/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala#L1107-L1114
As a result, the TaskSetManager can only schedule tasks at ANY level.
(I'll update the PR description with more detail explanation later)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635194757
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan closed pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan closed pull request #28656:
URL: https://github.com/apache/spark/pull/28656
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635821738
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] Ngone51 commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634735188
ping @bmarcott @cloud-fan @tgravescs
also cc @jiangxb1987 @dongjoon-hyun since this's the root cause leads to the flaky of #28584
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] okimcoder commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
okimcoder commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635650972
@Ngone51
Thanks for looking into this.
Let's make a few corrections to the description:
1. This happened with previous delay scheduling as well (although likely less noticeable as @tgravescs pointed out)
2. It won't "always" use ANY for "all" tasks in the taskset. For example if a single executor is added that is NODE_LOCAL and is used for scheduling, the very first resourceOffer would switch the taskset to NODE_LOCAL.
I do see how this handles the race condition on startup, although a more specific handling of that case may be preferable (@tgravescs also mentioned).
It seems questionable whether executor added should instantly reset the locality levels.
Deferring to @tgravescs and @cloud-fan to make the call here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635166623
can we fix https://github.com/apache/spark/pull/28584/files#diff-aa84904ad2d5693a375c82005b66824dR284 in this PR?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636432277
**[Test build #123325 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123325/testReport)** for PR 28656 at commit [`8744f1e`](https://github.com/apache/spark/commit/8744f1ec02db28975a80410b5a01d2c2557ad216).
* This patch **fails due to an unknown error code, -9**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636419309
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r431253567
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,16 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
+ // SPARK-31837: there's new higher level locality, so shift to
+ // the highest locality level in terms of better data locality
+ currentLocalityIndex = 0
Review comment:
Is the new higher locality level always 0? Can it be HOST_LOCAL?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
tgravescs commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634790789
Right when you have no tasks yet it gets set to ANY, but this wasn't changed, that was like that before (the changes to locality also) and when an executor was added it never reset it back to the highest, it just called recomputeLocality again which would have left it at ANY as well.
I'm assuming the difference here is that previously it would reset it as soon as any task got assigned at a higher locality. Note that wasn't necessarily all the way back to 0 though.
The issue I think on resetting it on every recomputeLocality is when you have dynamic allocation and you are adding a bunch of executors. You basically revert back to similar to the old behavior where you could really be delaying vs using executors you already have. You are delaying those tasks that could be running more than we should and wasting the executor resources you have.
let me look into this a bit more.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634874317
**[Test build #123195 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/123195/testReport)** for PR 28656 at commit [`cdc9706`](https://github.com/apache/spark/commit/cdc97066a0896dac4f10d0dc2c0b3c65032f768f).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] cloud-fan commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r431254185
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,16 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
+ // SPARK-31837: there's new higher level locality, so shift to
Review comment:
level locality -> locality level
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-635194757
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-636010492
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #28656:
URL: https://github.com/apache/spark/pull/28656#issuecomment-634782076
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] tgravescs commented on a change in pull request #28656: [SPARK-31837][CORE] Shift to the new highest locality level if there is when recomputeLocality
Posted by GitBox <gi...@apache.org>.
tgravescs commented on a change in pull request #28656:
URL: https://github.com/apache/spark/pull/28656#discussion_r432528304
##########
File path: core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
##########
@@ -1107,10 +1107,19 @@ private[spark] class TaskSetManager(
def recomputeLocality(): Unit = {
// A zombie TaskSetManager may reach here while executorLost happens
if (isZombie) return
+ val previousLocalityIndex = currentLocalityIndex
val previousLocalityLevel = myLocalityLevels(currentLocalityIndex)
+ val previousMyLocalityLevels = myLocalityLevels
myLocalityLevels = computeValidLocalityLevels()
localityWaits = myLocalityLevels.map(getLocalityWait)
currentLocalityIndex = getLocalityIndex(previousLocalityLevel)
+ if (currentLocalityIndex > previousLocalityIndex) {
+ // SPARK-31837: If the new level is more local, shift to the new most local locality
+ // level in terms of better data locality. For example, say the previous locality
+ // levels are [PROCESS, NODE, ANY] and current level is ANY. After recompute, the
+ // locality levels are [PROCESS, NODE, RACK, ANY]. Then, we'll shift to RACK level.
+ currentLocalityIndex = getLocalityIndex(myLocalityLevels.diff(previousMyLocalityLevels).head)
Review comment:
I just mean there are a bunch of corner cases and I don't think resetting this on every executor added is ideal. I did not list out all of them. On Yarn it actually defaults to default-rack rather then None so it actually won't have this issue because every node has a rack. I agree that the code you have here is an improvement to handle the rack case.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org