[
https://issues.apache.org/jira/browse/SPARK-20313?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16148463#comment-16148463
]
Tejas Patil commented on SPARK-20313:
-------------------------------------
I tried to replicate what you shared in the jira but dont see anything wrong
with what planner is doing. Comparing both the approaches, `SortMergeJoin` is
always being picked. The second approach does joins over individual partitions
one by one and then unions the results. Depending on your data size + configs,
it might be possible that for your case a hash based join was used which would
explain why the later approach is faster.
Approach #1
{noformat}
val df1 = hc.sql("SELECT * FROM
bucketed_partitioned_1").filter(functions.col("ds").between("1", "5"))
val df2 = hc.sql("SELECT * FROM
bucketed_partitioned_2").filter(functions.col("ds").between("1", "5"))
val df3 = df1.join(df2, Seq("ds", "user_id")).explain(true)
== Physical Plan ==
*Project [ds#38, user_id#36, name#37, name#45]
+- *SortMergeJoin [ds#38, user_id#36], [ds#46, user_id#44], [ds#38,
user_id#36], [ds#46, user_id#44], Inner
:- *Sort [ds#38 ASC NULLS FIRST, user_id#36 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(ds#38, user_id#36, 200)
: +- *Filter isnotnull(user_id#36)
: +- HiveTableScan [user_id#36, name#37, ds#38], HiveTableRelation
`default`.`bucketed_partitioned_1`,
org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [user_id#36, name#37],
[ds#38], [isnotnull(ds#38), (ds#38 >= 1), (ds#38 <= 5)]
+- *Sort [ds#46 ASC NULLS FIRST, user_id#44 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(ds#46, user_id#44, 200)
+- *Filter isnotnull(user_id#44)
+- HiveTableScan [user_id#44, name#45, ds#46], HiveTableRelation
`default`.`bucketed_partitioned_2`,
org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [user_id#44, name#45],
[ds#46], [isnotnull(ds#46), (ds#46 >= 1), (ds#46 <= 5)]
{noformat}
Approach #2
{noformat}
val df1 = hc.sql("SELECT * FROM bucketed_partitioned_1")
val df2 = hc.sql("SELECT * FROM bucketed_partitioned_2")
val dsValues = Seq("1111-11-11", "4444-44-44")
val df3 = dsValues.map(dsValue => {
val df1filtered = df1.filter(functions.col("ds") === dsValue)
val df2filtered = df2.filter(functions.col("ds") === dsValue)
df1filtered.join(df2filtered, Seq("user_id")) // part1 removed from join
}).reduce(_ union _)
== Physical Plan ==
Union
:- *Project [user_id#63, name#64, ds#65, name#71, ds#72]
: +- *SortMergeJoin [user_id#63], [user_id#70], [user_id#63], [user_id#70],
Inner
: :- *Sort [user_id#63 ASC NULLS FIRST], false, 0
: : +- Exchange hashpartitioning(user_id#63, 200)
: : +- *Filter isnotnull(user_id#63)
: : +- HiveTableScan [user_id#63, name#64, ds#65], HiveTableRelation
`default`.`bucketed_partitioned_1`,
org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [user_id#63, name#64],
[ds#65], [isnotnull(ds#65), (ds#65 = 1111-11-11)]
: +- *Sort [user_id#70 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(user_id#70, 200)
: +- *Filter isnotnull(user_id#70)
: +- HiveTableScan [user_id#70, name#71, ds#72], HiveTableRelation
`default`.`bucketed_partitioned_2`,
org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [user_id#70, name#71],
[ds#72], [isnotnull(ds#72), (ds#72 = 1111-11-11)]
+- *Project [user_id#63, name#64, ds#65, name#71, ds#72]
+- *SortMergeJoin [user_id#63], [user_id#70], [user_id#63], [user_id#70],
Inner
:- *Sort [user_id#63 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(user_id#63, 200)
: +- *Filter isnotnull(user_id#63)
: +- HiveTableScan [user_id#63, name#64, ds#65], HiveTableRelation
`default`.`bucketed_partitioned_1`,
org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [user_id#63, name#64],
[ds#65], [isnotnull(ds#65), (ds#65 = 4444-44-44)]
+- *Sort [user_id#70 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(user_id#70, 200)
+- *Filter isnotnull(user_id#70)
+- HiveTableScan [user_id#70, name#71, ds#72], HiveTableRelation
`default`.`bucketed_partitioned_2`,
org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe, [user_id#70, name#71],
[ds#72], [isnotnull(ds#72), (ds#72 = 4444-44-44)]
{noformat}
> Possible lack of join optimization when partitions are in the join condition
> ----------------------------------------------------------------------------
>
> Key: SPARK-20313
> URL: https://issues.apache.org/jira/browse/SPARK-20313
> Project: Spark
> Issue Type: Improvement
> Components: Optimizer
> Affects Versions: 2.1.0
> Reporter: Albert Meltzer
>
> Given two tables T1 and T2, partitioned on column part1, the following have
> vastly different execution performance:
> // initial, slow
> {noformat}
> val df1 = // load data from T1
> .filter(functions.col("part1").between("val1", "val2")
> val df2 = // load data from T2
> .filter(functions.col("part1").between("val1", "val2")
> val df3 = df1.join(df2, Seq("part1", "col1"))
> {noformat}
> // manually optimized, considerably faster
> {noformat}
> val df1 = // load data from T1
> val df2 = // load data from T2
> val part1values = Seq(...) // a collection of values between val1 and val2
> val df3 = part1values
> .map(part1value => {
> val df1filtered = df1.filter(functions.col("part1") === part1value)
> val df2filtered = df2.filter(functions.col("part1") === part1value)
> df1filtered.join(df2filtered, Seq("col1")) // part1 removed from join
> })
> .reduce(_ union _)
> {noformat}
--
This message was sent by Atlassian JIRA
(v6.4.14#64029)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
