Skip to content

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

Closed
ajithme wants to merge 13 commits into
apache:masterfrom
ajithme:broadcastlocalprop
Closed

[SPARK-22590][SQL] Copy sparkContext.localproperties to child thread in BroadcastExchangeExec.executionContext#27266
ajithme wants to merge 13 commits into
apache:masterfrom
ajithme:broadcastlocalprop

Conversation

@ajithme

Copy link
Copy Markdown
Contributor

What changes were proposed in this pull request?

In org.apache.spark.sql.execution.exchange.BroadcastExchangeExec#relationFuture make a copy of org.apache.spark.SparkContext#localProperties and pass it to the broadcast execution thread in org.apache.spark.sql.execution.exchange.BroadcastExchangeExec#executionContext

Why are the changes needed?

When executing BroadcastExchangeExec, the relationFuture is evaluated via a separate thread. The threads inherit the localProperties from sparkContext as they are the child threads.
These threads are created in the executionContext (thread pools). Each Thread pool has a default keepAliveSeconds of 60 seconds for idle threads.
Scenarios where the thread pool has threads which are idle and reused for a subsequent new query, the thread local properties will not be inherited from spark context (thread properties are inherited only on thread creation) hence end up having old or no properties set. This will cause taskset properties to be missing when properties are transferred by child thread via sparkContext.runJob/submitJob

Does this PR introduce any user-facing change?

No

How was this patch tested?

Added UT

@ajithme

Copy link
Copy Markdown
ContributorAuthor

@dongjoon-hyun

Copy link
Copy Markdown
Member

Thank you for pinging me, @ajithme .

@dongjoon-hyun

Copy link
Copy Markdown
Member

ok to test

@dongjoon-hyundongjoon-hyun changed the title [SPARK-22590] Broadcast thread propagates the local properties to task[SPARK-22590][SQL] Broadcast thread propagates the local properties to taskJan 17, 2020
@srowen

Copy link
Copy Markdown
Member

Closely related to #27267 ?

@SparkQA

Copy link
Copy Markdown

Test build #116962 has finished for PR 27266 at commit 8c9fbe6.

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

@ajithmeajithme changed the title [SPARK-22590][SQL] Broadcast thread propagates the local properties to task[WIP][SPARK-22590][SQL] Broadcast thread propagates the local properties to taskJan 22, 2020
@ajithmeajithme changed the title [WIP][SPARK-22590][SQL] Broadcast thread propagates the local properties to task[SPARK-22590][SQL] Copy sparkContext.localproperties to child thread in BroadcastExchangeExec.executionContextJan 23, 2020
@ajithme

Copy link
Copy Markdown
ContributorAuthor

@SparkQA

Copy link
Copy Markdown

Test build #117305 has finished for PR 27266 at commit f850797.

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

@SparkQA

Copy link
Copy Markdown

Test build #117310 has finished for PR 27266 at commit 939ee0d.

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

@ajithme

Copy link
Copy Markdown
ContributorAuthor

this failure seems unrelated. Please retest this

@ajithme

Copy link
Copy Markdown
ContributorAuthor

gentle ping cc @dongjoon-hyun@cloud-fan@hvanhovell

@hvanhovellhvanhovellJan 27, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

How about we make this also produce a (java) Future? If you do this then we can unify this method with the other withThreadLocalCaptured method.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I see previously it was using a Future which was moved to Callable in #24595

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It needed to be cancellable. So we are using java futures instead.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I understand, will update to use java Futures instead

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@hvanhovell org.apache.spark.sql.execution.SubqueryExec#executeCollect needs now to await over java Future instead of scala future which would mean we have to have a new method under org.apache.spark.util.ThreadUtils#awaitResult, is this acceptable.?

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it's fine to have another awaitResult

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1

@SparkQA

Copy link
Copy Markdown

Test build #117438 has finished for PR 27266 at commit 9ab74a0.

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

@SparkQA

Copy link
Copy Markdown

Test build #117439 has finished for PR 27266 at commit b2a1bab.

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

@SparkQA

Copy link
Copy Markdown

Test build #117441 has finished for PR 27266 at commit 5eb5a96.

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

Comment threadsql/core/src/main/scala/org/apache/spark/sql/execution/SQLExecution.scala Outdated

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I usually don't like such changes. Wonder if we can keep the indentation same and make it easier to track the history of commits.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@HyukjinKwon I have tried to adjust the indent so that diff seems to show only lines i have modified. Does it seem ok now.?

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The indentation is wrong now although the diff is smaller. how about

private[sql] lazy val relationFuture: Future[broadcast.Broadcast[Any]] = {
SQLExecution.withThreadLocalCaptured ... {
doBroadcast(..)
}
}
private def doBroadcast ... {
// original code
}

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@cloud-fan updated as per your suggestion, but still it changes indent than original. Is it ok.?

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ah sorry I misread the code. Seems we can't avoid changing the indentation as it was so nested before. I'm OK with your original code.

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

reverted

@cloud-fan

Copy link
Copy Markdown
Contributor

@ajithme are you still working on it? I'd like to get this in before 3.0 release and take over this one if you don't have time to complete it.

@ajithme

Copy link
Copy Markdown
ContributorAuthor

@ajithme are you still working on it? I'd like to get this in before 3.0 release and take over this one if you don't have time to complete it.

Yes. I will update the PR shortly. Sorry for delay

@ajithme

Copy link
Copy Markdown
ContributorAuthor

@cloud-fan@hvanhovell@HyukjinKwon Updated as per all the comments, please review

Comment threadcore/src/main/scala/org/apache/spark/util/ThreadUtils.scala
@SparkQA

Copy link
Copy Markdown

Test build #118583 has finished for PR 27266 at commit 11fffca.

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

@SparkQA

Copy link
Copy Markdown

Test build #118581 has finished for PR 27266 at commit 0f39043.

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

@SparkQA

Copy link
Copy Markdown

Test build #118585 has finished for PR 27266 at commit 763d1bc.

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

@hvanhovellhvanhovell left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM - pending jenkins

@SparkQA

Copy link
Copy Markdown

Test build #118586 has finished for PR 27266 at commit 2ed76c3.

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

@cloud-fan

Copy link
Copy Markdown
Contributor

thanks, merging to master/3.0!

cloud-fan pushed a commit that referenced this pull request Feb 17, 2020
…in BroadcastExchangeExec.executionContext
### What changes were proposed in this pull request?
In `org.apache.spark.sql.execution.exchange.BroadcastExchangeExec#relationFuture` make a copy of `org.apache.spark.SparkContext#localProperties` and pass it to the broadcast execution thread in `org.apache.spark.sql.execution.exchange.BroadcastExchangeExec#executionContext`
### Why are the changes needed?
When executing `BroadcastExchangeExec`, the relationFuture is evaluated via a separate thread. The threads inherit the `localProperties` from `sparkContext` as they are the child threads.
These threads are created in the executionContext (thread pools). Each Thread pool has a default `keepAliveSeconds` of 60 seconds for idle threads.
Scenarios where the thread pool has threads which are idle and reused for a subsequent new query, the thread local properties will not be inherited from spark context (thread properties are inherited only on thread creation) hence end up having old or no properties set. This will cause taskset properties to be missing when properties are transferred by child thread via `sparkContext.runJob/submitJob`
### Does this PR introduce any user-facing change?
No
### How was this patch tested?
Added UT
Closes#27266 from ajithme/broadcastlocalprop.
Authored-by: Ajith <ajith2489@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
(cherry picked from commit 2854091)
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
@SparkQA

Copy link
Copy Markdown

Test build #118582 has finished for PR 27266 at commit 742d322.

  • This patch fails from timeout after a configured wait of 400m.
  • This patch merges cleanly.
  • This patch adds no public classes.

@HyukjinKwon

Copy link
Copy Markdown
Member

+1 from me too. Thanks @ajithme.

sjincho pushed a commit to sjincho/spark that referenced this pull request Apr 15, 2020
…in BroadcastExchangeExec.executionContext
### What changes were proposed in this pull request?
In `org.apache.spark.sql.execution.exchange.BroadcastExchangeExec#relationFuture` make a copy of `org.apache.spark.SparkContext#localProperties` and pass it to the broadcast execution thread in `org.apache.spark.sql.execution.exchange.BroadcastExchangeExec#executionContext`
### Why are the changes needed?
When executing `BroadcastExchangeExec`, the relationFuture is evaluated via a separate thread. The threads inherit the `localProperties` from `sparkContext` as they are the child threads.
These threads are created in the executionContext (thread pools). Each Thread pool has a default `keepAliveSeconds` of 60 seconds for idle threads.
Scenarios where the thread pool has threads which are idle and reused for a subsequent new query, the thread local properties will not be inherited from spark context (thread properties are inherited only on thread creation) hence end up having old or no properties set. This will cause taskset properties to be missing when properties are transferred by child thread via `sparkContext.runJob/submitJob`
### Does this PR introduce any user-facing change?
No
### How was this patch tested?
Added UT
Closesapache#27266 from ajithme/broadcastlocalprop.
Authored-by: Ajith <ajith2489@gmail.com>
Signed-off-by: Wenchen Fan <wenchen@databricks.com>
Sign up for freeto join this conversation on GitHub. Already have an account? Sign in to comment

Labels

Projects

None yet

Development

Successfully merging this pull request may close these issues.

7 participants

@ajithme@dongjoon-hyun@srowen@SparkQA@cloud-fan@HyukjinKwon@hvanhovell