Apache Spark：广播连接行为：过滤连接表和临时表

Question

我需要在火花中连接2个表。但是我首先过滤掉第二个表的一部分，而不是完全连接2个表：

spark.sql("select * from a join b on a.key=b.key where b.value='xxx' ")

我想在这种情况下使用广播连接。

Spark有一个参数，用于定义广播连接的最大表大小：spark.sql.autoBroadcastJoinThreshold：

配置在执行联接时将广播到所有工作节点的表的最大大小（以字节为单位）。通过将该值设置为-1，可以禁用广播。请注意，当前统计信息仅支持已运行命令ANALYZE TABLE COMPUTE statistics noscan的配置单元元存储表。http://spark.apache.org/docs/2.4.0/sql-performance-tuning.html

关于此设置，我有以下问题：

1. spark会将哪个表格大小与autoBroadcastJoinThreshold的值进行比较：完整大小，还是应用where子句后的大小

Answer 1

我继续做了一些小实验来回答你的第一个问题。

问题一：

• 创建了一个包含3行[key，df_a_column]的数据帧a
• 创建了一个包含10行[键，值]的数据帧b
• ran：spark.sql（"SELECT*from a JOIN b ONa.key=b.key"）

== Physical Plan ==
*(1) BroadcastHashJoin [key#122], [key#111], Inner, BuildLeft, false
:- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [id=#168]
:  +- LocalTableScan [key#122, df_a_column#123]
+- *(1) LocalTableScan [key#111, value#112]

正如所料，广播的是3行较小的dfa。

• 运行：spark.sql（"SELECT*from a JOIN b ONa.key=b.key其中b.value=\"bat\"）。

== Physical Plan ==
*(1) BroadcastHashJoin [key#122], [key#111], Inner, BuildRight, false
:- *(1) LocalTableScan [key#122, df_a_column#123]
+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [id=#152]
   +- LocalTableScan [key#111, value#112]

在这里，您可以注意到数据帧b正在广播！Means spark在应用where选择要播放的内容后，会评估大小。

问题2：

是的，你是对的。从以前的输出中可以明显看出，它首先适用于哪里。

问题3：不，你不能分析，但是你可以通过暗示火花来广播临时视图表，即使在SQL。参考

如果你现在看到解释：

== Physical Plan ==
*(1) BroadcastHashJoin [key#122], [key#111], Inner, BuildRight, false
:- *(1) LocalTableScan [key#122, df_a_column#123]
+- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)),false), [id=#184]
   +- LocalTableScan [key#111, value#112]

现在，如果您看到，dataframeb是广播的，即使它有10行。在问题1中，没有提示，a是广播的。

注意：广播提示在SQL火花可用于2.2

了解物理计划的提示：

• 从LocalTableScan[列列表]
中找出数据帧
• BroadcastExchange的子树/列表下的数据帧正在广播。

Answer 2

对选项2的理解是正确的。不能在spark中分析临时表。读这里

如果您想带头指定要广播的数据帧，可以使用下面的代码段-

df = df1.join(F.broadcast(df2),df1.some_col == df2.some_col, "left")