[SPARK-20774][SPARK-27036][SQL] Cancel the running broadcast execution on BroadcastTimeout #24595

jiangxb1987 · 2019-05-13T22:10:09Z

What changes were proposed in this pull request?

In the existing code, a broadcast execution timeout for the Future only causes a query failure, but the job running with the broadcast and the computation in the Future are not canceled. This wastes resources and slows down the other jobs. This PR tries to cancel both the running job and the running hashed relation construction thread.

How was this patch tested?

Add new test suite BroadcastExchangeExec

SparkQA · 2019-05-14T01:02:11Z

Test build #105370 has finished for PR 24595 at commit d2bbdbe.

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

jiangxb1987 · 2019-05-14T06:56:53Z

cc @zsxwing @liyichao @HyukjinKwon @markhamstra @cloud-fan @gatorsmile

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala

sql/core/src/test/scala/org/apache/spark/sql/execution/BroadcastExchangeSuite.scala

cloud-fan · 2019-05-14T09:21:36Z

This has a more narrow scope compared to #24036 , and is easier to reason about. +1

SparkQA · 2019-05-15T02:24:07Z

Test build #105393 has finished for PR 24595 at commit da8d53c.

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

cloud-fan · 2019-05-15T03:05:12Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala

+        // with the correct execution.
+        SQLExecution.withExecutionId(sqlContext.sparkSession, executionId) {
+          try {
+            sparkContext.setJobGroup(runId.toString, s"broadcast exchange (runId $runId)",


let's add a comment to explain why we set up a job group here. There is no other public API that can cancel a specific job AFAIK.

Just curious, why can't we just inherit the job group id of the outside thread so that when the SQL statement was cancelled, these broadcast sub-jobs can be cancelled as a whole?

That sounds like a good idea. We should only set the job group if there is no one outside.

Wouldn't that cancelling the broadcast job cause the outer main job to cancel?

We should only set the job group if there is no one outside.

and I guess it would be a partial fix?

It's always better to have fewer configs if possible. And I don't think we can override the job group id here if the config is true, as this is used to cancel broadcast after timeout.

@HyukjinKwon @jiangxb1987 @yeshengm What‘s your opinion of my idea?

I am discussing about multiple job group support which will fundamentally fix all these problems. This is actually a general problem that's not speicfic to SQL broadcast here only.

@HyukjinKwon Could you please tell me where you are discussing? I also want to make a little contribution.

sorry I am discussing offline first. I will send out an email or JIRA soon for more open discussion soon.

gatorsmile · 2019-05-15T16:58:27Z

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala

  override protected[sql] def doExecuteBroadcast[T](): broadcast.Broadcast[T] = {
    try {
-      ThreadUtils.awaitResult(relationFuture, timeout).asInstanceOf[broadcast.Broadcast[T]]
+      relationFuture.get(timeout.toSeconds, TimeUnit.SECONDS).asInstanceOf[broadcast.Broadcast[T]]


Also fix -1 too?

SparkQA · 2019-05-15T21:24:45Z

Test build #105425 has finished for PR 24595 at commit 5d82dd7.

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

gatorsmile

Thanks! Merged to master.

…n on BroadcastTimeout In the existing code, a broadcast execution timeout for the Future only causes a query failure, but the job running with the broadcast and the computation in the Future are not canceled. This wastes resources and slows down the other jobs. This PR tries to cancel both the running job and the running hashed relation construction thread. Add new test suite `BroadcastExchangeExec` Closes apache#24595 from jiangxb1987/SPARK-20774. Authored-by: Xingbo Jiang <[email protected]> Signed-off-by: gatorsmile <[email protected]>

yeshengm · 2020-10-23T08:10:52Z

@jiangxb1987 Just out of curiosity, the timeout mechanism was added because we have no way to track the lineage of such broadcast sub job right? Even if the "main" RDD action was cancelled, these broadcast jobs running in other threads will just keep running until it fails or timed out (which was added in this diff).

…tement is cancelled ### What changes were proposed in this pull request? #24595 introduced `private val runId: UUID = UUID.randomUUID` in `BroadcastExchangeExec` to cancel the broadcast execution in the Future when timeout happens. Since the runId is a random UUID instead of inheriting the job group id, when a SQL statement is cancelled, these broadcast sub-jobs are still executing. This PR uses the job group id of the outside thread as its `runId` to abort these broadcast sub-jobs when the SQL statement is cancelled. ### Why are the changes needed? When broadcasting a table takes too long and the SQL statement is cancelled. However, the background Spark job is still running and it wastes resources. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Manually test. Since broadcasting a table is too fast to cancel in UT, but it is very easy to verify manually: 1. Start a Spark thrift-server with less resource in YARN. 2. When the driver is running but no executors are launched, submit a SQL which will broadcast tables from beeline. 3. Cancel the SQL in beeline Without the patch, broadcast sub-jobs won't be cancelled. ![Screen Shot 2021-01-11 at 12 03 13 PM](https://user-images.githubusercontent.com/1853780/104150975-ab024b00-5416-11eb-8bf9-b5167bdad80a.png) With this patch, broadcast sub-jobs will be cancelled. ![Screen Shot 2021-01-11 at 11 43 40 AM](https://user-images.githubusercontent.com/1853780/104150994-be151b00-5416-11eb-80ff-313d423c8a2e.png) Closes #31119 from LantaoJin/SPARK-34064. Authored-by: LantaoJin <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

…tement is cancelled ### What changes were proposed in this pull request? #24595 introduced `private val runId: UUID = UUID.randomUUID` in `BroadcastExchangeExec` to cancel the broadcast execution in the Future when timeout happens. Since the runId is a random UUID instead of inheriting the job group id, when a SQL statement is cancelled, these broadcast sub-jobs are still executing. This PR uses the job group id of the outside thread as its `runId` to abort these broadcast sub-jobs when the SQL statement is cancelled. ### Why are the changes needed? When broadcasting a table takes too long and the SQL statement is cancelled. However, the background Spark job is still running and it wastes resources. ### Does this PR introduce _any_ user-facing change? No ### How was this patch tested? Manually test. Since broadcasting a table is too fast to cancel in UT, but it is very easy to verify manually: 1. Start a Spark thrift-server with less resource in YARN. 2. When the driver is running but no executors are launched, submit a SQL which will broadcast tables from beeline. 3. Cancel the SQL in beeline Without the patch, broadcast sub-jobs won't be cancelled. ![Screen Shot 2021-01-11 at 12 03 13 PM](https://user-images.githubusercontent.com/1853780/104150975-ab024b00-5416-11eb-8bf9-b5167bdad80a.png) With this patch, broadcast sub-jobs will be cancelled. ![Screen Shot 2021-01-11 at 11 43 40 AM](https://user-images.githubusercontent.com/1853780/104150994-be151b00-5416-11eb-80ff-313d423c8a2e.png) Closes #31119 from LantaoJin/SPARK-34064. Authored-by: LantaoJin <[email protected]> Signed-off-by: Wenchen Fan <[email protected]> (cherry picked from commit f1b21ba) Signed-off-by: Wenchen Fan <[email protected]>

Shockang · 2021-07-10T16:58:13Z

@jiangxb1987 @cloud-fan @yeshengm I just deleted the following two lines of code to solve this bug:
SPARK-35508 job group and description do not apply on broadcasts

sparkContext.setJobGroup(runId.toString, s"broadcast exchange (runId $runId)", interruptOnCancel = true)

Could you tell me the necessity of setJobGroup here? It will override the configuration of job group and job description in the user code.

HyukjinKwon · 2021-07-11T23:44:44Z

Then I think we'd have to wait for the broadcast job to finish even when the job cancellation is triggered for the main job, right?

Shockang · 2021-07-12T02:31:03Z

Then I think we'd have to wait for the broadcast job to finish even when the job cancellation is triggered for the main job, right?

Thank you for your attention. I mean even if it should support cancellation, it should not overwrite the user's configuration.I'm trying better approach to solve this problem

HyukjinKwon · 2021-07-12T02:33:06Z

I think maybe we should have a way to set multiple job groups actaully. That will make everything easier.

cloud-fan · 2021-07-12T02:49:21Z

@Shockang how do you think about this proposal? https://github.com/apache/spark/pull/24595/files#r667590820

Shockang · 2021-07-14T02:41:56Z

@Shockang how do you think about this proposal? https://github.com/apache/spark/pull/24595/files#r667590820

Sorry, I'm busy in the past several days. You can take a look at my suggestions.

cancel timeout broadcast execution

d2bbdbe

cloud-fan reviewed May 14, 2019

View reviewed changes

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/BroadcastExchangeExec.scala Outdated Show resolved Hide resolved

cloud-fan reviewed May 14, 2019

View reviewed changes

sql/core/src/test/scala/org/apache/spark/sql/execution/BroadcastExchangeSuite.scala Outdated Show resolved Hide resolved

jiangxb1987 changed the title ~~[SPARK-20774][CORE] Cancel the running broadcast execution on BroadcastTimeout~~ [SPARK-20774]SPARK-27036[CORE] Cancel the running broadcast execution on BroadcastTimeout May 14, 2019

jiangxb1987 changed the title ~~[SPARK-20774]SPARK-27036[CORE] Cancel the running broadcast execution on BroadcastTimeout~~ [SPARK-20774][SPARK-27036][CORE] Cancel the running broadcast execution on BroadcastTimeout May 14, 2019

address comments

da8d53c

cloud-fan reviewed May 15, 2019

View reviewed changes

cloud-fan approved these changes May 15, 2019

View reviewed changes

gatorsmile reviewed May 15, 2019

View reviewed changes

gatorsmile changed the title ~~[SPARK-20774][SPARK-27036][CORE] Cancel the running broadcast execution on BroadcastTimeout~~ [SPARK-20774][SPARK-27036][SQL] Cancel the running broadcast execution on BroadcastTimeout May 15, 2019

fix broadcastTimeout = -1 problem

5d82dd7

gatorsmile reviewed May 15, 2019

View reviewed changes

gatorsmile closed this in 0bba5cf May 15, 2019

ajithme mentioned this pull request Jan 27, 2020

[SPARK-22590][SQL] Copy sparkContext.localproperties to child thread in BroadcastExchangeExec.executionContext #27266

Closed

LantaoJin mentioned this pull request Jan 11, 2021

[SPARK-34064][SQL] Cancel the running broadcast sub-jobs when SQL statement is cancelled #31119

Closed

[SPARK-20774][SPARK-27036][SQL] Cancel the running broadcast execution on BroadcastTimeout #24595

[SPARK-20774][SPARK-27036][SQL] Cancel the running broadcast execution on BroadcastTimeout #24595

Uh oh!

Conversation

jiangxb1987 commented May 13, 2019

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented May 14, 2019

Uh oh!

jiangxb1987 commented May 14, 2019

Uh oh!

Uh oh!

Uh oh!

cloud-fan commented May 14, 2019

Uh oh!

SparkQA commented May 15, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

yeshengm Oct 23, 2020 • 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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HyukjinKwon Jul 15, 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!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented May 15, 2019

Uh oh!

gatorsmile left a comment

Choose a reason for hiding this comment

Uh oh!

yeshengm commented Oct 23, 2020

Uh oh!

Shockang commented Jul 10, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

HyukjinKwon commented Jul 11, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Shockang commented Jul 12, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

HyukjinKwon commented Jul 12, 2021

Uh oh!

cloud-fan commented Jul 12, 2021

Uh oh!

Shockang commented Jul 14, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

7 participants

yeshengm Oct 23, 2020 •

edited

Loading

HyukjinKwon Jul 15, 2021 •

edited

Loading

Shockang commented Jul 10, 2021 •

edited

Loading

HyukjinKwon commented Jul 11, 2021 •

edited

Loading

Shockang commented Jul 12, 2021 •

edited

Loading

Shockang commented Jul 14, 2021 •

edited

Loading