[SPARK-26601][SQL] Make broadcast-exchange thread pool configurable #23670

caneGuy · 2019-01-28T08:27:08Z

What changes were proposed in this pull request?

Currently,thread number of broadcast-exchange thread pool is fixed and keepAliveSeconds is also fixed as 60s.

object BroadcastExchangeExec {
  private[execution] val executionContext = ExecutionContext.fromExecutorService(
    ThreadUtils.newDaemonCachedThreadPool("broadcast-exchange", 128))
}

 /**
   * Create a cached thread pool whose max number of threads is `maxThreadNumber`. Thread names
   * are formatted as prefix-ID, where ID is a unique, sequentially assigned integer.
   */
  def newDaemonCachedThreadPool(
      prefix: String, maxThreadNumber: Int, keepAliveSeconds: Int = 60): ThreadPoolExecutor = {
    val threadFactory = namedThreadFactory(prefix)
    val threadPool = new ThreadPoolExecutor(
      maxThreadNumber, // corePoolSize: the max number of threads to create before queuing the tasks
      maxThreadNumber, // maximumPoolSize: because we use LinkedBlockingDeque, this one is not used
      keepAliveSeconds,
      TimeUnit.SECONDS,
      new LinkedBlockingQueue[Runnable],
      threadFactory)
    threadPool.allowCoreThreadTimeOut(true)
    threadPool
  }

But some times, if the Thead object do not GC quickly it may caused server(driver) OOM. In such case,we need to make this thread pool configurable.
A case has described in https://issues.apache.org/jira/browse/SPARK-26601

How was this patch tested?

UT

SparkQA · 2019-01-28T08:34:51Z

Test build #101752 has finished for PR 23670 at commit 23ae9a0.

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

caneGuy · 2019-01-28T08:35:45Z

@viirya @maropu @HyukjinKwon
Sorry for faulty operation on #23519. i opened a new pr here and refined the comments except one.
And for the comment:
BroadcastExchangeExec.executionContext .getMaximumPoolSize
Actually i have searched the source code of ExecutionContext and could not find a solution to call getMaximumPoolSize since this api just declared by ThreadPoolExecutor. Any suggestion?Thanks
Unit test result on my desktop is below:

maropu · 2019-01-28T11:01:32Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

      .createWithDefault(1000)

+  val BROADCAST_EXCHANGE_MAX_THREAD_THREASHOLD =
+    buildStaticConf("spark.sql.broadcastExchange.maxThreadNumber")


spark.sql.broadcastExchange.maxThreadThreshold

maropu · 2019-01-28T11:19:13Z

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

+    // for other test
+    SparkSession.getActiveSession.get.sparkContext.conf.
+      set(StaticSQLConf.BROADCAST_EXCHANGE_MAX_THREAD_THREASHOLD, previousNumber)
+  }


We need this test? In the other similar prs (e.g., https://github.com/apache/spark/pull/22847/files), we added no test. If necessary, we can add tests in ExecutorSideSQLConfSuite to check if executors can reference this static value...

SparkQA · 2019-01-28T12:16:14Z

Test build #101755 has finished for PR 23670 at commit ba19ccc.

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

dongjoon-hyun · 2019-02-09T04:04:17Z

Hi, @caneGuy .
Newly added test case fails. Could you fix your test case failure?

org.apache.spark.sql.execution.ExchangeSuite.SPARK-26601: Make broadcast-exchange thread pool configurable

caneGuy · 2019-02-15T10:47:55Z

Thanks @dongjoon-hyun @maropu i remove the suite.since i think it's no need to test that.
And actually,i test locally,it has passed.

SparkQA · 2019-02-15T15:04:23Z

Test build #102387 has finished for PR 23670 at commit 2117ee2.

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

maropu · 2019-02-17T14:28:00Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala


+  val BROADCAST_EXCHANGE_MAX_THREAD_THREASHOLD =
+    buildStaticConf("spark.sql.broadcastExchange.maxThreadThreshold")
+      .doc("The maximum degree of parallelism to fetch and broadcast the table." +


nit: add a single space: ...table." ->...table. "

maropu · 2019-02-17T14:28:09Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

+    buildStaticConf("spark.sql.broadcastExchange.maxThreadThreshold")
+      .doc("The maximum degree of parallelism to fetch and broadcast the table." +
+        "If we encounter memory issue like frequently full GC or OOM when broadcast table " +
+        "we can decrease this number in order to reduce memory usage." +


Done @maropu

SparkQA · 2019-02-18T05:30:14Z

Test build #102439 has finished for PR 23670 at commit 27b0ec4.

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

caneGuy · 2019-02-21T02:59:47Z

Could you help retest this please? @maropu @dongjoon-hyun

SparkQA · 2019-02-21T04:42:16Z

Test build #102573 has finished for PR 23670 at commit 27b0ec4.

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

caneGuy · 2019-03-01T09:12:18Z

The failed unit tests is below which i think is not related with the pr @dongjoon-hyun @maropu
Caused by: sbt.ForkMain$ForkError: java.lang.ClassNotFoundException: javax.jdo.JDOException at java.net.URLClassLoader.findClass(URLClassLoader.java:382) at java.lang.ClassLoader.loadClass(ClassLoader.java:424) at org.apache.spark.sql.hive.client.IsolatedClientLoader$$anon$1.doLoadClass(IsolatedClientLoader.scala:231) at org.apache.spark.sql.hive.client.IsolatedClientLoader$$anon$1.loadClass(IsolatedClientLoader.scala:220) at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ... 47 more

viirya · 2019-03-01T09:32:41Z

retest this please.

SparkQA · 2019-03-01T12:52:59Z

Test build #102908 has finished for PR 23670 at commit 27b0ec4.

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

caneGuy · 2019-03-07T04:54:54Z

Sorry for bothering @viirya but the failed case i think is still not related with this pr.
Any suggestion?Thanks
org.scalatest.exceptions.TestFailedException: scala.Predef.Set.apply[Int](0, 1, 2, 3).map[org.apache.spark.sql.Row, scala.collection.immutable.Set[org.apache.spark.sql.Row]](((x$3: Int) => org.apache.spark.sql.Row.apply(x$3)))(immutable.this.Set.canBuildFrom[org.apache.spark.sql.Row]).subsetOf(scala.Predef.refArrayOps[org.apache.spark.sql.Row](results).toSet[org.apache.spark.sql.Row]) was false Stacktrace sbt.ForkMain$ForkError: org.scalatest.exceptions.TestFailedException: scala.Predef.Set.apply[Int](0, 1, 2, 3).map[org.apache.spark.sql.Row, scala.collection.immutable.Set[org.apache.spark.sql.Row]](((x$3: Int) => org.apache.spark.sql.Row.apply(x$3)))(immutable.this.Set.canBuildFrom[org.apache.spark.sql.Row]).subsetOf(scala.Predef.refArrayOps[org.apache.spark.sql.Row](results).toSet[org.apache.spark.sql.Row]) was false at org.scalatest.Assertions.newAssertionFailedException(Assertions.scala:528) at org.scalatest.Assertions.newAssertionFailedException$(Assertions.scala:527) at org.scalatest.FunSuite.newAssertionFailedException(FunSuite.scala:1560)

viirya · 2019-03-07T05:00:05Z

retest this please.

SparkQA · 2019-03-07T08:05:01Z

Test build #103120 has finished for PR 23670 at commit 27b0ec4.

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

dilipbiswal · 2019-03-07T08:09:44Z

retest this please

SparkQA · 2019-03-07T12:12:36Z

Test build #103129 has finished for PR 23670 at commit 27b0ec4.

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

caneGuy · 2019-03-26T11:46:25Z

@dilipbiswal I have resolve the conflict,please help check this review,thanks

SparkQA · 2019-03-26T15:49:22Z

Test build #103973 has finished for PR 23670 at commit 8da84e3.

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

caneGuy · 2019-04-11T02:57:54Z

Hi, @caneGuy . Could you fix the indentation as @attilapiros mentioned? (#23670 (comment))

Also, ping @hvanhovell , @cloud-fan , @gatorsmile . How do you think about this new configuration?

done @dongjoon-hyun thanks!

SparkQA · 2019-04-11T04:58:35Z

Test build #104495 has finished for PR 23670 at commit a4523e9.

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

caneGuy · 2019-04-18T01:20:37Z

failed case is not caused by this case
bt.ForkMain$ForkError: java.lang.RuntimeException: [unresolved dependency: com.sun.jersey#jersey-core;1.14: configuration not found in com.sun.jersey#jersey-core;1.14: 'master(compile)'. Missing configuration: 'compile'. It was required from org.apache.hadoop#hadoop-yarn-common;2.7.4 compile] at org.apache.spark.deploy.SparkSubmitUtils$.resolveMavenCoordinates(SparkSubmit.scala:1321) at
please help retest this please thanks @dongjoon-hyun

HyukjinKwon · 2019-04-18T01:25:11Z

retest this please

HyukjinKwon · 2019-04-18T01:25:28Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

+        "cause longer waiting for other broadcasting. Also, increasing parallelism may " +
+        "cause memory problem.")
+      .intConf
+      .checkValue(thres => thres > 0, "The threshold should be positive.")


If this number cannot be more than 128, let's add that condition.

+1 for checking thres <= 128, too.

HyukjinKwon · 2019-04-18T01:25:35Z

retest this please

HyukjinKwon · 2019-04-18T01:26:28Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

+        "cause longer waiting for other broadcasting. Also, increasing parallelism may " +
+        "cause memory problem.")
+      .intConf
+      .checkValue(thres => thres > 0, "The threshold should be positive.")


Let's make it .internal() so that we can remove this away if there's anything wrong found with this configuration later.

Yeah, I just want to comment like that. I also think this is advanced and should not be exposed to users in general. +1

HyukjinKwon · 2019-04-18T01:27:03Z

LGTM otheriwse. @cloud-fan, I will get this in. Please let me know if you have other concerns.

SparkQA · 2019-04-18T05:27:33Z

Test build #104686 has finished for PR 23670 at commit a4523e9.

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

SparkQA · 2019-04-18T05:54:12Z

Test build #104687 has finished for PR 23670 at commit a4523e9.

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

dongjoon-hyun · 2019-04-22T00:15:04Z

Retest this please.

SparkQA · 2019-04-22T03:24:31Z

Test build #104787 has finished for PR 23670 at commit a4523e9.

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

HyukjinKwon · 2019-04-22T09:56:17Z

ping @caneGuy

caneGuy · 2019-05-10T08:15:41Z

Sorry for late reply, i will fix today @HyukjinKwon

caneGuy · 2019-05-10T11:07:24Z

@HyukjinKwon refine!Thanks too much

SparkQA · 2019-05-10T14:22:18Z

Test build #105309 has finished for PR 23670 at commit 46f454f.

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

caneGuy · 2019-05-11T04:52:20Z

@HyukjinKwon ping thanks!

viirya · 2019-05-11T05:00:39Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

+        "cause longer waiting for other broadcasting. Also, increasing parallelism may " +
+        "cause memory problem.")
+      .intConf
+      .checkValue(thres => thres > 0 && thres <= 128, "The threshold should be positive.")


There is a maximum limit (128) for this config. The error message doesn't reflect that. Once an invalid value like 200 is set, it can't figure out what goes wrong with it, if not reading this code.

Yea let's improve this one before getting this in

Done @HyukjinKwon @viirya Thanks!

kiszk · 2019-05-12T10:05:06Z

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/StaticSQLConf.scala

      .intConf
      .createWithDefault(1000)

+  val BROADCAST_EXCHANGE_MAX_THREAD_THREASHOLD =


nit: BROADCAST_EXCHANGE_MAX_THREAD_THREASHOLD -> BROADCAST_EXCHANGE_MAX_THREAD_THRESHOLD

oh sorry for this error,i have fixed that

kiszk · 2019-05-12T10:05:17Z

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

  private[execution] val executionContext = ExecutionContext.fromExecutorService(
-    ThreadUtils.newDaemonCachedThreadPool("broadcast-exchange", 128))
+      ThreadUtils.newDaemonCachedThreadPool("broadcast-exchange",
+        SQLConf.get.getConf(StaticSQLConf.BROADCAST_EXCHANGE_MAX_THREAD_THREASHOLD)))


ping @caneGuy seems like the last nit

Done @HyukjinKwon @kiszk thanks!

SparkQA · 2019-05-12T10:08:33Z

Test build #105337 has finished for PR 23670 at commit a10dac3.

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

HyukjinKwon · 2019-05-13T11:40:56Z

Merged to master. Related tests and build passed

HyukjinKwon · 2019-05-13T11:41:25Z

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

 object BroadcastExchangeExec {
  private[execution] val executionContext = ExecutionContext.fromExecutorService(
-    ThreadUtils.newDaemonCachedThreadPool("broadcast-exchange", 128))
+      ThreadUtils.newDaemonCachedThreadPool("broadcast-exchange",


nit: the previous indentation was correct.

SparkQA · 2019-05-13T12:04:11Z

Test build #105351 has finished for PR 23670 at commit 40c3592.

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

caneGuy added 3 commits January 28, 2019 16:26

[SPARK-26601][SQL] Make broadcast-exchange thread pool configurable

0cd3149

Fix code style

cc9faa6

Fix code style

23ae9a0

Fix code style

ba19ccc

maropu reviewed Jan 28, 2019

View reviewed changes

Remove suite

2117ee2

maropu reviewed Feb 17, 2019

View reviewed changes

Refine comment

27b0ec4

Merge branch 'master' into zhoukang/make-broadcat-config

8da84e3

HyukjinKwon reviewed Apr 18, 2019

View reviewed changes

Refine comment

46f454f

HyukjinKwon approved these changes May 10, 2019

View reviewed changes

viirya reviewed May 11, 2019

View reviewed changes

Refine

a10dac3

kiszk reviewed May 12, 2019

View reviewed changes

Refine

40c3592

HyukjinKwon reviewed May 13, 2019

View reviewed changes

HyukjinKwon closed this in 126310c May 13, 2019

caneGuy deleted the zhoukang/make-broadcat-config branch June 1, 2020 02:15

[SPARK-26601][SQL] Make broadcast-exchange thread pool configurable #23670

[SPARK-26601][SQL] Make broadcast-exchange thread pool configurable #23670

Uh oh!

Conversation

caneGuy commented Jan 28, 2019

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

SparkQA commented Jan 28, 2019

Uh oh!

caneGuy commented Jan 28, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jan 28, 2019

Uh oh!

dongjoon-hyun commented Feb 9, 2019

Uh oh!

caneGuy commented Feb 15, 2019

Uh oh!

SparkQA commented Feb 15, 2019

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!

SparkQA commented Feb 18, 2019

Uh oh!

caneGuy commented Feb 21, 2019

Uh oh!

SparkQA commented Feb 21, 2019

Uh oh!

caneGuy commented Mar 1, 2019

Uh oh!

viirya commented Mar 1, 2019

Uh oh!

SparkQA commented Mar 1, 2019

Uh oh!

caneGuy commented Mar 7, 2019

Uh oh!

viirya commented Mar 7, 2019

Uh oh!

SparkQA commented Mar 7, 2019

Uh oh!

dilipbiswal commented Mar 7, 2019

Uh oh!

SparkQA commented Mar 7, 2019

Uh oh!

caneGuy commented Mar 26, 2019

Uh oh!

SparkQA commented Mar 26, 2019

Uh oh!

caneGuy commented Apr 11, 2019

Uh oh!

SparkQA commented Apr 11, 2019

Uh oh!

caneGuy commented Apr 18, 2019

Uh oh!

HyukjinKwon commented Apr 18, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

HyukjinKwon commented Apr 18, 2019

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 commented Apr 18, 2019

caneGuy commented Jan 28, 2019 •

edited

Loading

viirya May 11, 2019 •

edited

Loading