[SPARK-25664][SQL][TEST] Refactor JoinBenchmark to use main method #22661
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,80 @@ | ||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_151-b12 on Mac OS X 10.12.6 | ||
| Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz | ||
|
|
||
| Join w long: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| Join w long wholestage off 3200 / 3226 6.6 152.6 1.0X | ||
| Join w long wholestage on 223 / 261 94.2 10.6 14.4X | ||
|
|
||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_151-b12 on Mac OS X 10.12.6 | ||
| Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz | ||
|
|
||
| Join w long duplicated: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| Join w long duplicated wholestage off 3727 / 3745 5.6 177.7 1.0X | ||
| Join w long duplicated wholestage on 212 / 233 98.8 10.1 17.6X | ||
|
|
||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_151-b12 on Mac OS X 10.12.6 | ||
| Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz | ||
|
|
||
| Join w 2 ints: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| Join w 2 ints wholestage off 151556 / 152313 0.1 7226.8 1.0X | ||
| Join w 2 ints wholestage on 131599 / 138817 0.2 6275.1 1.2X | ||
|
|
||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_151-b12 on Mac OS X 10.12.6 | ||
| Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz | ||
|
|
||
| Join w 2 longs: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| Join w 2 longs wholestage off 5608 / 6035 3.7 267.4 1.0X | ||
| Join w 2 longs wholestage on 2256 / 2504 9.3 107.6 2.5X | ||
|
|
||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_151-b12 on Mac OS X 10.12.6 | ||
| Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz | ||
|
|
||
| Join w 2 longs duplicated: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| Join w 2 longs duplicated wholestage off 10976 / 11120 1.9 523.4 1.0X | ||
| Join w 2 longs duplicated wholestage on 1781 / 1849 11.8 84.9 6.2X | ||
|
|
||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_151-b12 on Mac OS X 10.12.6 | ||
| Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz | ||
|
|
||
| outer join w long: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| outer join w long wholestage off 2436 / 2453 8.6 116.1 1.0X | ||
| outer join w long wholestage on 140 / 144 149.4 6.7 17.3X | ||
|
|
||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_151-b12 on Mac OS X 10.12.6 | ||
| Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz | ||
|
|
||
| semi join w long: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| semi join w long wholestage off 1610 / 1625 13.0 76.8 1.0X | ||
| semi join w long wholestage on 142 / 148 148.0 6.8 11.4X | ||
|
|
||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_151-b12 on Mac OS X 10.12.6 | ||
| Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz | ||
|
|
||
| merge join: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| merge join wholestage off 864 / 910 2.4 412.0 1.0X | ||
| merge join wholestage on 611 / 617 3.4 291.4 1.4X | ||
|
|
||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_151-b12 on Mac OS X 10.12.6 | ||
| Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz | ||
|
|
||
| sort merge join with duplicates: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| sort merge join with duplicates wholestage off 1481 / 1597 1.4 706.2 1.0X | ||
| sort merge join with duplicates wholestage on 1192 / 1300 1.8 568.4 1.2X | ||
|
|
||
| Java HotSpot(TM) 64-Bit Server VM 1.8.0_151-b12 on Mac OS X 10.12.6 | ||
| Intel(R) Core(TM) i7-7820HQ CPU @ 2.90GHz | ||
|
|
||
| shuffle hash join: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| shuffle hash join wholestage off 1012 / 1034 4.1 241.2 1.0X | ||
| shuffle hash join wholestage on 799 / 861 5.2 190.5 1.3X | ||
|
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,229 +19,161 @@ package org.apache.spark.sql.execution.benchmark | |
|
|
||
| import org.apache.spark.sql.execution.joins._ | ||
| import org.apache.spark.sql.functions._ | ||
| import org.apache.spark.sql.internal.SQLConf | ||
| import org.apache.spark.sql.types.IntegerType | ||
|
|
||
| /** | ||
| * Benchmark to measure performance for aggregate primitives. | ||
| * To run this benchmark: | ||
| * {{{ | ||
| * 1. without sbt: | ||
| * bin/spark-submit --class <this class> --jars <spark core test jar> <spark sql test jar> | ||
| * 2. build/sbt "sql/test:runMain <this class>" | ||
| * 3. generate result: | ||
| * SPARK_GENERATE_BENCHMARK_FILES=1 build/sbt "sql/test:runMain <this class>" | ||
| * Results will be written to "benchmarks/JoinBenchmark-results.txt". | ||
| * }}} | ||
| */ | ||
| object JoinBenchmark extends SqlBasedBenchmark { | ||
|
|
||
| def broadcastHashJoinLongKey(): Unit = { | ||
| val N = 20 << 20 | ||
| val M = 1 << 16 | ||
|
|
||
| val dim = broadcast(spark.range(M).selectExpr("id as k", "cast(id as string) as v")) | ||
| codegenBenchmark("Join w long", N) { | ||
| val df = spark.range(N).join(dim, (col("id") % M) === col("k")) | ||
| assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[BroadcastHashJoinExec]).isDefined) | ||
| df.count() | ||
| } | ||
| } | ||
|
|
||
| def broadcastHashJoinLongKeyWithDuplicates(): Unit = { | ||
| val N = 20 << 20 | ||
| val M = 1 << 16 | ||
|
Member
There was a problem hiding this comment. So, this is a removal of redundant one, right? |
||
| val dim = broadcast(spark.range(M).selectExpr("cast(id/10 as long) as k")) | ||
|
Member
There was a problem hiding this comment. For this change, we need rerun the benchmark to get a new result. |
||
| codegenBenchmark("Join w long duplicated", N) { | ||
| val df = spark.range(N).join(dim, (col("id") % M) === col("k")) | ||
| assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[BroadcastHashJoinExec]).isDefined) | ||
| df.count() | ||
| } | ||
| } | ||
|
|
||
| def broadcastHashJoinTwoIntKey(): Unit = { | ||
| val N = 20 << 20 | ||
| val M = 1 << 16 | ||
| val dim2 = broadcast(spark.range(M) | ||
| .selectExpr("cast(id as int) as k1", "cast(id as int) as k2", "cast(id as string) as v")) | ||
|
|
||
| codegenBenchmark("Join w 2 ints", N) { | ||
| val df = spark.range(N).join(dim2, | ||
| (col("id") % M).cast(IntegerType) === col("k1") | ||
| && (col("id") % M).cast(IntegerType) === col("k2")) | ||
| assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[BroadcastHashJoinExec]).isDefined) | ||
| df.count() | ||
| } | ||
|
Member
There was a problem hiding this comment. Hi, @cloud-fan , @gatorsmile , @davies , @rxin . We are hitting some performance slowdown in benchmark. However, this is not a regression because it's consistent in 2.0.2 ~ 2.4.0-rc3. According to the original performance number, it seems to be a result when Did we really want to measure the difference in
Member
There was a problem hiding this comment. Any advice is welcome and thank you in advance, @cloud-fan , @gatorsmile , @davies , @rxin .
Contributor
There was a problem hiding this comment. This seems caused by the bug fix: #15390 So the performance is reasonable. |
||
| } | ||
|
|
||
| def broadcastHashJoinTwoLongKey(): Unit = { | ||
| val N = 20 << 20 | ||
| val M = 1 << 16 | ||
| val dim3 = broadcast(spark.range(M) | ||
| .selectExpr("id as k1", "id as k2", "cast(id as string) as v")) | ||
|
|
||
| codegenBenchmark("Join w 2 longs", N) { | ||
| val df = spark.range(N).join(dim3, | ||
| (col("id") % M) === col("k1") && (col("id") % M) === col("k2")) | ||
| assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[BroadcastHashJoinExec]).isDefined) | ||
| df.count() | ||
| } | ||
| } | ||
|
|
||
| def broadcastHashJoinTwoLongKeyWithDuplicates(): Unit = { | ||
| val N = 20 << 20 | ||
| val M = 1 << 16 | ||
| val dim4 = broadcast(spark.range(M) | ||
| .selectExpr("cast(id/10 as long) as k1", "cast(id/10 as long) as k2")) | ||
|
|
||
| codegenBenchmark("Join w 2 longs duplicated", N) { | ||
| val df = spark.range(N).join(dim4, | ||
| (col("id") bitwiseAND M) === col("k1") && (col("id") bitwiseAND M) === col("k2")) | ||
| assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[BroadcastHashJoinExec]).isDefined) | ||
| df.count() | ||
| } | ||
| } | ||
|
|
||
| def broadcastHashJoinOuterJoinLongKey(): Unit = { | ||
| val N = 20 << 20 | ||
| val M = 1 << 16 | ||
| val dim = broadcast(spark.range(M).selectExpr("id as k", "cast(id as string) as v")) | ||
| codegenBenchmark("outer join w long", N) { | ||
| val df = spark.range(N).join(dim, (col("id") % M) === col("k"), "left") | ||
| assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[BroadcastHashJoinExec]).isDefined) | ||
| df.count() | ||
| } | ||
| } | ||
|
|
||
| def broadcastHashJoinSemiJoinLongKey(): Unit = { | ||
| val N = 20 << 20 | ||
| val M = 1 << 16 | ||
| val dim = broadcast(spark.range(M).selectExpr("id as k", "cast(id as string) as v")) | ||
| codegenBenchmark("semi join w long", N) { | ||
| val df = spark.range(N).join(dim, (col("id") % M) === col("k"), "leftsemi") | ||
| assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[BroadcastHashJoinExec]).isDefined) | ||
| df.count() | ||
| } | ||
| } | ||
|
|
||
| def sortMergeJoin(): Unit = { | ||
| val N = 2 << 20 | ||
| codegenBenchmark("merge join", N) { | ||
|
||
| val df1 = spark.range(N).selectExpr(s"id * 2 as k1") | ||
| val df2 = spark.range(N).selectExpr(s"id * 3 as k2") | ||
| val df = df1.join(df2, col("k1") === col("k2")) | ||
| assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[SortMergeJoinExec]).isDefined) | ||
| df.count() | ||
| } | ||
| } | ||
|
|
||
| def sortMergeJoinWithDuplicates(): Unit = { | ||
| val N = 2 << 20 | ||
| codegenBenchmark("sort merge join with duplicates", N) { | ||
| val df1 = spark.range(N) | ||
| .selectExpr(s"(id * 15485863) % ${N*10} as k1") | ||
| val df2 = spark.range(N) | ||
| .selectExpr(s"(id * 15485867) % ${N*10} as k2") | ||
| val df = df1.join(df2, col("k1") === col("k2")) | ||
| assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[SortMergeJoinExec]).isDefined) | ||
| df.count() | ||
| } | ||
| } | ||
|
|
||
| def shuffleHashJoin(): Unit = { | ||
| val N: Long = 4 << 20 | ||
| withSQLConf( | ||
| SQLConf.SHUFFLE_PARTITIONS.key -> "2", | ||
| SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "10000000", | ||
| SQLConf.PREFER_SORTMERGEJOIN.key -> "false") { | ||
| codegenBenchmark("shuffle hash join", N) { | ||
| val df1 = spark.range(N).selectExpr(s"id as k1") | ||
| val df2 = spark.range(N / 3).selectExpr(s"id * 3 as k2") | ||
| val df = df1.join(df2, col("k1") === col("k2")) | ||
| assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[ShuffledHashJoinExec]).isDefined) | ||
| df.count() | ||
| } | ||
| } | ||
| } | ||
|
|
||
| override def runBenchmarkSuite(): Unit = { | ||
|
Member
There was a problem hiding this comment. Could you wrap the followings(line 168~177) with something like |
||
| broadcastHashJoinLongKey() | ||
| broadcastHashJoinLongKeyWithDuplicates() | ||
| broadcastHashJoinTwoIntKey() | ||
| broadcastHashJoinTwoLongKey() | ||
| broadcastHashJoinTwoLongKeyWithDuplicates() | ||
| broadcastHashJoinOuterJoinLongKey() | ||
| broadcastHashJoinSemiJoinLongKey() | ||
| sortMergeJoin() | ||
| sortMergeJoinWithDuplicates() | ||
| shuffleHashJoin() | ||
| } | ||
| } | ||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
aggregate primitives->joins