为什么分区连接（shuffle）并不总是比广播连接好？

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

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

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

这是一个连接策略为广播的示例计划：

[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                |
+-----------------------------------------------------------+

这是一个连接策略被分区的示例：

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）。