hadoop - 为什么分区连接(随机播放)并不总是比广播连接好？

Question

我已经进行了深入的研究，但我找不到足够详细的信息..我读过这些:1) http://www.cloudera.com/content/www/en-us/documentation/enterprise/latest/PDF/cloudera-impala.pdf2) http://www.cidrdb.org/cidr2015/Papers/CIDR15_Paper28.pdf

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

有人能解释一下为什么分区连接并不总是更好吗？我的意思是，如果我们有两个表 T1(大表)和 T2(小表)，如果我使用分区策略，它们都会被分区，并且我们将 T1/n-1 子集发送到其他节点，T2 也是如此.另一方面，如果我选择广播，Impala 将向其他 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                |
+-----------------------------------------------------------+

这是一个示例，其中加入策略是分区的:

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