[SPARK-35546][Shuffle] Enable push-based shuffle when multiple app attempts are enabled and manage concurrent access to the state in a better way #33078

zhouyejoe · 2021-06-25T06:53:41Z

What changes were proposed in this pull request?

This is one of the patches for SPIP SPARK-30602 which is needed for push-based shuffle.

Summary of the change:

When Executor registers with Shuffle Service, it will encode the merged shuffle dir created and also the application attemptId into the ShuffleManagerMeta into Json. Then in Shuffle Service, it will decode the Json string and get the correct merged shuffle dir and also the attemptId. If the registration comes from a newer attempt, the merged shuffle information will be updated to store the information from the newer attempt.

This PR also refactored the management of the merged shuffle information to avoid concurrency issues.

Why are the changes needed?

Refer to the SPIP in SPARK-30602.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added unit tests.
The reference PR with the consolidated changes covering the complete implementation is also provided in SPARK-30602.
We have already verified the functionality and the improved performance as documented in the SPIP doc.

zhouyejoe · 2021-06-25T06:56:09Z

PR created. CC @Ngone51 @mridulm @Victsm @otterc

mridulm

Took an initial pass through the PR

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java

mridulm · 2021-06-25T08:04:57Z

Jenkins, test this please.

mridulm · 2021-06-25T08:05:02Z

Ok to test

SparkQA · 2021-06-25T09:33:15Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44847/

SparkQA · 2021-06-25T10:10:06Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44847/

SparkQA · 2021-06-25T11:31:59Z

Test build #140315 has finished for PR 33078 at commit 65fca88.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public static class AppShuffleInfo

otterc

I don't see any UTs added for all the issues that were identified with concurrency. Is there a problem in adding UTs for these cases?

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java

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

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java

zhouyejoe · 2021-06-27T01:00:18Z

Thanks @mridulm @otterc @venkata91 for reviewing the PR. Addressed majority of the comments.
@otterc The unit test to simulate the concurrency control is yet to be added. A little bit tricky to get those added.

Hi, @Ngone51 . Can you also help review this one? Thanks.

mridulm

Looks mostly good, had a couple of comments.
Will wait for @Ngone51, @otterc and @venkata91's reviews to complete as well.

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java

core/src/main/scala/org/apache/spark/storage/BlockManager.scala

...network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/PushBlockStream.java

core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java

SparkQA · 2021-06-29T22:27:52Z

Test build #140368 has finished for PR 33078 at commit 1f915b1.

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

SparkQA · 2021-06-29T23:17:21Z

Kubernetes integration test unable to build dist.

exiting with code: 141
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/44900/

mridulm · 2021-07-01T02:07:05Z

@Ngone51 I am leaving on slightly long-ish vacation without access to my laptop .. will it be possible for you to shepard this pr and #33034 ?
These are the last two pending fixes for push based shuffle, and deal with correctness issues (and not functionality issues).
If you feel we cant get them through, @otterc will have a pr for branch-3.2 ready which will disable push based shuffle for 3.2 - we can merge that instead to prevent issues for users trying to leverage push based shuffle in later versions.
Thanks !

Ngone51 · 2021-07-01T02:54:17Z

@mridulm Sure! Enjoy your vacation!

Ngone51 · 2021-07-02T07:34:31Z

@zhouyejoe Could you address all the comments? Branch-3.2 is already cut, we now have very limited time to merge this into it.

zhouyejoe · 2021-07-02T18:20:57Z

@Ngone51 Sure, will update the PR ASAP.

core/src/main/scala/org/apache/spark/internal/config/package.scala

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java

SparkQA · 2021-07-05T02:37:46Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45136/

SparkQA · 2021-07-05T03:11:58Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45136/

SparkQA · 2021-07-17T00:49:35Z

Test build #141187 has finished for PR 33078 at commit fab51d4.

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

SparkQA · 2021-07-17T01:11:23Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45703/

SparkQA · 2021-07-17T02:16:53Z

Test build #141191 has finished for PR 33078 at commit 79816ad.

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

...network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/PushBlockStream.java

zhouyejoe · 2021-07-18T06:09:13Z

Are there any UTs that verify the changes to finalizeShuffleMerge when the message is from old attempt?

Added a unit test for this one yesterday.

SparkQA · 2021-07-18T07:25:56Z

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45721/

SparkQA · 2021-07-18T09:22:33Z

Test build #141208 has finished for PR 33078 at commit 1de689b.

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

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java

mridulm · 2021-07-19T14:53:36Z

Thanks for the approve @Ngone51 !
+CC @otterc, @venkata91 can you take another pass please ? I plan to merge this later today to master and branch-3.2.

venkata91 · 2021-07-19T16:49:40Z

LGTM.

otterc · 2021-07-19T17:05:33Z

Looks good to me. Thanks for working on it @zhouyejoe

zhouyejoe · 2021-07-19T19:11:07Z

Thanks for all the reviews.

mridulm · 2021-07-19T22:17:23Z

Jenkins, test this please

SparkQA · 2021-07-19T23:43:55Z

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45798/

SparkQA · 2021-07-20T00:17:58Z

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45798/

SparkQA · 2021-07-20T01:33:17Z

Kubernetes integration test unable to build dist.

exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/45800/

SparkQA · 2021-07-20T01:39:00Z

Test build #141284 has finished for PR 33078 at commit 1de689b.

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

SparkQA · 2021-07-20T03:39:52Z

Test build #141286 has finished for PR 33078 at commit 5310991.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public class ShuffleChecksumHelper
class MutableCheckedOutputStream(out: OutputStream) extends OutputStream
case class ShuffleChecksumBlockId(shuffleId: Int, mapId: Long, reduceId: Int) extends BlockId
case class SessionWindow(timeColumn: Expression, gapDuration: Long) extends UnaryExpression
protected abstract class ConnectionProviderBase extends Logging
case class SessionWindowStateStoreRestoreExec(
case class SessionWindowStateStoreSaveExec(

…tempts are enabled and manage concurrent access to the state in a better way ### What changes were proposed in this pull request? This is one of the patches for SPIP SPARK-30602 which is needed for push-based shuffle. ### Summary of the change: When Executor registers with Shuffle Service, it will encode the merged shuffle dir created and also the application attemptId into the ShuffleManagerMeta into Json. Then in Shuffle Service, it will decode the Json string and get the correct merged shuffle dir and also the attemptId. If the registration comes from a newer attempt, the merged shuffle information will be updated to store the information from the newer attempt. This PR also refactored the management of the merged shuffle information to avoid concurrency issues. ### Why are the changes needed? Refer to the SPIP in SPARK-30602. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added unit tests. The reference PR with the consolidated changes covering the complete implementation is also provided in SPARK-30602. We have already verified the functionality and the improved performance as documented in the SPIP doc. Closes #33078 from zhouyejoe/SPARK-35546. Authored-by: Ye Zhou <yezhou@linkedin.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com> (cherry picked from commit c77acf0) Signed-off-by: Mridul Muralidharan <mridulatgmail.com>

mridulm · 2021-07-20T05:05:34Z

Merged to master and branch-3.2
Thanks for working on this @zhouyejoe !
Thanks for all the reviews @Ngone51, @otterc, @venkata91 :-)

mridulm · 2021-07-20T05:06:14Z

+CC @gengliangwang

dongjoon-hyun · 2021-07-23T20:46:48Z

...rk-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/FinalizeShuffleMerge.java

@@ -64,25 +68,28 @@ public boolean equals(Object other) {
    if (other != null && other instanceof FinalizeShuffleMerge) {
      FinalizeShuffleMerge o = (FinalizeShuffleMerge) other;
      return Objects.equal(appId, o.appId)
+        && appAttemptId == appAttemptId


This is fixed by the following.

[SPARK-36273][Shuffle] Fix identical values comparison #33493

…tempts are enabled and manage concurrent access to the state in a better way This is one of the patches for SPIP SPARK-30602 which is needed for push-based shuffle. When Executor registers with Shuffle Service, it will encode the merged shuffle dir created and also the application attemptId into the ShuffleManagerMeta into Json. Then in Shuffle Service, it will decode the Json string and get the correct merged shuffle dir and also the attemptId. If the registration comes from a newer attempt, the merged shuffle information will be updated to store the information from the newer attempt. This PR also refactored the management of the merged shuffle information to avoid concurrency issues. Refer to the SPIP in SPARK-30602. No. Added unit tests. The reference PR with the consolidated changes covering the complete implementation is also provided in SPARK-30602. We have already verified the functionality and the improved performance as documented in the SPIP doc. Closes apache#33078 from zhouyejoe/SPARK-35546. Authored-by: Ye Zhou <yezhou@linkedin.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com> (cherry picked from commit c77acf0)

…tempts are enabled and manage concurrent access to the state in a better way ### What changes were proposed in this pull request? This is one of the patches for SPIP SPARK-30602 which is needed for push-based shuffle. ### Summary of the change: When Executor registers with Shuffle Service, it will encode the merged shuffle dir created and also the application attemptId into the ShuffleManagerMeta into Json. Then in Shuffle Service, it will decode the Json string and get the correct merged shuffle dir and also the attemptId. If the registration comes from a newer attempt, the merged shuffle information will be updated to store the information from the newer attempt. This PR also refactored the management of the merged shuffle information to avoid concurrency issues. ### Why are the changes needed? Refer to the SPIP in SPARK-30602. ### Does this PR introduce _any_ user-facing change? No. ### How was this patch tested? Added unit tests. The reference PR with the consolidated changes covering the complete implementation is also provided in SPARK-30602. We have already verified the functionality and the improved performance as documented in the SPIP doc. Closes apache#33078 from zhouyejoe/SPARK-35546. Authored-by: Ye Zhou <yezhou@linkedin.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com> (cherry picked from commit c77acf0) Signed-off-by: Mridul Muralidharan <mridulatgmail.com>

…tempts are enabled and manage concurrent access to the state in a better way This is one of the patches for SPIP SPARK-30602 which is needed for push-based shuffle. When Executor registers with Shuffle Service, it will encode the merged shuffle dir created and also the application attemptId into the ShuffleManagerMeta into Json. Then in Shuffle Service, it will decode the Json string and get the correct merged shuffle dir and also the attemptId. If the registration comes from a newer attempt, the merged shuffle information will be updated to store the information from the newer attempt. This PR also refactored the management of the merged shuffle information to avoid concurrency issues. Refer to the SPIP in SPARK-30602. No. Added unit tests. The reference PR with the consolidated changes covering the complete implementation is also provided in SPARK-30602. We have already verified the functionality and the improved performance as documented in the SPIP doc. Closes #33078 from zhouyejoe/SPARK-35546. Authored-by: Ye Zhou <yezhou@linkedin.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

github-actions bot added the CORE label Jun 25, 2021

mridulm reviewed Jun 25, 2021

View reviewed changes

otterc reviewed Jun 25, 2021

View reviewed changes

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java Outdated Show resolved Hide resolved

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java Outdated Show resolved Hide resolved

venkata91 suggested changes Jun 26, 2021

View reviewed changes

mridulm reviewed Jun 28, 2021

View reviewed changes

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java Show resolved Hide resolved

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java Outdated Show resolved Hide resolved

mridulm mentioned this pull request Jun 28, 2021

[WIP][SPARK-35917][SHUFFLE][CORE]Disable push-based shuffle feature to prevent it from being used #33118

Closed

otterc mentioned this pull request Jun 29, 2021

[SPARK-32922][SHUFFLE][CORE] Adds support for executors to fetch local and remote merged shuffle data #32140

Closed

otterc suggested changes Jun 29, 2021

View reviewed changes

Ngone51 reviewed Jun 29, 2021

View reviewed changes

...network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/PushBlockStream.java Show resolved Hide resolved

Ngone51 reviewed Jun 29, 2021

View reviewed changes

core/src/main/scala/org/apache/spark/storage/DiskBlockManager.scala Outdated Show resolved Hide resolved

Ngone51 reviewed Jun 29, 2021

View reviewed changes

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java Outdated Show resolved Hide resolved

Ngone51 reviewed Jun 29, 2021

View reviewed changes

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java Outdated Show resolved Hide resolved

Ngone51 reviewed Jun 29, 2021

View reviewed changes

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java Outdated Show resolved Hide resolved

Ngone51 reviewed Jun 29, 2021

View reviewed changes

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java Outdated Show resolved Hide resolved

venkata91 suggested changes Jul 5, 2021

View reviewed changes

core/src/main/scala/org/apache/spark/internal/config/package.scala Show resolved Hide resolved

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java Outdated Show resolved Hide resolved

venkata91 suggested changes Jul 18, 2021

View reviewed changes

...network-shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/PushBlockStream.java Outdated Show resolved Hide resolved

Address comments

1de689b

Ngone51 reviewed Jul 19, 2021

View reviewed changes

.../network-shuffle/src/main/java/org/apache/spark/network/shuffle/RemoteBlockPushResolver.java Show resolved Hide resolved

Ngone51 approved these changes Jul 19, 2021

View reviewed changes

venkata91 mentioned this pull request Jul 19, 2021

[SPARK-32923][CORE][SHUFFLE] Handle indeterminate stage retries for push-based shuffle #33034

Closed

Merge remote-tracking branch 'upstream/master' into SPARK-35546

5310991

asfgit closed this in c77acf0 Jul 20, 2021

dongjoon-hyun reviewed Jul 23, 2021

View reviewed changes

[SPARK-35546][Shuffle] Enable push-based shuffle when multiple app attempts are enabled and manage concurrent access to the state in a better way #33078

[SPARK-35546][Shuffle] Enable push-based shuffle when multiple app attempts are enabled and manage concurrent access to the state in a better way #33078

Conversation

zhouyejoe commented Jun 25, 2021

What changes were proposed in this pull request?

Summary of the change:

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

zhouyejoe commented Jun 25, 2021

mridulm left a comment

Choose a reason for hiding this comment

mridulm commented Jun 25, 2021

mridulm commented Jun 25, 2021

SparkQA commented Jun 25, 2021

SparkQA commented Jun 25, 2021

SparkQA commented Jun 25, 2021

otterc left a comment

Choose a reason for hiding this comment

zhouyejoe commented Jun 27, 2021

mridulm left a comment

Choose a reason for hiding this comment

SparkQA commented Jun 29, 2021

SparkQA commented Jun 29, 2021

mridulm commented Jul 1, 2021 • edited Loading

Ngone51 commented Jul 1, 2021

Ngone51 commented Jul 2, 2021

zhouyejoe commented Jul 2, 2021

SparkQA commented Jul 5, 2021

SparkQA commented Jul 5, 2021

SparkQA commented Jul 17, 2021

SparkQA commented Jul 17, 2021

SparkQA commented Jul 17, 2021

zhouyejoe commented Jul 18, 2021

SparkQA commented Jul 18, 2021

SparkQA commented Jul 18, 2021

mridulm commented Jul 19, 2021

venkata91 commented Jul 19, 2021

otterc commented Jul 19, 2021

zhouyejoe commented Jul 19, 2021

mridulm commented Jul 19, 2021

SparkQA commented Jul 19, 2021

SparkQA commented Jul 20, 2021

SparkQA commented Jul 20, 2021

SparkQA commented Jul 20, 2021

SparkQA commented Jul 20, 2021

mridulm commented Jul 20, 2021

mridulm commented Jul 20, 2021

dongjoon-hyun Jul 23, 2021

Choose a reason for hiding this comment

mridulm commented Jul 1, 2021 •

edited

Loading