为什么分区连接（随机播放）并不总是比广播连接更好？

Question

我做了深入的研究，但我找不到足够详细的内容。我读过这些：1）http://www.cloudera.com/content/www/en-us/documentation/enterprise/latest/PDF /cloudera-impala.pdf 2）http://www.cidrdb.org/cidr2015/Papers/CIDR15_Paper28.pdf

但我没有找到任何答案..

有人可以解释一下为什么分区连接并不总是更好吗？我的意思是，如果我们有两个表 T1（大表）和 T2（小表），如果我使用分区策略，它们都会被分区，并且我们将 T1/n-1 子集发送到其他节点，T2 也是如此。另一方面，如果我选择广播，Impala 会将 T2*n-1 的数据发送给其他人。

也许我不明白这些策略是如何运作的..如果我错了，有人可以解释一下吗？也许用简单的画法？（我已经在谷歌图片上搜索过了..）

提前致谢

Answer 1

分区不是免费的，构建和探测（左侧和右侧）侧都需要分区才能进行分区连接。每个分区都需要一个交换计划片段作为子分区，并且每个分区都会产生网络传输。然而，如果构建端很小，那么每个节点都可以拥有它的副本（即广播），然后使用未分区的左侧探测构建端哈希表，而无需在探测端引入额外的子交换。事实上，广播所需的交换特别昂贵，因为每个发送者需要发送给 N 个接收者。

什么是“足够小”来执行广播连接？这取决于许多因素，但最明显和最重要的是构建端哈希表应该适合内存。

以下是加入策略为 BROADCAST 的示例计划：

[localhost:21000] > explain select * from alltypes t1 join alltypessmall t2 on t1.id = t2.id;
Query: explain select * from alltypes t1 join alltypessmall t2 on t1.id = t2.id
+-----------------------------------------------------------+
| Explain String                                            |
+-----------------------------------------------------------+
| Estimated Per-Host Requirements: Memory=160.01MB VCores=2 |
|                                                           |
| 04:EXCHANGE [UNPARTITIONED]                               |
| |                                                         |
| 02:HASH JOIN [INNER JOIN, BROADCAST]                      |
| |  hash predicates: t1.id = t2.id                         |
| |                                                         |
| |--03:EXCHANGE [BROADCAST]                                |
| |  |                                                      |
| |  01:SCAN HDFS [functional.alltypessmall t2]             |
| |     partitions=4/4 files=4 size=6.32KB                  |
| |                                                         |
| 00:SCAN HDFS [functional.alltypes t1]                     |
|    partitions=24/24 files=24 size=478.45KB                |
+-----------------------------------------------------------+

这是一个连接策略为 PARTITIONED 的示例：

Query: explain select * from tpch.lineitem t1 join tpch.lineitem t2 on t1.l_orderkey = t2.l_orderkey
+-----------------------------------------------------------+
| Explain String                                            |
+-----------------------------------------------------------+
| Estimated Per-Host Requirements: Memory=815.44MB VCores=2 |
|                                                           |
| 05:EXCHANGE [UNPARTITIONED]                               |
| |                                                         |
| 02:HASH JOIN [INNER JOIN, PARTITIONED]                    |
| |  hash predicates: t1.l_orderkey = t2.l_orderkey         |
| |                                                         |
| |--04:EXCHANGE [HASH(t2.l_orderkey)]                      |
| |  |                                                      |
| |  01:SCAN HDFS [tpch.lineitem t2]                        |
| |     partitions=1/1 files=1 size=718.94MB                |
| |                                                         |
| 03:EXCHANGE [HASH(t1.l_orderkey)]                         |
| |                                                         |
| 00:SCAN HDFS [tpch.lineitem t1]                           |
|    partitions=1/1 files=1 size=718.94MB                   |
+-----------------------------------------------------------+
Fetched 16 row(s) in 0.03s

请注意，后一个计划有一个额外的交换。这意味着扫描有一个额外的计划片段（id 00）。