[SPARK-32917][SHUFFLE][CORE] Adds support for executors to push shuffle blocks after successful map task completion #30312

otterc · 2020-11-10T07:37:54Z

What changes were proposed in this pull request?

This is the shuffle writer side change where executors can push data to remote shuffle services. This is needed for push-based shuffle - SPIP SPARK-30602.
Summary of changes:

This adds support for executors to push shuffle blocks after map tasks complete writing shuffle data.
This also introduces a timeout specifically for creating connection to remote shuffle services.

Why are the changes needed?

These changes are needed for push-based shuffle. Refer to the SPIP in SPARK-30602.
The main reason to create a separate connection creation timeout is because the existing connectionTimeoutMs is overloaded and is used for connection creation timeouts as well as connection idle timeout. The connection creation timeout should be much lower than the idle timeouts. The default for connectionTimeoutMs is 120s. This is quite high for just establishing the connections. If a shuffle server node is bad then the connection creation will fail within few seconds. However, an overloaded shuffle server may take much longer to respond to a request and the channel can stay idle for a much longer time which is expected. Another reason is that with push-based shuffle, an executor may be fetching shuffle data and pushing shuffle data (next stage) simultaneously. Both these tasks will share the same connections with the shuffle service. If there is a bad shuffle server node and the connection creation timeout is very high then both these tasks end up waiting a long time time eventually impacting the performance.

Does this PR introduce any user-facing change?

Yes. This PR introduces client-side configs for push-based shuffle. If push-based shuffle is turned-off then the users will not see any change.

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: Min Shen [email protected]
Co-authored-by: Chandni Singh [email protected]
Co-authored-by: Ye Zhou [email protected]

core/src/main/scala/org/apache/spark/util/Utils.scala

otterc · 2020-11-10T19:20:56Z

@Victsm @mridulm @tgravescs @jiangxb1987 @attilapiros @Ngone51 Please take a look.

dongjoon-hyun · 2020-11-12T02:44:05Z

ok to test

SparkQA · 2020-11-12T03:27:13Z

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

SparkQA · 2020-11-12T03:56:25Z

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

SparkQA · 2020-11-12T08:05:03Z

Test build #130957 has finished for PR 30312 at commit c658423.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

Ngone51 · 2020-11-12T13:59:30Z

retest this please

SparkQA · 2020-11-12T15:23:14Z

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

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

core/src/main/scala/org/apache/spark/shuffle/ShuffleWriter.scala

SparkQA · 2020-11-12T15:49:21Z

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

core/src/main/scala/org/apache/spark/shuffle/ShuffleWriter.scala

SparkQA · 2020-11-12T17:29:42Z

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

SparkQA · 2020-11-12T17:58:46Z

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

SparkQA · 2020-11-12T18:03:29Z

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

SparkQA · 2020-11-12T18:27:39Z

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

SparkQA · 2020-11-12T22:07:53Z

Test build #131004 has finished for PR 30312 at commit c658423.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-11-13T00:04:04Z

Test build #131010 has finished for PR 30312 at commit 9029993.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-11-13T00:39:12Z

Test build #131013 has finished for PR 30312 at commit 1c43ac0.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class ShufflePushBlockId(shuffleId: Int, mapIndex: Int, reduceId: Int) extends BlockId

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

core/src/main/scala/org/apache/spark/shuffle/ShuffleWriter.scala

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

core/src/main/scala/org/apache/spark/shuffle/ShuffleWriter.scala

core/src/main/scala/org/apache/spark/shuffle/ShuffleBlockPusher.scala

…r.scala Co-authored-by: Min Shen <[email protected]>

SparkQA · 2020-12-19T02:25:43Z

Test build #133044 has finished for PR 30312 at commit 6aae02a.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2020-12-19T05:07:02Z

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

SparkQA · 2020-12-19T05:36:31Z

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

SparkQA · 2020-12-19T06:45:08Z

Test build #133049 has finished for PR 30312 at commit 21ea881.

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

otterc · 2021-01-04T20:35:21Z

Have addressed all the comments so far. The failing test seems unrelated.
cc @Ngone51 @Victsm @mridulm

mridulm · 2021-01-07T16:34:32Z

ok to test

mridulm · 2021-01-07T16:35:47Z

Given no other comments, will merge once tests pass (have retriggered it).

SparkQA · 2021-01-07T17:21:40Z

Test build #133803 has started for PR 30312 at commit 21ea881.

SparkQA · 2021-01-07T17:57:28Z

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

SparkQA · 2021-01-07T18:23:14Z

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

shaneknapp · 2021-01-07T18:51:06Z

test this please

SparkQA · 2021-01-07T19:56:00Z

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

SparkQA · 2021-01-07T20:24:16Z

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

SparkQA · 2021-01-07T21:41:22Z

Test build #133807 has finished for PR 30312 at commit 21ea881.

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

mridulm · 2021-01-08T18:24:19Z

Merged to master.
Thanks for working on this @otterc !
And thanks for all the reviews and comments @Ngone51, @dongjoon-hyun, @Victsm

otterc · 2021-01-08T21:18:01Z

Thanks @mridulm for merging and also reviewing. Thanks @Ngone51, @dongjoon-hyun, and @Victsm for the reviews as well.

dongjoon-hyun · 2021-01-08T21:50:09Z

Thank you, @otterc and @mridulm and all!

…ush.based.enabled in ShuffleBlockPusherSuite ### What changes were proposed in this pull request? It is a trivial change to remove the reference to an incorrect configuration for push-based shuffle from a test suite. Ref: #30312 With SPARK-32917, `ShuffleBlockPusher` and its test suite was introduced. `ShuffleBlockPusher` is created only when push-based shuffle is enabled and the tests in `ShuffleBlockPusherSuite` are just testing the functionality in the pusher. So there is no need to have these configs enabled in these test. ### Why are the changes needed? This change removes an incorrect configuration from the test suite. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? This change just removes an incorrect configuration from the test suite so haven't added any UTs for it. Closes #32992 from otterc/SPARK-35836. Authored-by: Chandni Singh <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

…le blocks after successful map task completion ### What changes were proposed in this pull request? This is the shuffle writer side change where executors can push data to remote shuffle services. This is needed for push-based shuffle - SPIP [SPARK-30602](https://issues.apache.org/jira/browse/SPARK-30602). Summary of changes: - This adds support for executors to push shuffle blocks after map tasks complete writing shuffle data. - This also introduces a timeout specifically for creating connection to remote shuffle services. ### 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). - The main reason to create a separate connection creation timeout is because the existing `connectionTimeoutMs` is overloaded and is used for connection creation timeouts as well as connection idle timeout. The connection creation timeout should be much lower than the idle timeouts. The default for `connectionTimeoutMs` is 120s. This is quite high for just establishing the connections. If a shuffle server node is bad then the connection creation will fail within few seconds. However, an overloaded shuffle server may take much longer to respond to a request and the channel can stay idle for a much longer time which is expected. Another reason is that with push-based shuffle, an executor may be fetching shuffle data and pushing shuffle data (next stage) simultaneously. Both these tasks will share the same connections with the shuffle service. If there is a bad shuffle server node and the connection creation timeout is very high then both these tasks end up waiting a long time time eventually impacting the performance. ### Does this PR introduce _any_ user-facing change? Yes. This PR introduces client-side configs for push-based shuffle. If push-based shuffle is turned-off then the users will not see any change. ### 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: Min Shen mshenlinkedin.com Co-authored-by: Chandni Singh chsinghlinkedin.com Co-authored-by: Ye Zhou yezhoulinkedin.com Closes apache#30312 from otterc/SPARK-32917. Lead-authored-by: Chandni Singh <[email protected]> Co-authored-by: Chandni Singh <[email protected]> Co-authored-by: Min Shen <[email protected]> Co-authored-by: Ye Zhou <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

…ush.based.enabled in ShuffleBlockPusherSuite ### What changes were proposed in this pull request? It is a trivial change to remove the reference to an incorrect configuration for push-based shuffle from a test suite. Ref: apache#30312 With SPARK-32917, `ShuffleBlockPusher` and its test suite was introduced. `ShuffleBlockPusher` is created only when push-based shuffle is enabled and the tests in `ShuffleBlockPusherSuite` are just testing the functionality in the pusher. So there is no need to have these configs enabled in these test. ### Why are the changes needed? This change removes an incorrect configuration from the test suite. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? This change just removes an incorrect configuration from the test suite so haven't added any UTs for it. Closes apache#32992 from otterc/SPARK-35836. Authored-by: Chandni Singh <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

…le blocks after successful map task completion This is the shuffle writer side change where executors can push data to remote shuffle services. This is needed for push-based shuffle - SPIP [SPARK-30602](https://issues.apache.org/jira/browse/SPARK-30602). Summary of changes: - This adds support for executors to push shuffle blocks after map tasks complete writing shuffle data. - This also introduces a timeout specifically for creating connection to remote shuffle services. - These changes are needed for push-based shuffle. Refer to the SPIP in [SPARK-30602](https://issues.apache.org/jira/browse/SPARK-30602). - The main reason to create a separate connection creation timeout is because the existing `connectionTimeoutMs` is overloaded and is used for connection creation timeouts as well as connection idle timeout. The connection creation timeout should be much lower than the idle timeouts. The default for `connectionTimeoutMs` is 120s. This is quite high for just establishing the connections. If a shuffle server node is bad then the connection creation will fail within few seconds. However, an overloaded shuffle server may take much longer to respond to a request and the channel can stay idle for a much longer time which is expected. Another reason is that with push-based shuffle, an executor may be fetching shuffle data and pushing shuffle data (next stage) simultaneously. Both these tasks will share the same connections with the shuffle service. If there is a bad shuffle server node and the connection creation timeout is very high then both these tasks end up waiting a long time time eventually impacting the performance. Yes. This PR introduces client-side configs for push-based shuffle. If push-based shuffle is turned-off then the users will not see any change. 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: Min Shen mshenlinkedin.com Co-authored-by: Chandni Singh chsinghlinkedin.com Co-authored-by: Ye Zhou yezhoulinkedin.com Closes #30312 from otterc/SPARK-32917. Lead-authored-by: Chandni Singh <[email protected]> Co-authored-by: Chandni Singh <[email protected]> Co-authored-by: Min Shen <[email protected]> Co-authored-by: Ye Zhou <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

…ush.based.enabled in ShuffleBlockPusherSuite ### What changes were proposed in this pull request? It is a trivial change to remove the reference to an incorrect configuration for push-based shuffle from a test suite. Ref: #30312 With SPARK-32917, `ShuffleBlockPusher` and its test suite was introduced. `ShuffleBlockPusher` is created only when push-based shuffle is enabled and the tests in `ShuffleBlockPusherSuite` are just testing the functionality in the pusher. So there is no need to have these configs enabled in these test. ### Why are the changes needed? This change removes an incorrect configuration from the test suite. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? This change just removes an incorrect configuration from the test suite so haven't added any UTs for it. Closes #32992 from otterc/SPARK-35836. Authored-by: Chandni Singh <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

github-actions bot added the CORE label Nov 10, 2020

otterc force-pushed the SPARK-32917 branch from 97bded5 to 3c84e07 Compare November 10, 2020 17:42

Victsm reviewed Nov 10, 2020

View reviewed changes

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

otterc force-pushed the SPARK-32917 branch 2 times, most recently from fa5a778 to b54589b Compare November 11, 2020 23:10

Ngone51 reviewed Nov 12, 2020

View reviewed changes

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

otterc force-pushed the SPARK-32917 branch from 43cf6e9 to 1c43ac0 Compare November 12, 2020 17:16