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

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

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

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

也许我不明白这些策略是如何运作的..如果我错了，有人能解释一下吗？也许用一个简单的平局？（我已经搜索了 google 张图片..）

提前致谢

Answer 1

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

什么是 "small enough" 执行广播加入？这取决于许多因素，但最明显和最重要的是构建端哈希 table 应该适合内存。

这是加入策略为 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                |
+-----------------------------------------------------------+

这是一个示例，其中连接策略是分区的：

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

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

Why partitioned join (shuffle) isn't always better than broadcast join?

hadoop

hive

hdfs

cloudera

impala