[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
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,75 @@ | ||
| ================================================================================================ | ||
| Join Benchmark | ||
| ================================================================================================ | ||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_181-b13 on Linux 3.10.0-862.3.2.el7.x86_64 | ||
| Intel(R) Xeon(R) CPU E5-2670 v2 @ 2.50GHz | ||
| Join w long: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| Join w long wholestage off 4464 / 4483 4.7 212.9 1.0X | ||
| Join w long wholestage on 289 / 339 72.6 13.8 15.5X | ||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_181-b13 on Linux 3.10.0-862.3.2.el7.x86_64 | ||
| Intel(R) Xeon(R) CPU E5-2670 v2 @ 2.50GHz | ||
| Join w long duplicated: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| Join w long duplicated wholestage off 5662 / 5678 3.7 270.0 1.0X | ||
| Join w long duplicated wholestage on 332 / 345 63.1 15.8 17.0X | ||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_181-b13 on Linux 3.10.0-862.3.2.el7.x86_64 | ||
| Intel(R) Xeon(R) CPU E5-2670 v2 @ 2.50GHz | ||
| Join w 2 ints: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| Join w 2 ints wholestage off 173174 / 173183 0.1 8257.6 1.0X | ||
| Join w 2 ints wholestage on 166350 / 198362 0.1 7932.2 1.0X | ||
|
Contributor
There was a problem hiding this comment. this surprises me that whole stage codegen doesn't help. We should investigate it later. |
||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_181-b13 on Linux 3.10.0-862.3.2.el7.x86_64 | ||
| Intel(R) Xeon(R) CPU E5-2670 v2 @ 2.50GHz | ||
| Join w 2 longs: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| Join w 2 longs wholestage off 7055 / 7214 3.0 336.4 1.0X | ||
| Join w 2 longs wholestage on 1869 / 1985 11.2 89.1 3.8X | ||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_181-b13 on Linux 3.10.0-862.3.2.el7.x86_64 | ||
| Intel(R) Xeon(R) CPU E5-2670 v2 @ 2.50GHz | ||
| Join w 2 longs duplicated: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| Join w 2 longs duplicated wholestage off 19256 / 20283 1.1 918.2 1.0X | ||
| Join w 2 longs duplicated wholestage on 2467 / 2544 8.5 117.7 7.8X | ||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_181-b13 on Linux 3.10.0-862.3.2.el7.x86_64 | ||
| Intel(R) Xeon(R) CPU E5-2670 v2 @ 2.50GHz | ||
| outer join w long: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| outer join w long wholestage off 3756 / 3761 5.6 179.1 1.0X | ||
| outer join w long wholestage on 218 / 250 96.2 10.4 17.2X | ||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_181-b13 on Linux 3.10.0-862.3.2.el7.x86_64 | ||
| Intel(R) Xeon(R) CPU E5-2670 v2 @ 2.50GHz | ||
| semi join w long: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| semi join w long wholestage off 2393 / 2416 8.8 114.1 1.0X | ||
| semi join w long wholestage on 214 / 218 97.9 10.2 11.2X | ||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_181-b13 on Linux 3.10.0-862.3.2.el7.x86_64 | ||
| Intel(R) Xeon(R) CPU E5-2670 v2 @ 2.50GHz | ||
| sort merge join: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| sort merge join wholestage off 2318 / 2392 0.9 1105.3 1.0X | ||
| sort merge join wholestage on 1669 / 1811 1.3 795.9 1.4X | ||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_181-b13 on Linux 3.10.0-862.3.2.el7.x86_64 | ||
| Intel(R) Xeon(R) CPU E5-2670 v2 @ 2.50GHz | ||
| sort merge join with duplicates: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| sort merge join with duplicates wholestage off 2966 / 2976 0.7 1414.5 1.0X | ||
| sort merge join with duplicates wholestage on 2413 / 2641 0.9 1150.5 1.2X | ||
|
|
||
| OpenJDK 64-Bit Server VM 1.8.0_181-b13 on Linux 3.10.0-862.3.2.el7.x86_64 | ||
| Intel(R) Xeon(R) CPU E5-2670 v2 @ 2.50GHz | ||
| shuffle hash join: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative | ||
| ------------------------------------------------------------------------------------------------ | ||
| shuffle hash join wholestage off 1475 / 1479 2.8 351.7 1.0X | ||
| shuffle hash join wholestage on 1209 / 1238 3.5 288.3 1.2X | ||
|
|
||
|
|
||
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,163 @@ 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 joins. | ||
| * 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("sort 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 |
||
| runBenchmark("Join Benchmark") { | ||
| 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.
Because the Mac has one more line than Linux:
28f9b9a#diff-45c96c65f7c46bc2d84843a7cb92f22fL7
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.
Ur.. I'm not a fan to piggy-backing. Okay.