[SPARK-4570][SQL]add BroadcastLeftSemiJoinHash #3442

wangxiaojing · 2014-11-25T04:09:19Z

JIRA issue: SPARK-4570
We are planning to create a BroadcastLeftSemiJoinHash to implement the broadcast join for left semijoin
In left semijoin :
If the size of data from right side is smaller than the user-settable threshold AUTO_BROADCASTJOIN_THRESHOLD,
the planner would mark it as the broadcast relation and mark the other relation as the stream side. The broadcast table will be broadcasted to all of the executors involved in the join, as a org.apache.spark.broadcast.Broadcast object. It will use joins.BroadcastLeftSemiJoinHash.,else it will use joins.LeftSemiJoinHash.

The benchmark suggests these made the optimized version 4x faster when left semijoin


Original:
left semi join : 9288 ms 
Optimized:
left semi join : 1963 ms

The micro benchmark load data1/kv3.txt into a normal Hive table.
Benchmark code:


 def benchmark(f: => Unit) = {
    val begin = System.currentTimeMillis()
    f
    val end = System.currentTimeMillis()
    end - begin
  }
  val sc = new SparkContext(
    new SparkConf()
      .setMaster("local")
      .setAppName(getClass.getSimpleName.stripSuffix("$")))
  val hiveContext = new HiveContext(sc)
  import hiveContext._
  sql("drop table if exists left_table")
  sql("drop table if exists right_table")
  sql( """create table left_table (key int, value string)
       """.stripMargin)
  sql( s"""load data local inpath "/data1/kv3.txt" into table left_table""")
  sql( """create table right_table (key int, value string)
       """.stripMargin)
  sql(
    """
      |from left_table
      |insert overwrite table right_table
      |select left_table.key, left_table.value
    """.stripMargin)

  val leftSimeJoin = sql(
    """select a.key from left_table a
      |left semi join right_table b on a.key = b.key""".stripMargin)
  val leftSemiJoinDuration = benchmark(leftSimeJoin.count())
  println(s"left semi join : $leftSemiJoinDuration ms ")

AmplabJenkins · 2014-11-25T04:12:10Z

Can one of the admins verify this patch?

wangxiaojing · 2014-12-01T01:45:33Z

@liancheng

liancheng · 2014-12-01T05:31:08Z

ok to test

SparkQA · 2014-12-01T05:32:38Z

Test build #23972 has started for PR 3442 at commit 3a63ecb.

This patch merges cleanly.

SparkQA · 2014-12-01T06:33:57Z

Test build #23972 has finished for PR 3442 at commit 3a63ecb.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class BroadcastLeftSemiJoinHash(

AmplabJenkins · 2014-12-01T06:33:59Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/23972/
Test FAILed.

wangxiaojing · 2014-12-01T07:59:40Z

@liancheng

liancheng · 2014-12-01T08:46:27Z

retest this please

SparkQA · 2014-12-01T08:50:12Z

Test build #23983 has started for PR 3442 at commit 3a63ecb.

This patch merges cleanly.

SparkQA · 2014-12-01T09:53:52Z

Test build #23983 has finished for PR 3442 at commit 3a63ecb.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class BroadcastLeftSemiJoinHash(

AmplabJenkins · 2014-12-01T09:53:55Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/23983/
Test PASSed.

marmbrus · 2014-12-01T23:45:03Z

sql/core/src/main/scala/org/apache/spark/sql/execution/joins/BroadcastLeftSemiJoinHash.scala

+
+  override def execute() = {
+
+    val buildIter= buildPlan.execute().map(_.copy()).collect().toIterator


Remove blank line

marmbrus · 2014-12-01T23:48:24Z

Thanks for working on this!

liancheng · 2014-12-02T04:07:53Z

sql/core/src/test/scala/org/apache/spark/sql/JoinSuite.scala

+      case (query, joinClass) => assertJoin(query, joinClass)
+    }
+
+    sql( s"""SET ${SQLConf.AUTO_BROADCASTJOIN_THRESHOLD}=-$tmp""")


-$tmp: typo? And we can just use setConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD, tmp.toString) here.

SparkQA · 2014-12-02T05:04:59Z

Test build #24025 has started for PR 3442 at commit d410f67.

This patch merges cleanly.

SparkQA · 2014-12-02T06:11:00Z

Test build #24025 has finished for PR 3442 at commit d410f67.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class BroadcastLeftSemiJoinHash(

AmplabJenkins · 2014-12-02T06:11:02Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24025/
Test PASSed.

SparkQA · 2014-12-04T01:25:21Z

Test build #24122 has started for PR 3442 at commit 4fdcfe7.

This patch merges cleanly.

SparkQA · 2014-12-04T02:26:11Z

Test build #24122 has finished for PR 3442 at commit 4fdcfe7.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class BroadcastLeftSemiJoinHash(

AmplabJenkins · 2014-12-04T02:26:14Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24122/
Test PASSed.

SparkQA · 2014-12-17T06:42:24Z

Test build #24533 has started for PR 3442 at commit 3a58191.

This patch merges cleanly.

SparkQA · 2014-12-17T06:42:28Z

Test build #24533 has finished for PR 3442 at commit 3a58191.

This patch fails RAT tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
- case class BroadcastLeftSemiJoinHash(

AmplabJenkins · 2014-12-17T06:42:30Z

Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24533/
Test FAILed.

SparkQA · 2014-12-24T02:52:23Z

Test build #24753 has started for PR 3442 at commit a4a43c9.

This patch merges cleanly.

SparkQA · 2014-12-24T04:02:46Z

Test build #24753 has finished for PR 3442 at commit a4a43c9.

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

AmplabJenkins · 2014-12-24T04:02:49Z

Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/24753/
Test PASSed.

marmbrus · 2014-12-30T21:54:31Z

Thanks, merged to master.

marmbrus reviewed Dec 1, 2014
View reviewed changes

liancheng reviewed Dec 2, 2014
View reviewed changes

wangxiaojing force-pushed the SPARK-4570 branch from 4fdcfe7 to 3a58191 Compare December 17, 2014 06:38

wangxiaojing added 2 commits December 23, 2014 00:33

add BroadcastLeftSemiJoinHash

1a8da2a

add testsuite

ff2e618

wangxiaojing added 2 commits December 23, 2014 00:36

change style

fbe4887

change style

f103983

wangxiaojing force-pushed the SPARK-4570 branch from 3a58191 to f103983 Compare December 24, 2014 02:49

rebase

a4a43c9

asfgit closed this in 07fa191 Dec 30, 2014

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-4570][SQL]add BroadcastLeftSemiJoinHash #3442

[SPARK-4570][SQL]add BroadcastLeftSemiJoinHash #3442

wangxiaojing commented Nov 25, 2014

AmplabJenkins commented Nov 25, 2014

wangxiaojing commented Dec 1, 2014

liancheng commented Dec 1, 2014

SparkQA commented Dec 1, 2014

SparkQA commented Dec 1, 2014

AmplabJenkins commented Dec 1, 2014

wangxiaojing commented Dec 1, 2014

liancheng commented Dec 1, 2014

SparkQA commented Dec 1, 2014

SparkQA commented Dec 1, 2014

AmplabJenkins commented Dec 1, 2014

marmbrus Dec 1, 2014

marmbrus commented Dec 1, 2014

liancheng Dec 2, 2014

SparkQA commented Dec 2, 2014

SparkQA commented Dec 2, 2014

AmplabJenkins commented Dec 2, 2014

SparkQA commented Dec 4, 2014

SparkQA commented Dec 4, 2014

AmplabJenkins commented Dec 4, 2014

SparkQA commented Dec 17, 2014

SparkQA commented Dec 17, 2014

AmplabJenkins commented Dec 17, 2014

SparkQA commented Dec 24, 2014

SparkQA commented Dec 24, 2014

AmplabJenkins commented Dec 24, 2014

marmbrus commented Dec 30, 2014


		override def execute() = {

		val buildIter= buildPlan.execute().map(_.copy()).collect().toIterator

[SPARK-4570][SQL]add BroadcastLeftSemiJoinHash #3442

[SPARK-4570][SQL]add BroadcastLeftSemiJoinHash #3442

Conversation

wangxiaojing commented Nov 25, 2014

AmplabJenkins commented Nov 25, 2014

wangxiaojing commented Dec 1, 2014

liancheng commented Dec 1, 2014

SparkQA commented Dec 1, 2014

SparkQA commented Dec 1, 2014

AmplabJenkins commented Dec 1, 2014

wangxiaojing commented Dec 1, 2014

liancheng commented Dec 1, 2014

SparkQA commented Dec 1, 2014

SparkQA commented Dec 1, 2014

AmplabJenkins commented Dec 1, 2014

marmbrus Dec 1, 2014

Choose a reason for hiding this comment

marmbrus commented Dec 1, 2014

liancheng Dec 2, 2014

Choose a reason for hiding this comment

SparkQA commented Dec 2, 2014

SparkQA commented Dec 2, 2014

AmplabJenkins commented Dec 2, 2014

SparkQA commented Dec 4, 2014

SparkQA commented Dec 4, 2014

AmplabJenkins commented Dec 4, 2014

SparkQA commented Dec 17, 2014

SparkQA commented Dec 17, 2014

AmplabJenkins commented Dec 17, 2014

SparkQA commented Dec 24, 2014

SparkQA commented Dec 24, 2014

AmplabJenkins commented Dec 24, 2014

marmbrus commented Dec 30, 2014