[SPARK-33701][SHUFFLE] Adaptive shuffle merge finalization for push-based shuffle #33896

venkata91 · 2021-09-02T05:21:50Z

What changes were proposed in this pull request?

As part of SPARK-32920 implemented a simple approach to finalization for push-based shuffle. Shuffle merge finalization is the final operation happens at the end of the stage when all the tasks are completed asking all the external shuffle services to complete the shuffle merge for the stage. Once this request is completed no more shuffle pushes will be accepted. With this approach, DAGScheduler waits for a fixed time of 10s (spark.shuffle.push.finalize.timeout) to allow some time for the inflight shuffle pushes to complete, but this adds additional overhead to stages with very little shuffles.

In this PR, instead of waiting for fixed amount of time before shuffle merge finalization now this is controlled adaptively if min threshold number of map tasks shuffle push (spark.shuffle.push.minPushRatio) completed then shuffle merge finalization will be scheduled. Also additionally if the total shuffle generated is lesser than min threshold shuffle size (spark.shuffle.push.minShuffleSizeToWait) then immediately shuffle merge finalization is scheduled.

Why are the changes needed?

This is a performance improvement to the existing functionality

Does this PR introduce any user-facing change?

Yes additional user facing configs spark.shuffle.push.minPushRatio and spark.shuffle.push.minShuffleSizeToWait

How was this patch tested?

Added unit tests in DAGSchedulerSuite, ShuffleBlockPusherSuite

Lead-authored-by: Min Shen mshen@linkedin.com
Co-authored-by: Venkata krishnan Sowrirajan vsowrirajan@linkedin.com

Victsm · 2021-09-02T14:18:18Z

Please add corresponding empty commits from authors of the original internal patch and tag that information in the PR description.

SparkQA · 2021-09-16T17:59:54Z

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

SparkQA · 2021-09-16T18:08:03Z

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

SparkQA · 2021-09-16T19:34:56Z

Test build #143358 has finished for PR 33896 at commit 76c98ad.

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

SparkQA · 2021-09-17T19:39:28Z

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

SparkQA · 2021-09-17T19:48:15Z

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

SparkQA · 2021-09-17T21:00:56Z

Test build #143423 has finished for PR 33896 at commit 5b611bb.

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

venkata91 · 2021-09-20T21:17:57Z

Removed the WIP tag, this is now reviewable. cc @mridulm @Ngone51 @Victsm @otterc @zhouyejoe @rmcyang

venkata91 · 2021-09-20T21:22:26Z

Also regarding the comment in the JIRA that not to finalize when there are no blocks pushed due to the threshold. I almost implemented that change, but then I thought anyway we are having a timeout (spark.shuffle.push.results.timeout (default 10 secs)) if finalize request don't get responded by then, scheduler moves forward submitting the next stages. Do you think we really get much with avoiding finalize when there are no blocks pushed? Thoughts? cc @Ngone51 @Victsm

venkata91 · 2021-10-25T23:22:08Z

Gentle ping @mridulm @Ngone51 @Victsm

mridulm

Thanks for working on this @venkata91 !
Took first pass through the PR, yet to go over the test suite.

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

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

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

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

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

mridulm · 2021-11-01T05:41:17Z

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

+          task.cancel(false)
+          // The current task should be coming from handleShufflePushCompleted, thus the
+          // delay should be 0 and registerMergeResults should be true.
+          assert(delay == 0 && registerMergeResults)


Move this out of the if condition ?

Shouldn't this not be the case outside of the if itself ?

Yeah since we are scheduling the finalize only if there is no scheduled task when the stage completes (one that checks if totalShuffleSize < shuffleMergeWaitMinSizeThreshold. We can move this assertion outside. Will make the change.

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

SparkQA · 2021-11-08T01:08:34Z

Test build #144974 has finished for PR 33896 at commit 04a4b6c.

This patch fails Scala style tests.
This patch does not merge cleanly.
This patch adds no public classes.

SparkQA · 2021-11-08T01:50:45Z

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

SparkQA · 2021-11-08T01:58:06Z

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

SparkQA · 2021-11-08T01:59:32Z

Test build #144975 has finished for PR 33896 at commit 9eeb7bf.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2021-11-08T02:38:32Z

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

SparkQA · 2021-11-08T02:45:17Z

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

SparkQA · 2021-12-15T06:12:22Z

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

SparkQA · 2021-12-15T06:57:36Z

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

SparkQA · 2021-12-15T08:06:38Z

Test build #146216 has finished for PR 33896 at commit d7d7546.

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

SparkQA · 2021-12-16T06:14:28Z

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

SparkQA · 2021-12-16T07:00:04Z

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

SparkQA · 2021-12-16T08:22:36Z

Test build #146261 has finished for PR 33896 at commit 616739c.

This patch fails SparkR unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
(defaultdict(<class 'list'>,

SparkQA · 2021-12-17T00:18:00Z

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

SparkQA · 2021-12-17T01:02:39Z

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

SparkQA · 2021-12-17T01:40:47Z

Test build #146301 has finished for PR 33896 at commit d5ff70c.

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

mridulm

Just had a few minor comments, thanks for working on this @venkata91 - looks pretty good to me !

core/src/test/scala/org/apache/spark/scheduler/DAGSchedulerSuite.scala

mridulm · 2021-12-19T07:32:42Z

@Ngone51 I am fine with the pr (barring the minor testcase related comments).
Please feel free to merge when you are happy with the PR - I will be on vacation, so dont want to block your review/merge :-)

venkata91 · 2021-12-20T17:55:27Z

@Ngone51 Addressed @mridulm comments. Please take a look.

SparkQA · 2021-12-20T18:41:00Z

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

SparkQA · 2021-12-20T19:36:12Z

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

SparkQA · 2021-12-20T20:40:12Z

Test build #146403 has finished for PR 33896 at commit 4930b3f.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
public class RocksDB implements KVStore
public static class TypeAliases
class RocksDBIterator<T> implements KVStoreIterator<T>
class RocksDBTypeInfo
class Index
trait SwitchToDiskStoreListener
case class RegrCount(left: Expression, right: Expression)
public class LogDivertAppender extends AbstractWriterAppender<WriterManager>

venkata91 · 2021-12-23T21:45:43Z

Not sure why this test is failing org.apache.spark.sql.streaming.StreamingAggregationSuite. Any thoughts? @Ngone51 Let me try updating the master and merging it.

venkata91 · 2021-12-23T22:15:46Z

Not sure why this test is failing org.apache.spark.sql.streaming.StreamingAggregationSuite. Any thoughts? @Ngone51 Let me try updating the master and merging it.

I tried running it locally that succeeded though.

venkata91 · 2021-12-29T03:47:54Z

Ok. Updating the PR with the latest master changes fixed the issue. But some of those test failures are transient and random. Every run has different failures compared to the previous one. @Ngone51 @mridulm Gentle reminder.

mridulm

Looks good to me.
+CC @Ngone51

mridulm · 2022-01-05T07:50:06Z

Thanks for working on this @venkata91 !
Thanks for the review @Ngone51 :-)

…ush-based shuffle apache#33896

…ased shuffle As part of SPARK-32920 implemented a simple approach to finalization for push-based shuffle. Shuffle merge finalization is the final operation happens at the end of the stage when all the tasks are completed asking all the external shuffle services to complete the shuffle merge for the stage. Once this request is completed no more shuffle pushes will be accepted. With this approach, `DAGScheduler` waits for a fixed time of 10s (`spark.shuffle.push.finalize.timeout`) to allow some time for the inflight shuffle pushes to complete, but this adds additional overhead to stages with very little shuffles. In this PR, instead of waiting for fixed amount of time before shuffle merge finalization now this is controlled adaptively if min threshold number of map tasks shuffle push (`spark.shuffle.push.minPushRatio`) completed then shuffle merge finalization will be scheduled. Also additionally if the total shuffle generated is lesser than min threshold shuffle size (`spark.shuffle.push.minShuffleSizeToWait`) then immediately shuffle merge finalization is scheduled. This is a performance improvement to the existing functionality Yes additional user facing configs `spark.shuffle.push.minPushRatio` and `spark.shuffle.push.minShuffleSizeToWait` Added unit tests in `DAGSchedulerSuite`, `ShuffleBlockPusherSuite` Lead-authored-by: Min Shen <mshenlinkedin.com> Co-authored-by: Venkata krishnan Sowrirajan <vsowrirajanlinkedin.com> Closes #33896 from venkata91/SPARK-33701. Lead-authored-by: Venkata krishnan Sowrirajan <vsowrirajan@linkedin.com> Co-authored-by: Min Shen <mshen@linkedin.com> Signed-off-by: Mridul Muralidharan <mridul<at>gmail.com>

SPARK-33701: Adaptive shuffle merge finalization

2efa1d7

venkata91 changed the title ~~[WIP] [SPARK-33701][SHUFFLE] Adaptive shuffle merge finalization for push-based shuffle~~ [WIP][SPARK-33701][SHUFFLE] Adaptive shuffle merge finalization for push-based shuffle Sep 2, 2021

github-actions bot added the CORE label Sep 2, 2021

venkata91 and others added 2 commits September 2, 2021 11:24

remove the commented block

46d9206

Add Min Shen as lead author

0bd5f4a

venkata91 force-pushed the SPARK-33701 branch from 10b7896 to 0bd5f4a Compare September 2, 2021 18:25

Address test failures in ShuffleBlockPusherSuite

76c98ad

Address build failure

5b611bb

venkata91 changed the title ~~[WIP][SPARK-33701][SHUFFLE] Adaptive shuffle merge finalization for push-based shuffle~~ [SPARK-33701][SHUFFLE] Adaptive shuffle merge finalization for push-based shuffle Sep 20, 2021

mridulm reviewed Nov 1, 2021

View reviewed changes

Address review comments from mridulm

04a4b6c

venkata91 added 2 commits November 7, 2021 17:14

Merge branch 'master' into SPARK-33701

6550aad

Update configuration.md with new configs

9eeb7bf

github-actions bot added the DOCS label Nov 8, 2021

Merge branch 'master' into SPARK-33701

616739c

Merge branch 'master' into SPARK-33701

d5ff70c

mridulm reviewed Dec 19, 2021

View reviewed changes

venkata91 added 2 commits December 20, 2021 09:26

Address mridulm review comments

3b8444c

Merge branch 'master' into SPARK-33701

4930b3f

venkata91 added 2 commits December 26, 2021 11:09

Merge branch 'master' into SPARK-33701

c38726f

Merge branch 'master' into SPARK-33701

ef1bf4b

mridulm approved these changes Dec 31, 2021

View reviewed changes

Merge branch 'master' into SPARK-33701

48e06d4

asfgit closed this in f6128a6 Jan 5, 2022

domybest11 pushed a commit to domybest11/spark that referenced this pull request Jun 15, 2022

[WIP][SPARK-33701][SHUFFLE] Adaptive shuffle merge finalization for p…

60dd20c

…ush-based shuffle apache#33896

[SPARK-33701][SHUFFLE] Adaptive shuffle merge finalization for push-based shuffle #33896

[SPARK-33701][SHUFFLE] Adaptive shuffle merge finalization for push-based shuffle #33896

Conversation

venkata91 commented Sep 2, 2021 • edited Loading

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?

Victsm commented Sep 2, 2021

SparkQA commented Sep 16, 2021

SparkQA commented Sep 16, 2021

SparkQA commented Sep 16, 2021

SparkQA commented Sep 17, 2021

SparkQA commented Sep 17, 2021

SparkQA commented Sep 17, 2021

venkata91 commented Sep 20, 2021

venkata91 commented Sep 20, 2021 • edited Loading

venkata91 commented Oct 25, 2021

mridulm left a comment

Choose a reason for hiding this comment

mridulm Nov 1, 2021

Choose a reason for hiding this comment

mridulm Nov 14, 2021

Choose a reason for hiding this comment

venkata91 Nov 15, 2021

Choose a reason for hiding this comment

SparkQA commented Nov 8, 2021

SparkQA commented Nov 8, 2021

SparkQA commented Nov 8, 2021

SparkQA commented Nov 8, 2021

SparkQA commented Nov 8, 2021

SparkQA commented Nov 8, 2021

SparkQA commented Dec 15, 2021

SparkQA commented Dec 15, 2021

SparkQA commented Dec 15, 2021

SparkQA commented Dec 16, 2021

SparkQA commented Dec 16, 2021

SparkQA commented Dec 16, 2021

SparkQA commented Dec 17, 2021

SparkQA commented Dec 17, 2021

SparkQA commented Dec 17, 2021

mridulm left a comment

Choose a reason for hiding this comment

mridulm commented Dec 19, 2021

venkata91 commented Dec 20, 2021

SparkQA commented Dec 20, 2021

SparkQA commented Dec 20, 2021

SparkQA commented Dec 20, 2021

venkata91 commented Dec 23, 2021

venkata91 commented Dec 23, 2021

venkata91 commented Dec 29, 2021 • edited Loading

mridulm left a comment

Choose a reason for hiding this comment

mridulm commented Jan 5, 2022

venkata91 commented Sep 2, 2021 •

edited

Loading

venkata91 commented Sep 20, 2021 •

edited

Loading

venkata91 commented Dec 29, 2021 •

edited

Loading