假设我想把三个表A,B,C和内部连接连接在一起,C是很小的。
#DUMMY EXAMPLE with IN-MEMORY table, but same issue if load table using spark.read.parquet("")
var A = (1 to 1000000).toSeq.toDF("A")
var B = (1 to 1000000).toSeq.toDF("B")
var C = (1 to 10).toSeq.toDF("C")我也无法控制接合的顺序:
CASE1 = A.join(B,expr("A=B"),"inner").join(C,expr("A=C"),"inner")
CASE2 = A.join(C,expr("A=C"),"inner").join(B,expr("A=B"),"inner")两种运行都显示CASE1运行速度比CASE2慢30-40%。
因此,问题是:如何利用星火的CBO自动将CASE1转换为内存中的表或从斯帕克的地板阅读器加载的表的CASE2?
我试过这样做:
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", -1)
spark.conf.set("spark.sql.cbo.enabled", "true")
A.createOrReplaceTempView("A")
spark.sql("ANALYZE TABLE A COMPUTE STATISTICS")但这会抛出:
org.apache.spark.sql.catalyst.analysis.NoSuchTableException: Table or view 'a' not found in database 'default'还有其他方法可以激活CBO而不必保存在蜂巢中的表吗?
附件:
CASE1.explain
== Physical Plan ==
*(5) SortMergeJoin [A#3], [C#13], Inner
:- *(3) SortMergeJoin [A#3], [B#8], Inner
: :- *(1) Sort [A#3 ASC NULLS FIRST], false, 0
: : +- Exchange hashpartitioning(A#3, 200)
: : +- LocalTableScan [A#3]
: +- *(2) Sort [B#8 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(B#8, 200)
: +- LocalTableScan [B#8]
+- *(4) Sort [C#13 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(C#13, 200)
+- LocalTableScan [C#13]CASE2.explain
== Physical Plan ==
*(5) SortMergeJoin [A#3], [B#8], Inner
:- *(3) SortMergeJoin [A#3], [C#13], Inner
: :- *(1) Sort [A#3 ASC NULLS FIRST], false, 0
: : +- Exchange hashpartitioning(A#3, 200)
: : +- LocalTableScan [A#3]
: +- *(2) Sort [C#13 ASC NULLS FIRST], false, 0
: +- Exchange hashpartitioning(C#13, 200)
: +- LocalTableScan [C#13]
+- *(4) Sort [B#8 ASC NULLS FIRST], false, 0
+- Exchange hashpartitioning(B#8, 200)
+- LocalTableScan [B#8]发布于 2019-03-18 16:41:40
不,简单的回答是这是不可能的。
这个https://databricks.com/blog/2017/08/31/cost-based-optimizer-in-apache-spark-2-2.html很好地概述了什么是可能的,以及持久化数据存储的要点。
https://stackoverflow.com/questions/55213505
复制相似问题