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 2022/12/09 15:25:00 UTC
[GitHub] [spark] Ngone51 opened a new pull request, #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Ngone51 opened a new pull request, #39011:
URL: https://github.com/apache/spark/pull/39011
<!--
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'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### 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 avoid rerunning the finished shuffle map task in `TaskSetManager.executorLost()` if the executor lost is caused by decommission and the shuffle data has been successfully migrated.
### 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.
-->
To avoid unnecessary task recomputation.
### 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.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
(Will try to add unit tests tmr)
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-1352796615
> Could you take a look at the failure? It looks like relevant.
The failure is not reproducible every time. I suspect it is still a flaky test. I will keep an eye on 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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 diff in pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1045505675
##########
core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala:
##########
@@ -51,6 +52,7 @@ import org.apache.spark.rdd.RDD
* at the same time for a barrier stage.
*/
private[spark] class ShuffleMapTask(
+ @transient val shuffleId: Int,
Review Comment:
Yea, I think we can do that which is more efficient actually. My original concern was that only the shuffle map stage has the shuffle id. But since `TaskSetManager` works for both the shuffle map stage and the result stage, it could be a bit hacky to add a shuffle id field to `TaskSetManager`.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-1364830733
@mridulm Thanks for the help. Let me take another look. (Sorry for the delay was unhealthy last week...
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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] mridulm commented on a diff in pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1248442046
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1046,17 +1048,45 @@ private[spark] class TaskSetManager(
/** Called by TaskScheduler when an executor is lost so we can re-enqueue our tasks */
override def executorLost(execId: String, host: String, reason: ExecutorLossReason): Unit = {
- // Re-enqueue any tasks that ran on the failed executor if this is a shuffle map stage,
- // and we are not using an external shuffle server which could serve the shuffle outputs.
- // The reason is the next stage wouldn't be able to fetch the data from this dead executor
- // so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
+ // Re-enqueue any tasks with potential shuffle data loss that ran on the failed executor
+ // if this is a shuffle map stage, and we are not using an external shuffle server which
+ // could serve the shuffle outputs or the executor lost is caused by decommission (which
+ // can destroy the whole host). The reason is the next stage wouldn't be able to fetch the
+ // data from this dead executor so we would need to rerun these tasks on other executors.
+ val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
+ (reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)
Review Comment:
Sorry for the delay - this message got lost in my inbox.
You are correct @JoshRosen, we should indeed check for `ExecutorDecommission.workerHost.isDefined`
+CC @Ngone51 in case I am missing something.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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] mridulm commented on pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-1366193206
Merged to master.
Thanks for working on this @Ngone51 !
Thanks for review @dongjoon-hyun :-)
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 #39011: [WIP][SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-1344456134
Mark as WIP first regarding the compilation error and missing ut. Any feedback is still welcome.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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] JoshRosen commented on a diff in pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1177257697
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1046,17 +1048,45 @@ private[spark] class TaskSetManager(
/** Called by TaskScheduler when an executor is lost so we can re-enqueue our tasks */
override def executorLost(execId: String, host: String, reason: ExecutorLossReason): Unit = {
- // Re-enqueue any tasks that ran on the failed executor if this is a shuffle map stage,
- // and we are not using an external shuffle server which could serve the shuffle outputs.
- // The reason is the next stage wouldn't be able to fetch the data from this dead executor
- // so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
+ // Re-enqueue any tasks with potential shuffle data loss that ran on the failed executor
+ // if this is a shuffle map stage, and we are not using an external shuffle server which
+ // could serve the shuffle outputs or the executor lost is caused by decommission (which
+ // can destroy the whole host). The reason is the next stage wouldn't be able to fetch the
+ // data from this dead executor so we would need to rerun these tasks on other executors.
+ val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
+ (reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)
Review Comment:
@Ngone51 @mridulm I have a question about the logic here:
Executor decommissioning does not necessarily imply worker decommissioning. If an external shuffle service is used and the executor is decommissioned without the worker also being decommissioned then the shuffle files will continue to be available at the original host.
Prior to this PR, I don't think this `executorLost` method would not have scheduled re-runs in that case because the `&& !env.blockManager.externalShuffleServiceEnabled` condition would evaluate to `false` when the ESS was used, causing us to skip all of the resubmission logic here.
With this PR, though, I think these changes might actually _cause_ unnecessary task re-submission in that case because the `(reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)` condition would evaluate to true and `locationOpt.exists(_.host != host)` would evaluate to false because the original outputs are still available because no migration is needed.
I think this could be addressed by checking whether [`ExecutorDecommission.workerHost`](https://github.com/apache/spark/blob/b2f9aab8a2cabf04ac3267de0670bdd403831390/core/src/main/scala/org/apache/spark/scheduler/ExecutorLossReason.scala#L83) is defined, i.e. to do
```scala
val workerIsDecommissioned = reason match {
case e: ExecutorDecommission if e.workerHost.isDefined => true
case _ => false
}
val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
(workerIsDecommissioned || !env.blockManager.externalShuffleServiceEnabled)
```
That said, my argument above doesn't hold for Spark-on-Kubernetes because it never sets `workerHost`:
https://github.com/apache/spark/blob/a2a5299cb2d252e85d3b05dbabf7a6db3f488cb2/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesClusterSchedulerBackend.scala#L336-L341
Given this, I am wondering whether this PR's change might represent a regression when Dynamic Allocation is used alongside an external shuffle service in YARN.
WDYT? Am I interpreting the code correctly or have I overlooked something 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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-1344447148
cc @warrenzhu25 too
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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] mridulm commented on a diff in pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1046308698
##########
core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala:
##########
@@ -51,6 +52,7 @@ import org.apache.spark.rdd.RDD
* at the same time for a barrier stage.
*/
private[spark] class ShuffleMapTask(
+ @transient val shuffleId: Int,
Review Comment:
Make it an Option ?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 diff in pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1047069384
##########
core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala:
##########
@@ -51,6 +52,7 @@ import org.apache.spark.rdd.RDD
* at the same time for a barrier stage.
*/
private[spark] class ShuffleMapTask(
+ @transient val shuffleId: Int,
Review Comment:
Updated to use Option.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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
Re: [PR] [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated [spark]
Posted by "09306677806 (via GitHub)" <gi...@apache.org>.
09306677806 commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-2068221273
#(&£€&445₩)*khakqganHynosql sharsing
ShahrzadMahro
در تاریخ یکشنبه ۲۱ آوریل ۲۰۲۴، ۲۲:۳۶ Parth Shyara ***@***.***>
نوشت:
> ***@***.**** commented on this pull request.
> ------------------------------
>
> In core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
> <https://github.com/apache/spark/pull/39011#discussion_r1573887387>:
>
> > + val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
> + (reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)
>
> @Ngone51 <https://github.com/Ngone51> @mridulm
> <https://github.com/mridulm> Is the above issue being tracked elsewhere?
>
> —
> Reply to this email directly, view it on GitHub
> <https://github.com/apache/spark/pull/39011#discussion_r1573887387>, or
> unsubscribe
> <https://github.com/notifications/unsubscribe-auth/ANEZZRHCBDW5SHLP2PE25W3Y6QE45AVCNFSM6AAAAAASZOZVDSVHI2DSMVQWIX3LMV43YUDVNRWFEZLROVSXG5CSMV3GSZLXHMZDAMJTGUYDCMBYHA>
> .
> You are receiving this because you are subscribed to this thread.Message
> ID: ***@***.***>
>
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-1346059915
The failed test seems to be flaky:
```scala
- decommission workers ensure that shuffle output is regenerated even with shuffle service *** FAILED *** (18 seconds, 479 milliseconds)
[info] 5 did not equal 4 Expected 4 tasks but got List(0:0:0:0-SUCCESS, 0:0:0:0-SUCCESS, 0:0:1:0-SUCCESS, 0:0:0:1-SUCCESS, 1:0:0:0-SUCCESS) (DecommissionWorkerSuite.scala:191)
[info] org.scalatest.exceptions.TestFailedException:
[info] at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:472)
[info] at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:471)
[info] at org.scalatest.Assertions$.newAssertionFailedException(Assertions.scala:1231)
[info] at org.scalatest.Assertions$AssertionsHelper.macroAssert(Assertions.scala:1295)
[info] at org.apache.spark.deploy.DecommissionWorkerSuite.$anonfun$new$6(DecommissionWorkerSuite.scala:191)
....
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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] dongjoon-hyun commented on pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-1352828843
Got it. If then, it's okay. Thank you for checking and confirming 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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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
Re: [PR] [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated [spark]
Posted by "parthshyara (via GitHub)" <gi...@apache.org>.
parthshyara commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1573887387
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1046,17 +1048,45 @@ private[spark] class TaskSetManager(
/** Called by TaskScheduler when an executor is lost so we can re-enqueue our tasks */
override def executorLost(execId: String, host: String, reason: ExecutorLossReason): Unit = {
- // Re-enqueue any tasks that ran on the failed executor if this is a shuffle map stage,
- // and we are not using an external shuffle server which could serve the shuffle outputs.
- // The reason is the next stage wouldn't be able to fetch the data from this dead executor
- // so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
+ // Re-enqueue any tasks with potential shuffle data loss that ran on the failed executor
+ // if this is a shuffle map stage, and we are not using an external shuffle server which
+ // could serve the shuffle outputs or the executor lost is caused by decommission (which
+ // can destroy the whole host). The reason is the next stage wouldn't be able to fetch the
+ // data from this dead executor so we would need to rerun these tasks on other executors.
+ val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
+ (reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)
Review Comment:
@Ngone51 @mridulm Is the above issue being tracked elsewhere?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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] mridulm commented on pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-1353493312
The failure is indicating two task end events for the same task ? I dont see how that can happen due to this PR - but wondering how that could have happened in general ... any thoughts @Ngone51 ?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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
Re: [PR] [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated [spark]
Posted by "Ngone51 (via GitHub)" <gi...@apache.org>.
Ngone51 commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1740261348
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1046,17 +1048,45 @@ private[spark] class TaskSetManager(
/** Called by TaskScheduler when an executor is lost so we can re-enqueue our tasks */
override def executorLost(execId: String, host: String, reason: ExecutorLossReason): Unit = {
- // Re-enqueue any tasks that ran on the failed executor if this is a shuffle map stage,
- // and we are not using an external shuffle server which could serve the shuffle outputs.
- // The reason is the next stage wouldn't be able to fetch the data from this dead executor
- // so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
+ // Re-enqueue any tasks with potential shuffle data loss that ran on the failed executor
+ // if this is a shuffle map stage, and we are not using an external shuffle server which
+ // could serve the shuffle outputs or the executor lost is caused by decommission (which
+ // can destroy the whole host). The reason is the next stage wouldn't be able to fetch the
+ // data from this dead executor so we would need to rerun these tasks on other executors.
+ val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
+ (reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)
Review Comment:
@wangyum No. The improvement is applied for both shuffle service on/off. I also replied you in the JIRA https://issues.apache.org/jira/browse/SPARK-49472. I don't think it's a bug introduced by 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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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] mridulm commented on a diff in pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1248442046
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1046,17 +1048,45 @@ private[spark] class TaskSetManager(
/** Called by TaskScheduler when an executor is lost so we can re-enqueue our tasks */
override def executorLost(execId: String, host: String, reason: ExecutorLossReason): Unit = {
- // Re-enqueue any tasks that ran on the failed executor if this is a shuffle map stage,
- // and we are not using an external shuffle server which could serve the shuffle outputs.
- // The reason is the next stage wouldn't be able to fetch the data from this dead executor
- // so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
+ // Re-enqueue any tasks with potential shuffle data loss that ran on the failed executor
+ // if this is a shuffle map stage, and we are not using an external shuffle server which
+ // could serve the shuffle outputs or the executor lost is caused by decommission (which
+ // can destroy the whole host). The reason is the next stage wouldn't be able to fetch the
+ // data from this dead executor so we would need to rerun these tasks on other executors.
+ val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
+ (reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)
Review Comment:
Sorry for the delay - this message got lost in my inbox.
You are correct @JoshRosen, we should indeed check for `ExecutorDecommission.workerHost.isDefined` for standalone
+CC @Ngone51 in case I am missing something.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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
Re: [PR] [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated [spark]
Posted by "wangyum (via GitHub)" <gi...@apache.org>.
wangyum commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1734303934
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1046,17 +1048,45 @@ private[spark] class TaskSetManager(
/** Called by TaskScheduler when an executor is lost so we can re-enqueue our tasks */
override def executorLost(execId: String, host: String, reason: ExecutorLossReason): Unit = {
- // Re-enqueue any tasks that ran on the failed executor if this is a shuffle map stage,
- // and we are not using an external shuffle server which could serve the shuffle outputs.
- // The reason is the next stage wouldn't be able to fetch the data from this dead executor
- // so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
+ // Re-enqueue any tasks with potential shuffle data loss that ran on the failed executor
+ // if this is a shuffle map stage, and we are not using an external shuffle server which
+ // could serve the shuffle outputs or the executor lost is caused by decommission (which
+ // can destroy the whole host). The reason is the next stage wouldn't be able to fetch the
+ // data from this dead executor so we would need to rerun these tasks on other executors.
+ val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
+ (reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)
Review Comment:
@Ngone51 Is this improvement to avoid unnecessary re-run in the scenario of `spark.shuffle.service.enabled=false`?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 diff in pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1044572392
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1046,17 +1046,46 @@ private[spark] class TaskSetManager(
/** Called by TaskScheduler when an executor is lost so we can re-enqueue our tasks */
override def executorLost(execId: String, host: String, reason: ExecutorLossReason): Unit = {
- // Re-enqueue any tasks that ran on the failed executor if this is a shuffle map stage,
- // and we are not using an external shuffle server which could serve the shuffle outputs.
- // The reason is the next stage wouldn't be able to fetch the data from this dead executor
- // so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
+ // Re-enqueue any tasks with potential shuffle data loss that ran on the failed executor
+ // if this is a shuffle map stage, and we are not using an external shuffle server which
+ // could serve the shuffle outputs or the executor lost is caused by decommission (which
+ // can destroy the whole host). The reason is the next stage wouldn't be able to fetch the
+ // data from this dead executor so we would need to rerun these tasks on other executors.
+ val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
+ (reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)
+ if (maybeShuffleMapOutputLoss && !isZombie) {
for ((tid, info) <- taskInfos if info.executorId == execId) {
val index = info.index
+ lazy val isShuffleMapOutputAvailable = reason match {
+ case ExecutorDecommission(_, _) =>
+ val shuffleId = tasks(0).asInstanceOf[ShuffleMapTask].shuffleId
+ val mapId = if (conf.get(config.SHUFFLE_USE_OLD_FETCH_PROTOCOL)) {
+ info.partitionId
+ } else {
+ tid
+ }
+ val locationOpt = env.mapOutputTracker.asInstanceOf[MapOutputTrackerMaster]
+ .getMapOutputLocation(shuffleId, mapId)
+ // There are 3 cases of locationOpt:
+ // 1) locationOpt.isDefined && locationOpt.get.host == host:
+ // this case implies that the shuffle map output is still on the lost executor. The
+ // map output file is supposed to lose so we should rerun this task;
+ // 2) locationOpt.isDefined && locationOpt.get.host != host:
+ // this case implies that the shuffle map output has been migrated to another
+ // host. The task doesn't need to rerun;
+ // 3) location.isEmpty:
Review Comment:
```suggestion
// 3) locationOpt.isEmpty:
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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] mridulm commented on a diff in pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1045132885
##########
core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala:
##########
@@ -51,6 +52,7 @@ import org.apache.spark.rdd.RDD
* at the same time for a barrier stage.
*/
private[spark] class ShuffleMapTask(
+ @transient val shuffleId: Int,
Review Comment:
Do we need to add this ? We can query `DAGScheduler` given a stageId to find the shuffleId (from `ShuffleMapStage.shuffleDep,shuffleId`) - this can be done once, at `TaskSetManager` creation, and persisted as a private field, to make sure it is minimal cost.
Thoughts ?
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1128,6 +1128,17 @@ private[spark] class MapOutputTrackerMaster(
}
}
+ /**
+ * Get map output location by (shuffleId, mapId)
+ */
+ def getMapOutputLocation(shuffleId: Int, mapId: Long): Option[BlockManagerId] = {
+ shuffleStatuses.get(shuffleId).flatMap { shuffleStatus =>
+ shuffleStatus.withMapStatuses { mapStatues =>
+ mapStatues.filter(_ != null).find(_.mapId == mapId).map(_.location)
+ }
Review Comment:
nit: In case of `SHUFFLE_USE_OLD_FETCH_PROTOCOL`, we can possibly simply lookup the index into the array, right ? Instead of needing to scan it ?
Given it is turned off by default, I am fine with leaving it as is though ...
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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
Re: [PR] [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated [spark]
Posted by "wangyum (via GitHub)" <gi...@apache.org>.
wangyum commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1732356898
##########
core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala:
##########
@@ -1046,17 +1048,45 @@ private[spark] class TaskSetManager(
/** Called by TaskScheduler when an executor is lost so we can re-enqueue our tasks */
override def executorLost(execId: String, host: String, reason: ExecutorLossReason): Unit = {
- // Re-enqueue any tasks that ran on the failed executor if this is a shuffle map stage,
- // and we are not using an external shuffle server which could serve the shuffle outputs.
- // The reason is the next stage wouldn't be able to fetch the data from this dead executor
- // so we would need to rerun these tasks on other executors.
- if (isShuffleMapTasks && !env.blockManager.externalShuffleServiceEnabled && !isZombie) {
+ // Re-enqueue any tasks with potential shuffle data loss that ran on the failed executor
+ // if this is a shuffle map stage, and we are not using an external shuffle server which
+ // could serve the shuffle outputs or the executor lost is caused by decommission (which
+ // can destroy the whole host). The reason is the next stage wouldn't be able to fetch the
+ // data from this dead executor so we would need to rerun these tasks on other executors.
+ val maybeShuffleMapOutputLoss = isShuffleMapTasks &&
+ (reason.isInstanceOf[ExecutorDecommission] || !env.blockManager.externalShuffleServiceEnabled)
Review Comment:
It seems that this change may cause task re-run many times and can't complete for a long running stage if
both `spark.decommission.enabled` and `spark.shuffle.service.enabled` are enabled.
<img width="1650" alt="image" src="https://github.com/user-attachments/assets/342de17f-0515-4a18-b2bb-cf5402a0bb9c">
<img width="554" alt="image" src="https://github.com/user-attachments/assets/eae968f4-8bac-4646-824f-a59d39a5741f">
Below is the additional log we added:
```diff
---
.../scala/org/apache/spark/scheduler/TaskSetManager.scala | 8 ++++++--
1 file changed, 6 insertions(+), 2 deletions(-)
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
index a1a54daf5f8..5846827c832 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
@@ -1117,6 +1117,7 @@ private[spark] class TaskSetManager(
/** Called by TaskScheduler when an executor is lost so we can re-enqueue our tasks */
override def executorLost(execId: String, host: String, reason: ExecutorLossReason): Unit = {
+ logInfo(s"Executor lost: execId: $execId, host: $host, reason: $reason.")
// Re-enqueue any tasks with potential shuffle data loss that ran on the failed executor
// if this is a shuffle map stage, and we are not using an external shuffle server which
// could serve the shuffle outputs or the executor lost is caused by decommission (which
@@ -1148,8 +1149,11 @@ private[spark] class TaskSetManager(
// This shouldn't not happen ideally since TaskSetManager handles executor lost first
// before DAGScheduler. So the map statues for the successful task must be available
// at this moment. keep it here in case the handling order changes.
- locationOpt.exists(_.host != host)
-
+ val isShuffleMapOutputLoss = locationOpt.exists(_.host != host)
+ logInfo(s"Is shuffle map output available: partition id: ${info.partitionId}, " +
+ s"tid: ${tid}, locationOpt: ${locationOpt}, " +
+ s"isShuffleMapOutputLoss: ${isShuffleMapOutputLoss}.")
+ isShuffleMapOutputLoss
case _ => false
}
// We may have a running task whose partition has been marked as successful,
```
```
24/08/26 16:56:22 INFO YarnClusterSchedulerBackend: Decommission executors: 1608
24/08/26 16:56:22 INFO YarnClusterSchedulerBackend: Notify executor 1608 to decommission.
24/08/26 16:56:22 INFO BlockManagerMasterEndpoint: Mark BlockManagers (BlockManagerId(1608, hdc42-mcc10-01-0110-7302-007.company.com, 30502, None)) as being decommissioning.
24/08/26 16:56:22 INFO ExecutorAllocationManager: Executors 1608 removed due to idle timeout.
24/08/26 16:56:23 INFO TaskSetManager: Finished task 4992.1 in stage 7.0 (TID 16851) in 807662 ms on hdc42-mcc10-01-0710-4001-001.company.com (executor 1300) (5000/6000)
24/08/26 16:56:23 INFO TaskSetManager: Finished task 1335.2 in stage 7.0 (TID 16713) in 903010 ms on hdc42-mcc10-01-0110-7303-009.company.com (executor 1141) (5001/6000)
24/08/26 16:56:23 INFO TaskSetManager: Finished task 2290.1 in stage 7.0 (TID 16573) in 1115189 ms on hdc42-mcc10-01-1110-3305-038.company (executor 568) (5002/6000)
24/08/26 16:56:23 INFO TaskSetManager: Finished task 349.1 in stage 7.0 (TID 16916) in 777120 ms on hdc42-mcc10-01-0110-5803-003.company.com (executor 1345) (5003/6000)
24/08/26 16:56:23 INFO YarnAllocator: Driver requested a total number of 499 executor(s) for resource profile id: 0.
24/08/26 16:56:24 INFO YarnClusterScheduler: Executor 1608 on hdc42-mcc10-01-0110-7302-007.company.com is decommissioned after 1.3 s.
24/08/26 16:56:24 INFO TaskSetManager: Executor lost: execId: 1608, host: hdc42-mcc10-01-0110-7302-007.company.com, reason: Executor decommission: spark scale down.
24/08/26 16:56:24 INFO TaskSetManager: Is shuffle map output available: partition id: 3749, tid: 10302, locationOpt: Some(BlockManagerId(1608, hdc42-mcc10-01-0110-7302-007.company.com, 7337, None)), isShuffleMapOutputLoss: false.
24/08/26 16:56:24 INFO DAGScheduler: Resubmitted ShuffleMapTask(7, 3749), so marking it as still running.
24/08/26 16:56:24 INFO TaskSetManager: Is shuffle map output available: partition id: 763, tid: 16636, locationOpt: Some(BlockManagerId(1608, hdc42-mcc10-01-0110-7302-007.company.com, 7337, None)), isShuffleMapOutputLoss: false.
24/08/26 16:56:24 INFO DAGScheduler: Resubmitted ShuffleMapTask(7, 763), so marking it as still running.
24/08/26 16:56:24 INFO TaskSetManager: Is shuffle map output available: partition id: 2971, tid: 15433, locationOpt: Some(BlockManagerId(1608, hdc42-mcc10-01-0110-7302-007.company.com, 7337, None)), isShuffleMapOutputLoss: false.
24/08/26 16:56:24 INFO DAGScheduler: Resubmitted ShuffleMapTask(7, 2971), so marking it as still running.
24/08/26 16:56:24 INFO TaskSetManager: Is shuffle map output available: partition id: 5835, tid: 16587, locationOpt: Some(BlockManagerId(1608, hdc42-mcc10-01-0110-7302-007.company.com, 7337, None)), isShuffleMapOutputLoss: false.
24/08/26 16:56:24 INFO DAGScheduler: Resubmitted ShuffleMapTask(7, 5835), so marking it as still running.
24/08/26 16:56:24 INFO TaskSetManager: Is shuffle map output available: partition id: 611, tid: 15118, locationOpt: Some(BlockManagerId(1608, hdc42-mcc10-01-0110-7302-007.company.com, 7337, None)), isShuffleMapOutputLoss: false.
24/08/26 16:56:24 INFO DAGScheduler: Resubmitted ShuffleMapTask(7, 611), so marking it as still running.
24/08/26 16:56:24 INFO TaskSetManager: Is shuffle map output available: partition id: 3750, tid: 10303, locationOpt: Some(BlockManagerId(1608, hdc42-mcc10-01-0110-7302-007.company.com, 7337, None)), isShuffleMapOutputLoss: false.
24/08/26 16:56:24 INFO DAGScheduler: Resubmitted ShuffleMapTask(7, 3750), so marking it as still running.
24/08/26 16:56:24 INFO TaskSetManager: Is shuffle map output available: partition id: 3740, tid: 13610, locationOpt: Some(BlockManagerId(1608, hdc42-mcc10-01-0110-7302-007.company.com, 7337, None)), isShuffleMapOutputLoss: false.
24/08/26 16:56:24 INFO DAGScheduler: Resubmitted ShuffleMapTask(7, 3740), so marking it as still running.
24/08/26 16:56:24 INFO DAGScheduler: Executor lost: 1608 (epoch 1)
24/08/26 16:56:24 INFO BlockManagerMasterEndpoint: Trying to remove executor 1608 from BlockManagerMaster.
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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] mridulm closed pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
mridulm closed pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
URL: https://github.com/apache/spark/pull/39011
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-1366436959
Thanks @mridulm @dongjoon-hyun
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 diff in pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1045502828
##########
core/src/main/scala/org/apache/spark/MapOutputTracker.scala:
##########
@@ -1128,6 +1128,17 @@ private[spark] class MapOutputTrackerMaster(
}
}
+ /**
+ * Get map output location by (shuffleId, mapId)
+ */
+ def getMapOutputLocation(shuffleId: Int, mapId: Long): Option[BlockManagerId] = {
+ shuffleStatuses.get(shuffleId).flatMap { shuffleStatus =>
+ shuffleStatus.withMapStatuses { mapStatues =>
+ mapStatues.filter(_ != null).find(_.mapId == mapId).map(_.location)
+ }
Review Comment:
> nit: In case of SHUFFLE_USE_OLD_FETCH_PROTOCOL, we can possibly simply lookup the index into the array, right ? Instead of needing to scan it ?
Right. It's actually a good idea.
> Given it is turned off by default, I am fine with leaving it as is though ...
Yes, prefer to leave it as is. Thanks.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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 diff in pull request #39011: [SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #39011:
URL: https://github.com/apache/spark/pull/39011#discussion_r1045505675
##########
core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala:
##########
@@ -51,6 +52,7 @@ import org.apache.spark.rdd.RDD
* at the same time for a barrier stage.
*/
private[spark] class ShuffleMapTask(
+ @transient val shuffleId: Int,
Review Comment:
Yea, I think we can do that which is more efficient actually. My original concern was that only the shuffle map stage has the shuffle id. But since `TaskSetManager` works for both the shuffle map stage and the result stage, it could be a bit hacky to add a shuffle id field to `TaskSetManager`.
WDYT?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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] warrenzhu25 commented on pull request #39011: [WIP][SPARK-41469][CORE] Avoid unnecessary task rerun on decommissioned executor lost if shuffle data migrated
Posted by GitBox <gi...@apache.org>.
warrenzhu25 commented on PR #39011:
URL: https://github.com/apache/spark/pull/39011#issuecomment-1344553081
> cc @warrenzhu25 too
It's really the change I want. Great work.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
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