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

otterc · 2021-04-13T03:46:25Z

What changes were proposed in this pull request?

This is the shuffle fetch side change where executors can fetch local/remote push-merged shuffle data from shuffle services. This is needed for push-based shuffle - SPIP SPARK-30602.
The change adds support to the ShuffleBlockFetchIterator to fetch push-merged block meta and shuffle chunks from local and remote ESS. If the fetch of any of these fails, then the iterator fallsback to fetch the original shuffle blocks that belonged to the push-merged block.

Why are the changes needed?

These changes are needed for push-based shuffle. Refer to the SPIP in SPARK-30602.

Does this PR introduce any user-facing change?

When push-based shuffle is turned on then that will fetch push-merged blocks from the remote shuffle service. The client logs will indicate this.

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.

Lead-authored-by: Chandni Singh [email protected]
Co-authored-by: Min Shen [email protected]
Co-authored-by: Ye Zhou [email protected]

otterc · 2021-04-13T03:47:22Z

Adding @Victsm @mridulm @zhouyejoe

otterc · 2021-05-05T21:41:04Z

Will resolve conflicts when this PR is merged as.

otterc · 2021-05-27T00:19:17Z

I have rebased the changes against the latest master. It only depends on #32140 but it interfaces at just 1 place with that PR so it can still be reviewed in parallel.
Given the size of this change and the feature freeze in early July, would really appreciate if folks can help with the review.
@mridulm @Victsm @Ngone51 @tgravescs @attilapiros

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

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

otterc · 2021-05-27T00:46:09Z

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

Note to self: Most of this is as before. Have added conditions for shuffleChunks

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

otterc · 2021-05-27T06:06:30Z

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

Note to reviewers: Both of these are created locally in the initialize and fetchFallbackBlocks. With push-based shuffle, partitionBlocksByFetchMode can be called multiple times because failure to fetch merged shuffle blocks/chunks (fallback) finds original blocks that made up the failed merged block/chunk and then we need to create new FetchRequests from these.

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

otterc · 2021-06-03T00:48:10Z

Gentle ping to help review this PR @tgravescs @attilapiros @Ngone51 @mridulm @Victsm @zhouyejoe

mridulm

Took an initial pass, yet to look at ShuffleBlockFetcherIterator or test suites.
I am wondering, given the volume, whether we want to split between ESS side and client side. Thoughts ?

common/network-common/src/main/java/org/apache/spark/network/client/BaseResponseCallback.java

...n/network-common/src/main/java/org/apache/spark/network/client/TransportResponseHandler.java

...n/network-common/src/main/java/org/apache/spark/network/protocol/MergedBlockMetaRequest.java

mridulm · 2021-06-04T04:47:59Z

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java

getNumBlocks makes this code cleaner.

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalBlockHandler.java

...on/network-shuffle/src/main/java/org/apache/spark/network/shuffle/OneForOneBlockFetcher.java

...shuffle/src/main/java/org/apache/spark/network/shuffle/protocol/FetchShuffleBlockChunks.java

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

otterc · 2021-06-04T15:01:49Z

Took an initial pass, yet to look at ShuffleBlockFetcherIterator or test suites.
I am wondering, given the volume, whether we want to split between ESS side and client side. Thoughts ?

Thanks Mridul for reviewing!
My thoughts about splitting this change is that this PR completely encapsulates the fetch-side changes so it is easier to understand how the new messages introduced on the client side are being handled on the server-side. One of the feedbacks we got last year was that we broke things up in a way that made it difficult to understand.
On the server side, this PR mostly adds the wiring needed in ExternalBlockHandler to server the merged meta/data requests from RemoteBlockPushResolver.

That being said, I am still okay to break this change into client/sever PRs if that makes the review easier for the reviewers.
cc. @mridulm @Ngone51 @Victsm @tgravescs @attilapiros

…erIterator.scala Co-authored-by: Mridul Muralidharan <[email protected]>

otterc · 2021-06-28T23:01:20Z

@mridulm Resolved the conflict.

mridulm

Just had a minor nit - can you fix it pls ?
Will merge it once you are done.

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

Ngone51 · 2021-06-29T04:59:34Z

Sorry for the delay. I'll do a review today. BTW, are there any other necessary magnet PRs that have to be merged for the 3.2 cut/release?

otterc · 2021-06-29T05:59:19Z

Sorry for the delay. I'll do a review today. BTW, are there any other necessary magnet PRs that have to be merged for the 3.2 cut/release?

There are 2 pending tasks which are necessary for Magnet:

SPARK-35546: #33078. This one is ready for review
SPARK-32923: #33034. This is WIP

otterc · 2021-06-29T07:01:04Z

The test failures are unrelated.

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

Ngone51 · 2021-06-29T08:02:59Z

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

+      s"(${Utils.bytesToString(pushMergedLocalBlockBytes)}) " +
+      s"local push-merged and $numRemoteBlocks (${Utils.bytesToString(remoteBlockBytes)}) " +
+      s"remote blocks")
+    this.hostLocalBlocks ++= hostLocalBlocksCurrentIteration


Shall we reuse hostLocalBlocksByExecutor here? e.g.,

this.hostLocalBlocks ++= hostLocalBlocksByExecutor.values .flatMap { infos => infos.map(info => (info._1, info._3)) }

so we can get rid of hostLocalBlocksCurrentIteration.

Would need to do something for finding out the number of hostLocalBlocks in the assertions before as well.

I have made this change but also added a var for counting num of hostLocalBlocks which is needed for the assertions. PTAL

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

Ngone51 · 2021-06-29T14:42:52Z

I left some minor comments. I think we're ready to merge after addressing these comments.

SparkQA · 2021-06-29T18:39:39Z

Test build #140388 has started for PR 32140 at commit ad89a02.

mridulm · 2021-06-29T22:45:09Z

Thanks for working on this @otterc !
Thanks for the detailed reviews @Ngone51 :-)

mridulm · 2021-06-29T22:45:18Z

Merged to master

otterc · 2021-06-29T23:57:21Z

Thanks @mridulm and @Ngone51 for the thorough reviews!

…utor tries to fetch push-merged blocks ### What changes were proposed in this pull request? Below 2 bugs were introduced with #32140 1. Instead of requesting the local-dirs for push-merged-local blocks from the ESS, `PushBasedFetchHelper` requests it from other executors. Push-based shuffle is only enabled when the ESS is enabled so it should always fetch the dirs from the ESS and not from other executors which is not yet supported. 2. The size of the push-merged blocks is logged incorrectly. ### Why are the changes needed? This fixes the above mentioned bugs and is needed for push-based shuffle to work properly. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Tested this by running an application on the cluster. The UTs mock the call `hostLocalDirManager.getHostLocalDirs` which is why didn't catch (1) with the UT. However, the fix is trivial and checking this in the UT will require a lot more effort so I haven't modified it in the UT. Logs of the executor with the bug ``` 21/07/15 15:42:46 WARN ExternalBlockStoreClient: Error while trying to get the host local dirs for [shuffle-push-merger] 21/07/15 15:42:46 WARN PushBasedFetchHelper: Error while fetching the merged dirs for push-merged-local blocks: shuffle_0_-1_13. Fetch the original blocks instead java.lang.RuntimeException: java.lang.IllegalStateException: Invalid executor id: shuffle-push-merger, expected 92. at org.apache.spark.network.netty.NettyBlockRpcServer.receive(NettyBlockRpcServer.scala:130) at org.apache.spark.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:163) ``` After the fix, the executors were able to fetch the local push-merged blocks. Closes #33378 from otterc/SPARK-32922-followup. Authored-by: Chandni Singh <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

…utor tries to fetch push-merged blocks ### What changes were proposed in this pull request? Below 2 bugs were introduced with #32140 1. Instead of requesting the local-dirs for push-merged-local blocks from the ESS, `PushBasedFetchHelper` requests it from other executors. Push-based shuffle is only enabled when the ESS is enabled so it should always fetch the dirs from the ESS and not from other executors which is not yet supported. 2. The size of the push-merged blocks is logged incorrectly. ### Why are the changes needed? This fixes the above mentioned bugs and is needed for push-based shuffle to work properly. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Tested this by running an application on the cluster. The UTs mock the call `hostLocalDirManager.getHostLocalDirs` which is why didn't catch (1) with the UT. However, the fix is trivial and checking this in the UT will require a lot more effort so I haven't modified it in the UT. Logs of the executor with the bug ``` 21/07/15 15:42:46 WARN ExternalBlockStoreClient: Error while trying to get the host local dirs for [shuffle-push-merger] 21/07/15 15:42:46 WARN PushBasedFetchHelper: Error while fetching the merged dirs for push-merged-local blocks: shuffle_0_-1_13. Fetch the original blocks instead java.lang.RuntimeException: java.lang.IllegalStateException: Invalid executor id: shuffle-push-merger, expected 92. at org.apache.spark.network.netty.NettyBlockRpcServer.receive(NettyBlockRpcServer.scala:130) at org.apache.spark.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:163) ``` After the fix, the executors were able to fetch the local push-merged blocks. Closes #33378 from otterc/SPARK-32922-followup. Authored-by: Chandni Singh <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com> (cherry picked from commit 6d2cbad) Signed-off-by: Mridul Muralidharan <mridulatgmail.com>

…l and remote merged shuffle data ### What changes were proposed in this pull request? This is the shuffle fetch side change where executors can fetch local/remote push-merged shuffle data from shuffle services. This is needed for push-based shuffle - SPIP [SPARK-30602](https://issues.apache.org/jira/browse/SPARK-30602). The change adds support to the `ShuffleBlockFetchIterator` to fetch push-merged block meta and shuffle chunks from local and remote ESS. If the fetch of any of these fails, then the iterator fallsback to fetch the original shuffle blocks that belonged to the push-merged block. ### Why are the changes needed? These changes are needed for push-based shuffle. Refer to the SPIP in [SPARK-30602](https://issues.apache.org/jira/browse/SPARK-30602). ### Does this PR introduce _any_ user-facing change? When push-based shuffle is turned on then that will fetch push-merged blocks from the remote shuffle service. The client logs will indicate this. ### 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](https://issues.apache.org/jira/browse/SPARK-30602). We have already verified the functionality and the improved performance as documented in the SPIP doc. Lead-authored-by: Chandni Singh chsinghlinkedin.com Co-authored-by: Min Shen mshenlinkedin.com Co-authored-by: Ye Zhou yezhoulinkedin.com Closes apache#32140 from otterc/SPARK-32922. Lead-authored-by: Chandni Singh <[email protected]> Co-authored-by: Chandni Singh <[email protected]> Co-authored-by: Min Shen <[email protected]> Co-authored-by: otterc <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

…l and remote merged shuffle data This is the shuffle fetch side change where executors can fetch local/remote push-merged shuffle data from shuffle services. This is needed for push-based shuffle - SPIP [SPARK-30602](https://issues.apache.org/jira/browse/SPARK-30602). The change adds support to the `ShuffleBlockFetchIterator` to fetch push-merged block meta and shuffle chunks from local and remote ESS. If the fetch of any of these fails, then the iterator fallsback to fetch the original shuffle blocks that belonged to the push-merged block. These changes are needed for push-based shuffle. Refer to the SPIP in [SPARK-30602](https://issues.apache.org/jira/browse/SPARK-30602). When push-based shuffle is turned on then that will fetch push-merged blocks from the remote shuffle service. The client logs will indicate this. Added unit tests. The reference PR with the consolidated changes covering the complete implementation is also provided in [SPARK-30602](https://issues.apache.org/jira/browse/SPARK-30602). We have already verified the functionality and the improved performance as documented in the SPIP doc. Lead-authored-by: Chandni Singh chsinghlinkedin.com Co-authored-by: Min Shen mshenlinkedin.com Co-authored-by: Ye Zhou yezhoulinkedin.com Closes #32140 from otterc/SPARK-32922. Lead-authored-by: Chandni Singh <[email protected]> Co-authored-by: Chandni Singh <[email protected]> Co-authored-by: Min Shen <[email protected]> Co-authored-by: otterc <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

…utor tries to fetch push-merged blocks ### What changes were proposed in this pull request? Below 2 bugs were introduced with #32140 1. Instead of requesting the local-dirs for push-merged-local blocks from the ESS, `PushBasedFetchHelper` requests it from other executors. Push-based shuffle is only enabled when the ESS is enabled so it should always fetch the dirs from the ESS and not from other executors which is not yet supported. 2. The size of the push-merged blocks is logged incorrectly. ### Why are the changes needed? This fixes the above mentioned bugs and is needed for push-based shuffle to work properly. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Tested this by running an application on the cluster. The UTs mock the call `hostLocalDirManager.getHostLocalDirs` which is why didn't catch (1) with the UT. However, the fix is trivial and checking this in the UT will require a lot more effort so I haven't modified it in the UT. Logs of the executor with the bug ``` 21/07/15 15:42:46 WARN ExternalBlockStoreClient: Error while trying to get the host local dirs for [shuffle-push-merger] 21/07/15 15:42:46 WARN PushBasedFetchHelper: Error while fetching the merged dirs for push-merged-local blocks: shuffle_0_-1_13. Fetch the original blocks instead java.lang.RuntimeException: java.lang.IllegalStateException: Invalid executor id: shuffle-push-merger, expected 92. at org.apache.spark.network.netty.NettyBlockRpcServer.receive(NettyBlockRpcServer.scala:130) at org.apache.spark.network.server.TransportRequestHandler.processRpcRequest(TransportRequestHandler.java:163) ``` After the fix, the executors were able to fetch the local push-merged blocks. Closes #33378 from otterc/SPARK-32922-followup. Authored-by: Chandni Singh <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

otterc force-pushed the SPARK-32922 branch from 6082ca6 to 331560c Compare April 13, 2021 04:07

github-actions bot added CORE YARN labels Apr 13, 2021

otterc mentioned this pull request Apr 16, 2021

[SPARK-32921][SHUFFLE] MapOutputTracker extensions to support push-based shuffle #30480

Closed

otterc mentioned this pull request May 11, 2021

[SPARK-35263] [TEST] Refactor ShuffleBlockFetcherIteratorSuite to reduce duplicated code #32389

Closed

otterc force-pushed the SPARK-32922 branch from ddb2946 to 55105c6 Compare May 26, 2021 07:39

otterc commented May 27, 2021

View reviewed changes

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

otterc commented May 27, 2021

View reviewed changes

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

otterc mentioned this pull request May 27, 2021

[SPARK-33350][SHUFFLE] Add support to DiskBlockManager to create merge directory and to get the local shuffle merged data #32007

Closed

github-actions bot added BUILD DOCS INFRA KUBERNETES ML MLLIB PYTHON R SQL STRUCTURED STREAMING WEB UI labels Jun 3, 2021

otterc force-pushed the SPARK-32922 branch from 94f8e72 to 11ac0fb Compare June 3, 2021 00:13

mridulm reviewed Jun 4, 2021

View reviewed changes

otterc force-pushed the SPARK-32922 branch from 697c255 to 051acc1 Compare June 4, 2021 23:30

otterc and others added 6 commits June 28, 2021 15:31

Fixed the line length

94d7c5e

Addressed yi.wu's comments

67bd821

Update core/src/main/scala/org/apache/spark/storage/ShuffleBlockFetch…

fc1b9f1

…erIterator.scala Co-authored-by: Mridul Muralidharan <[email protected]>

Addressed Mridul's comments

390300a

Removed passing numChunks

d53293e

Rebased against apache master

69fcc23

otterc force-pushed the SPARK-32922 branch from 4a967bd to 69fcc23 Compare June 28, 2021 22:42

mridulm approved these changes Jun 29, 2021

View reviewed changes

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

Addressed Mridul's comments

c10b943

Ngone51 reviewed Jun 29, 2021

View reviewed changes

otterc force-pushed the SPARK-32922 branch from 7de8bd3 to 726023b Compare June 29, 2021 15:58

Addressed review comments

ad89a02

otterc force-pushed the SPARK-32922 branch from 726023b to ad89a02 Compare June 29, 2021 16:19

asfgit closed this in 9a5cd15 Jun 29, 2021

otterc mentioned this pull request Jul 15, 2021

[SPARK-32922][SHUFFLE][CORE][FOLLOWUP] Fixes few issues when the executor tries to fetch push-merged blocks #33378

Closed

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

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

Uh oh!

Conversation

otterc commented Apr 13, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

Uh oh!

otterc commented Apr 13, 2021

Uh oh!

otterc commented May 5, 2021

Uh oh!

otterc commented May 27, 2021

Uh oh!

Uh oh!

Uh oh!

otterc May 27, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

otterc May 27, 2021

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

otterc commented Jun 3, 2021

Uh oh!

mridulm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mridulm Jun 4, 2021

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

otterc commented Jun 4, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

otterc commented Jun 28, 2021

Uh oh!

mridulm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Ngone51 commented Jun 29, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

otterc commented Jun 29, 2021

Uh oh!

otterc commented Jun 29, 2021

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Ngone51 Jun 29, 2021

Choose a reason for hiding this comment

Uh oh!

otterc Jun 29, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

otterc Jun 29, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

otterc commented Apr 13, 2021 •

edited

Loading

otterc May 27, 2021 •

edited

Loading

otterc commented Jun 4, 2021 •

edited

Loading

Ngone51 commented Jun 29, 2021 •

edited

Loading

otterc Jun 29, 2021 •

edited

Loading

otterc Jun 29, 2021 •

edited

Loading