[SPARK-32919][SHUFFLE][test-maven][test-hadoop2.7] Driver side changes for coordinating push based shuffle by selecting external shuffle services for merging partitions #30164

venkata91 · 2020-10-27T22:44:22Z

What changes were proposed in this pull request?

Driver side changes for coordinating push based shuffle by selecting external shuffle services for merging partitions.

This PR includes changes related to ShuffleMapStage preparation which is selection of merger locations and initializing them as part of ShuffleDependency.

Currently this code is not used as some of the changes would come subsequently as part of https://issues.apache.org/jira/browse/SPARK-32917 (shuffle blocks push as part of ShuffleMapTask), https://issues.apache.org/jira/browse/SPARK-32918 (support for finalize API) and https://issues.apache.org/jira/browse/SPARK-32920 (finalization of push/merge phase). This is why the tests here are also partial, once these above mentioned changes are raised as PR we will have enough tests for DAGScheduler piece of code as well.

Why are the changes needed?

Added a new API in SchedulerBackend to get merger locations for push based shuffle. This is currently implemented for Yarn and other cluster managers can have separate implementations which is why a new API is introduced.

Does this PR introduce any user-facing change?

Yes, user facing config to enable push based shuffle is introduced

How was this patch tested?

Added unit tests partially and some of the changes in DAGScheduler depends on future changes, DAGScheduler tests will be added along with those changes.

Lead-authored-by: Venkata krishnan Sowrirajan [email protected]
Co-authored-by: Min Shen [email protected]

…huffle by selecting external shuffle services for merging partitions

venkata91 · 2020-10-27T22:46:58Z

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

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

tgravescs · 2020-10-28T15:46:19Z

ok to test

SparkQA · 2020-10-28T16:30:58Z

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

SparkQA · 2020-10-28T16:59:37Z

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

tgravescs · 2020-10-28T16:59:41Z

the paper talks about:
we choose Magnet shu�e ser-vices in locations beyond the active Spark executors, andlaunch Spark executors later via dynamic allocation basedon locations of the chosen Magnet shu�e services. Thisway, instead of choosing Magnet shu�e services based onSpark executor locations, we launch Spark executors basedon locations of Magnet shu�e services. This optimization ispossible because of Magnet's integration with Spark native

I know this was also talked about in the SPIP. The current implementation seems to not do this. can the description here please be updated to state what it does and why. Is this something to be PR'd later or just talked about as a possibility

Victsm · 2020-10-28T18:46:04Z

@tgravescs
What we have is the same as what's described in the paper and the SPIP doc.
For handling DRA, we are essentially doing 2 things:

Choose shuffle service locations beyond the current active Spark executors.
Launching Spark executors with DRA based on locations of the chosen shuffle services.

This PR enables the first.
By keeping track of all historical locations of executors launched for a given Spark application, we get 2 benefits.

When DRA kicks in later on, and significantly reduces the number of available active executors, we can still look into the historical locations of past executors to get sufficient shuffle service locations to perform block push/merge.
On a YARN cluster with authentication enabled, picking historical locations of past executors would ensure that the executor can talk to the shuffle service performing SASL authentication, and upon application finishing up the local dirs storing the merged shuffle files get cleaned up.

In a follow up patch for driver side change (MapOutputTracker#getPreferredLocationsForShuffle), the second is enabled.
Preferred location for shuffle now takes into consideration of shuffle service locations for a given shuffle.
This would set the preferred locations for the corresponding ShuffleRDD as well, which would then have 2 impacts.

When TaskSetManager schedules tasks to executors, this would impact the task placement strategy.
When ExecutorAllocationManager requests more executors for DRA, this preferred location would be passed to YARN to request containers with the preferred locality.

SparkQA · 2020-10-28T18:46:43Z

Test build #130376 has finished for PR 30164 at commit 2688df2.

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

tgravescs · 2020-10-29T17:34:43Z

thanks for the explanation

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

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

core/src/main/scala/org/apache/spark/scheduler/SchedulerBackend.scala

...e-managers/yarn/src/main/scala/org/apache/spark/scheduler/cluster/YarnSchedulerBackend.scala

mridulm · 2020-10-30T06:43:10Z

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

+   * at the same time. Will improve this in a later version.
+   */
+  def isPushBasedShuffleEnabled(conf: SparkConf): Boolean = {
+    conf.get(PUSH_BASED_SHUFFLE_ENABLED) && conf.get(SHUFFLE_SERVICE_ENABLED)


I look forward to the day when the second condition will be disabled :-)
It will be relevant for both k8s and spark streaming !

+CC @dongjoon-hyun you might be interested in this in future.

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

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

core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala

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

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

SparkQA · 2020-11-02T21:09:32Z

Test build #130533 has finished for PR 30164 at commit 0423970.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class GetShufflePushMergerLocations(numMergersNeeded: Int, hostsToFilter: Set[String])

SparkQA · 2020-11-02T21:47:52Z

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

SparkQA · 2020-11-02T22:12:38Z

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

SparkQA · 2020-11-03T19:45:19Z

Test build #130575 has finished for PR 30164 at commit 3a6219f.

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

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

SparkQA · 2020-11-03T20:27:07Z

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

SparkQA · 2020-11-03T20:50:02Z

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

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

SparkQA · 2020-11-05T02:01:15Z

Test build #130628 has finished for PR 30164 at commit ca44d03.

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

SparkQA · 2020-11-18T23:09:23Z

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

SparkQA · 2020-11-18T23:17:55Z

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

SparkQA · 2020-11-19T01:58:40Z

Test build #131302 has finished for PR 30164 at commit 050a5ae.

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

SparkQA · 2020-11-19T02:02:16Z

Test build #131298 has finished for PR 30164 at commit 050a5ae.

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

SparkQA · 2020-11-19T02:12:42Z

Test build #131303 has finished for PR 30164 at commit 1714829.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
class ExecutorSource(

SparkQA · 2020-11-19T02:15:38Z

Test build #131297 has finished for PR 30164 at commit 2849051.

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

mridulm · 2020-11-19T19:21:21Z

The scala-2.13 failure is unrelated to this PR, and is getting fixed.
Given all review comments have been addressed, I will merge this once there is a clean build.
@venkata91 If there are pending commit's, please let me know - I saw some recent changes made after reviews.

venkata91 · 2020-11-19T19:25:11Z

The scala-2.13 failure is unrelated to this PR, and is getting fixed.
Given all review comments have been addressed, I will merge this once there is a clean build.
@venkata91 If there are pending commit's, please let me know - I saw some recent changes made after reviews.

No thats it, only addressing @dongjoon-hyun comments and merging the upstream master as it had some merge conflicts.

SparkQA · 2020-11-19T19:30:55Z

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

SparkQA · 2020-11-19T19:54:36Z

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

SparkQA · 2020-11-19T22:27:03Z

Test build #131367 has finished for PR 30164 at commit 1ba7668.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
abstract class LikeAllBase extends UnaryExpression with ImplicitCastInputTypes with NullIntolerant
case class LikeAll(child: Expression, patterns: Seq[UTF8String]) extends LikeAllBase
case class NotLikeAll(child: Expression, patterns: Seq[UTF8String]) extends LikeAllBase
implicit class MetadataColumnsHelper(metadata: Array[MetadataColumn])

mridulm · 2020-11-20T00:31:10Z

Can you update the PR @venkata91 ? The failures are unrelated, but want to make sure - the upstream changes should help fix the issues. Thx !

venkata91 · 2020-11-20T00:33:58Z

Can you update the PR @venkata91 ? The failures are unrelated, but want to make sure - the upstream changes should help fix the issues. Thx !

Done. Thanks.

mridulm · 2020-11-20T00:36:40Z

ok to test

SparkQA · 2020-11-20T01:15:40Z

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

SparkQA · 2020-11-20T01:26:35Z

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

SparkQA · 2020-11-20T01:51:29Z

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

SparkQA · 2020-11-20T01:51:37Z

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

mridulm · 2020-11-20T02:04:45Z

+CC @dongjoon-hyun Looks like the test failures in jenkins are from executor decomissioning. The github actions went through fine.
Is this known to be flakey ? Thx

SparkQA · 2020-11-20T04:39:11Z

Test build #131378 has finished for PR 30164 at commit 5ce2934.

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

dongjoon-hyun · 2020-11-20T04:57:41Z

I agree with you. @mridulm . Please proceed to merge.

dongjoon-hyun · 2020-11-20T04:59:40Z

Due to the K8s IT flakiness of - Test basic decommissioning *** FAILED ***, I have been monitoring it.

https://amplab.cs.berkeley.edu/jenkins/view/Spark%20K8s%20Builds/job/spark-master-test-k8s/

mridulm · 2020-11-20T12:48:21Z

Merging to master, thanks for working on this @venkata91 !
Thanks for the reviews, @tgravescs, @Ngone51, @attilapiros, @Victsm, @otterc, @dongjoon-hyun

dongjoon-hyun · 2020-11-20T18:49:14Z

Thank you all!

FYI, K8s IT is happy on master branch with this patch.

https://amplab.cs.berkeley.edu/jenkins/view/Spark%20K8s%20Builds/job/spark-master-test-k8s/683/

venkata91 · 2020-11-20T19:00:53Z

Merging to master, thanks for working on this @venkata91 !
Thanks for the reviews, @tgravescs, @Ngone51, @attilapiros, @Victsm, @otterc, @dongjoon-hyun

Thanks Mridul for shepherding this all along :)

…s for coordinating push based shuffle by selecting external shuffle services for merging partitions Driver side changes for coordinating push based shuffle by selecting external shuffle services for merging partitions. This PR includes changes related to `ShuffleMapStage` preparation which is selection of merger locations and initializing them as part of `ShuffleDependency`. Currently this code is not used as some of the changes would come subsequently as part of https://issues.apache.org/jira/browse/SPARK-32917 (shuffle blocks push as part of `ShuffleMapTask`), https://issues.apache.org/jira/browse/SPARK-32918 (support for finalize API) and https://issues.apache.org/jira/browse/SPARK-32920 (finalization of push/merge phase). This is why the tests here are also partial, once these above mentioned changes are raised as PR we will have enough tests for DAGScheduler piece of code as well. Added a new API in `SchedulerBackend` to get merger locations for push based shuffle. This is currently implemented for Yarn and other cluster managers can have separate implementations which is why a new API is introduced. Yes, user facing config to enable push based shuffle is introduced Added unit tests partially and some of the changes in DAGScheduler depends on future changes, DAGScheduler tests will be added along with those changes. Lead-authored-by: Venkata krishnan Sowrirajan vsowrirajanlinkedin.com Co-authored-by: Min Shen mshenlinkedin.com Closes #30164 from venkata91/upstream-SPARK-32919. Lead-authored-by: Venkata krishnan Sowrirajan <[email protected]> Co-authored-by: Min Shen <[email protected]> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

[SPARK-32919][CORE] Driver side changes for coordinating push based s…

68fc364

…huffle by selecting external shuffle services for merging partitions

venkata91 changed the title ~~[SPARK-32919][CORE] Driver side changes for coordinating push based s…~~ [SPARK-32919][SHUFFLE] Driver side changes for coordinating push based s… Oct 27, 2020

venkata91 changed the title ~~[SPARK-32919][SHUFFLE] Driver side changes for coordinating push based s…~~ [SPARK-32919][SHUFFLE] Driver side changes for coordinating push based shuffle by selecting external shuffle services for merging partitions Oct 27, 2020

Empty commit to add Min Shen to authors list

2688df2

Victsm reviewed Oct 28, 2020

View reviewed changes

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

mridulm reviewed Oct 30, 2020

View reviewed changes

tgravescs reviewed Oct 30, 2020

View reviewed changes

Addressed some of the comments both from Mridul and Thomas

0423970

venkata91 added 3 commits November 2, 2020 17:09

Address style check issues

eb93fe1

Address style checks and Mridul's comment

f2f61e2

Address Tom's comment

3a6219f

otterc reviewed Nov 3, 2020

View reviewed changes

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

venkata91 mentioned this pull request Nov 3, 2020

[SPARK-32916][SHUFFLE] Implementation of shuffle service that leverages push-based shuffle in YARN deployment mode #30062

Closed

mridulm reviewed Nov 4, 2020

View reviewed changes

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

Prefer active executors for merger locations

ca44d03

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

1ba7668

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

5ce2934

asfgit closed this in 8218b48 Nov 20, 2020

Victsm mentioned this pull request Dec 2, 2020

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

Closed

[SPARK-32919][SHUFFLE][test-maven][test-hadoop2.7] Driver side changes for coordinating push based shuffle by selecting external shuffle services for merging partitions #30164

[SPARK-32919][SHUFFLE][test-maven][test-hadoop2.7] Driver side changes for coordinating push based shuffle by selecting external shuffle services for merging partitions #30164

Uh oh!

Conversation

venkata91 commented Oct 27, 2020 • 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!

venkata91 commented Oct 27, 2020

Uh oh!

Uh oh!

tgravescs commented Oct 28, 2020

Uh oh!

SparkQA commented Oct 28, 2020

Uh oh!

SparkQA commented Oct 28, 2020

Uh oh!

tgravescs commented Oct 28, 2020

Uh oh!

Victsm commented Oct 28, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

SparkQA commented Oct 28, 2020

Uh oh!

tgravescs commented Oct 29, 2020

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

mridulm Oct 30, 2020

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

SparkQA commented Nov 2, 2020

Uh oh!

SparkQA commented Nov 2, 2020

Uh oh!

SparkQA commented Nov 2, 2020

Uh oh!

SparkQA commented Nov 3, 2020

Uh oh!

Uh oh!

SparkQA commented Nov 3, 2020

Uh oh!

SparkQA commented Nov 3, 2020

Uh oh!

Uh oh!

SparkQA commented Nov 5, 2020

Uh oh!

SparkQA commented Nov 18, 2020

Uh oh!

SparkQA commented Nov 18, 2020

Uh oh!

SparkQA commented Nov 19, 2020

Uh oh!

SparkQA commented Nov 19, 2020

Uh oh!

SparkQA commented Nov 19, 2020

Uh oh!

SparkQA commented Nov 19, 2020

Uh oh!

mridulm commented Nov 19, 2020

Uh oh!

venkata91 commented Oct 27, 2020 •

edited

Loading

Victsm commented Oct 28, 2020 •

edited

Loading

mridulm commented Nov 20, 2020 •

edited

Loading