Hive(六)--Join的使用

男娘i 2022-09-29 11:58 140阅读 0赞

**原理：**

笼统的说，Hive中的Join可分为Common Join（Reduce阶段完成join）和Map Join（Map阶段完成join）。本文简单介绍一下两种join的原理和机制。

## Hive Common Join ##

如果不指定MapJoin或者不符合MapJoin的条件，那么Hive解析器会将Join操作转换成Common Join,即：在Reduce阶段完成join.  
整个过程包含Map、Shuffle、Reduce阶段。

*  Map阶段

读取源表的数据，Map输出时候以Join on条件中的列为key，如果Join有多个关联键，则以这些关联键的组合作为key;  
Map输出的value为join之后所关心的(select或者where中需要用到的)列；同时在value中还会包含表的Tag信息，用于标明此value对应哪个表；  
按照key进行排序

*  Shuffle阶段

根据key的值进行hash,并将key/value按照hash值推送至不同的reduce中，这样确保两个表中相同的key位于同一个reduce中

*  Reduce阶段  
    根据key的值完成join操作，期间通过Tag来识别不同表中的数据。

以下面的HQL为例，图解其过程：

1.  SELECT 
2.  a.id,a.dept,b.age 
3.  FROM a join b 
4.  ON (a.id = b.id);

![Hive Common Join][]

看了这个图，应该知道如何使用MapReduce进行join操作了吧。

## Hive Map Join ##

MapJoin通常用于一个很小的表和一个大表进行join的场景，具体小表有多小，由参数**hive.mapjoin.smalltable.filesize**来决定，该参数表示小表的总大小，默认值为25000000字节，即25M。  
Hive0.7之前，需要使用hint提示 /\*+ mapjoin(table) \*/才会执行MapJoin,否则执行Common Join，但在0.7版本之后，默认自动会转换Map Join，由参数**hive.auto.convert.join**来控制，默认为true.  
仍然以9.1中的HQL来说吧，假设a表为一张大表，b为小表，并且hive.auto.convert.join=true,那么Hive在执行时候会自动转化为MapJoin。

![Hive MapJoin][]

*  如图中的流程，首先是Task A，它是一个Local Task（在客户端本地执行的Task），负责扫描小表b的数据，将其转换成一个HashTable的数据结构，并写入本地的文件中，之后将该文件加载到 DistributeCache中，该HashTable的数据结构可以抽象为：

<table style="border:1px solid silver; border-collapse:collapse; word-break:break-word; color:rgb(0,0,0); font-family:Verdana,Geneva,Arial,Helvetica,sans-serif; font-size:13px; height:113px; width:742px"> 
 <tbody> 
  <tr> 
   <td style="border:1px solid silver; border-collapse:collapse; padding:3px"> <span style="color:rgb(255,0,255)">key</span></td> 
   <td style="border:1px solid silver; border-collapse:collapse; padding:3px"> <span style="color:rgb(255,0,255)">value</span></td> 
  </tr> 
  <tr> 
   <td style="border:1px solid silver; border-collapse:collapse; padding:3px"> 1</td> 
   <td style="border:1px solid silver; border-collapse:collapse; padding:3px"> 26</td> 
  </tr> 
  <tr> 
   <td style="border:1px solid silver; border-collapse:collapse; padding:3px"> 2</td> 
   <td style="border:1px solid silver; border-collapse:collapse; padding:3px"> 34</td> 
  </tr> 
 </tbody> 
</table>

![MapReduce Local Task][]

图中红框圈出了执行Local Task的信息。

*   *  接下来是Task B，该任务是一个没有Reduce的MR，启动MapTasks扫描大表a,在Map阶段，根据a的每一条记录去和DistributeCache中b表对应的HashTable关联，并直接输出结果。
     *  由于MapJoin没有Reduce，所以由Map直接输出结果文件，有多少个Map Task，就有多少个结果文件。

**使用：**

[Hive][]表连接的语法支持如下：

Sql代码 ![收藏代码][icon_star.png]

1.  join\_table: 
2.   table\_reference JOIN table\_factor \[join\_condition\] 
3.   | table\_reference \{ LEFT|RIGHT|**FULL**\} \[OUTER\] JOIN table\_reference join\_condition 
4.   | table\_reference LEFT SEMI JOIN table\_reference join\_condition 
5.   | table\_reference CROSS JOIN table\_reference \[join\_condition\] (**as** **of** Hive 0.10) 
6.  
7.  table\_reference: 
8.   table\_factor 
9.   | join\_table 
10. 
11. table\_factor: 
12.  tbl\_name \[alias\] 
13.  | table\_subquery alias 
14.  | ( table\_references ) 
15. 
16. join\_condition: 
17. **ON** equality\_expression ( AND equality\_expression )\* 
18. 
19. equality\_expression: 
20.  expression = expression

hive只支持等连接，外连接，左半连接。hive不支持非相等的join条件（通过其他方式实现，如left outer join），因为它很难在map/reduce job实现这样的条件。而且，hive可以join两个以上的表。

**例子**

写join查询时，有几个典型的点要考虑，如下：

**等连接**

只有等连接才允许

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.\* **FROM** a JOIN b **ON** (a.id = b.id) 
2.  **SELECT** a.\* **FROM** a JOIN b **ON** (a.id = b.id AND a.department = b.department)

这两个是合法的连接

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.\* **FROM** a JOIN b **ON** (a.id <> b.id)

这个是不允许的。

**多表连接**

同个查询，可以join两个以上的表

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.val, b.val, c.val **FROM** a JOIN b **ON** (a.**key** = b.key1) JOIN c **ON** (c.**key** = b.key2)

**join的缓存和任务转换**

hive转换多表join时，如果每个表在join字句中，使用的都是同一个列，只会转换为一个单独的map/reduce。

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.val, b.val, c.val **FROM** a JOIN b **ON** (a.**key** = b.key1) JOIN c **ON** (c.**key** = b.key1)

这个会转换为单独的map/reduce任务，只有b表的key1列在join被调用。

另一方面

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.val, b.val, c.val **FROM** a JOIN b **ON** (a.**key** = b.key1) JOIN c **ON** (c.**key** = b.key2)

被转换为两个map/reduce任务，因为b的key1列在第一个join条件使用，而b表的key2列在第二个join条件使用。第一个map/reduce任务join a和b。第二个任务是第一个任务的结果join c。

在join的每个map/reduce阶段，序列中的最后一个表，当其他被缓存时，它会流到reducers。所以，reducers需要缓存join关键字的特定值组成的行，通过组织最大的表出现在序列的最后，有助于减少reducers的内存。

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.val, b.val, c.val **FROM** a JOIN b **ON** (a.**key** = b.key1) JOIN c **ON** (c.**key** = b.key1)

三个表，在同一个独立的map/reduce任务做join。a和b的key对应的特定值组成的行，会缓存在reducers的内存。然后reducers接受c的每一行，和缓存的每一行做join计算。

类似

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.val, b.val, c.val **FROM** a JOIN b **ON** (a.**key** = b.key1) JOIN c **ON** (c.**key** = b.key2)

这里有两个map/reduce任务在join计算被调用。第一个是a和b做join，然后reducers缓存a的值，另一边，从流接收b的值。第二个阶段，reducers缓存第一个join的结果，另一边从流接收c的值。

在join的每个map/reduce阶段，通过关键字，可以指定哪个表从流接收。

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** /\*+ STREAMTABLE(a) \*/ a.val, b.val, c.val **FROM** a JOIN b **ON** (a.**key** = b.key1) JOIN c **ON** (c.**key** = b.key1)

三个表的连接，会转换为一个map/reduce任务，reducer会把b和c的key的特定值缓存在内存里，然后从流接收a的每一行，和缓存的行做join。

**join的结果**

LEFT，RIGHT，FULL OUTER连接存在是为了提供ON语句在没有匹配时的更多控制。例如，这个查询：

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.val, b.val **FROM** a LEFT OUTER JOIN b **ON** (a.**key**=b.**key**)

将会返回a的每一行。如果b.key等于a.key,输出将是a.val,b.val,如果a没有和b.key匹配，输出的行将是 a.val,NULL。如果b的行没有和a.key匹配上，将被抛弃。语法"FROM a LEFT OUTER JOIN b"必须写在一行，为了理解它如何工作——这个查询，a是b的左边，a的所有行会被保持；RIGHT OUTER JOIN将保持b的所有行， FULL OUTER JOIN将会保存a和b的所有行。OUTER JOIN语义应该符合标准的SQL规范。

**join的过滤**

Joins发生在where字句前，所以，如果要限制join的输出，需要写在where字句，否则写在JOIN字句。现在讨论的一个混乱的大点，就是分区表

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.val, b.val **FROM** a LEFT OUTER JOIN b **ON** (a.**key**=b.**key**) 
2.  **WHERE** a.ds='2009-07-07' AND b.ds='2009-07-07'

将会连接a和b，产生a.val和b.val的列表。WHERE字句，也可以引用join的输出列，然后过滤他们。

但是，无论何时JOIN的行找到a的key，但是找不到b的key时，b的所有列会置成NULL，包括ds列。这就是说，将过滤join输出的所有行，包括没有合法的b.key的行。然后你会在LEFT OUTER的要求扑空。

也就是说，如果你在WHERE字句引用b的任何列，LEFT OUTER的部分join结果是不相关的。所以，当外连接时，使用这个语句

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.val, b.val **FROM** a LEFT OUTER JOIN b **ON** (a.**key**=b.**key** AND b.ds='2009-07-07' AND a.ds='2009-07-07'

join的输出会预先过滤，然后你不用对有a.key而没有b.key的行做过滤。RIGHT和FULL join也是一样的逻辑。

**join的顺序**

join是不可替换的，连接是从左到右，不管是LEFT或RIGHT join。

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.val1, a.val2, b.val, c.val 
2.  **FROM** a 
3.  JOIN b **ON** (a.**key** = b.**key**) 
4.  LEFT OUTER JOIN c **ON** (a.**key** = c.**key**)

首先，连接a和b，扔掉a和b中没有匹配的key的行。结果表再连接c。这提供了直观的结果，如果有一个键都存在于A和C，但不是B：完整行(包括 a.val1,a.val2,a.key)会在"a jOIN b"步骤，被丢弃，因为它不在b中。结果没有a.key，所以当它和c做LEFT OUTER JOIN,c.val也无法做到，因为没有c.key匹配a.key(因为a的行都被移除了)。类似的，RIGHT OUTER JOIN(替换为LEFT),我们最终会更怪的效果,NULL, NULL, NULL, c.val。因为尽管指定了join key是a.key=c.key，我们已经在第一个JOIN丢弃了不匹配的a的所有行。

为了达到更直观的效果，相反，我们应该从

Sql代码 ![收藏代码][icon_star.png]

1.  **FROM** c LEFT OUTER JOIN a **ON** (c.**key** = a.**key**) LEFT OUTER JOIN b **ON** (c.**key** = b.**key**).

LEFT SEMI JOIN实现了相关的IN / EXISTS的子查询语义的有效途径。由于Hive目前不支持IN / EXISTS的子查询，所以你可以用 LEFT SEMI JOIN 重写你的子查询语句。LEFT SEMI JOIN 的限制是， JOIN 子句中右边的表只能在 ON 子句中设置过滤条件，在 WHERE 子句、SELECT 子句或其他地方过滤都不行。

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.**key**, a.value **FROM** a **WHERE** a.**key** in (**SELECT** b.**key** **FROM** B);

可以重写为

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** a.**key**, a.val **FROM** a LEFT SEMI JOIN b **on** (a.**key** = b.**key**)

**mapjoin**

但如果所有被连接的表是小表，join可以被转换为只有一个map任务。查询是

SELECT /*+ MAPJOIN(b) */ a.key, a.value
    FROM a join b on a.key = b.key

不需要reducer。对于每一个mapper,A和B已经被完全读出。限制是a FULL/RIGHT OUTER JOIN b不能使用。

如果表在join的列已经分桶了，其中一张表的桶的数量，是另一个表的桶的数量的整倍，那么两者可以做桶的连接。如果A有4个桶，表B有4个桶，下面的连接：

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** /\*+ MAPJOIN(b) \*/ a.**key**, a.value **FROM** a join b **on** a.**key** = b.**key**

只能在mapper工作。为了为A的每个mapper完整抽取B。对于上面的查询，mapper处理A的桶1，只会抽取B的桶1，这不是默认行为，要使用以下参数：

Sql代码 ![收藏代码][icon_star.png]

1.  **set** hive.optimize.bucketmapjoin = **true**;

如果表在join的列经过排序，分桶，而且他们有相同数量的桶，可以使用排序-合并 join。每个mapper，相关的桶会做连接。如果A和B有4个桶，

Sql代码 ![收藏代码][icon_star.png]

1.  **SELECT** /\*+ MAPJOIN(b) \*/ a.**key**, a.value **FROM** A a join B b **on** a.**key** = b.**key**

只能在mapper使用。使用A的桶的mapper，也会遍历B相关的桶。这个不是默认行为，需要配置以下参数：

Sql代码 ![收藏代码][icon_star.png]

1.  **set** hive.input.format=org.apache.hadoop.hive.ql.io.BucketizedHiveInputFormat; 
2.  **set** hive.optimize.bucketmapjoin = **true**; 
3.  **set** hive.optimize.bucketmapjoin.sortedmerge = **true**;

[Hive Common Join]: http://7xipth.com1.z0.glb.clouddn.com/0625-1.jpg
[Hive MapJoin]: http://7xipth.com1.z0.glb.clouddn.com/0625-3.jpg
[MapReduce Local Task]: http://7xipth.com1.z0.glb.clouddn.com/0625-2.jpg
[Hive]: http://lib.csdn.net/base/hive
[icon_star.png]: /images/20220708/736a838d86a1487e875feaa9b3ffa942.png