Hive基础二（Join原理和机制，Join的几种类型，数据倾斜简单处理）
Hive基础二（Join原理和机制，Join的几种类型，数据倾斜简单处理）
Hive是基于Hadoop的一个数据仓库工具，可以查询和管理PB级别的数据。在Hive中，Join操作是一个非常重要的功能，它允许我们从多个表中获取数据并进行关联分析。本文将详细介绍Hive中Join的基本原理、各种Join类型以及Join操作中需要注意的事项，并提供一些处理数据倾斜问题的方法。
一、Hive中Join的原理和机制
Hive中的Join可分为Common Join（Reduce阶段完成join）和Map Join（Map阶段完成join）。
1. Hive Common Join
如果不指定MapJoin或者不符合MapJoin的条件，那么Hive解析器会默认把执行Common Join，即在Reduce阶段完成join。整个过程包含Map、Shuffle、Reduce阶段。
Map阶段
读取源表的数据，Map输出时候以Join on条件中的列为key，如果Join有多个关联键，则以这些关联键的组合作为key；Map输出的value为join之后所关心的(select或者where中需要用到的)列，同时在value中还会包含表的Tag信息，用于标明此value对应哪个表。Shuffle阶段
根据key的值进行hash，并将key/value按照hash值推送至不同的reduce中，这样确保两个表中相同的key位于同一个reduce中。Reduce阶段
根据key的值完成join操作，期间通过Tag来识别不同表中的数据。
以下面的HQL为例，图解其过程：
SELECT a.id,a.dept,b.age
FROM a join b
ON (a.id = b.id);
2. Hive Map Join
MapJoin通常用于一个很小的表和一个大表进行join的场景，具体小表有多小，由参数hive.mapjoin.smalltable.filesize来决定，默认值为25M。满足条件的话Hive在执行时候会自动转化为MapJoin，或使用hint提示 /*+ mapjoin(table) */执行MapJoin。
如上图中的流程，首先Task A在客户端本地执行，负责扫描小表b的数据，将其转换成一个HashTable的数据结构，并写入本地的文件中，之后将该文件加载到DistributeCache中。
接下来的Task B任务是一个没有Reduce的MapReduce，启动MapTasks扫描大表a，在Map阶段，根据a的每一条记录去和DistributeCache中b表对应的HashTable关联，并直接输出结果，因为没有Reduce，所以有多少个Map Task，就有多少个结果文件。
注意：Map JOIN不适合FULL/RIGHT OUTER JOIN。
二、Join的几种类型
Hive中除了支持和传统数据库中一样的内关联（JOIN）、左关联（LEFT JOIN）、右关联（RIGHT JOIN）、全关联（FULL JOIN），还支持左半关联（LEFT SEMI JOIN）。
注意：Hive中Join的关联键必须在ON()中指定，不能在Where中指定。
数据准备：
hive> desc lxw1234_a;
OK
id string
name string
Time taken: 0.094 seconds, Fetched: 2 row(s)
hive> select * from lxw1234_a;
OK
1 zhangsan
2 lisi
3 wangwu
Time taken: 0.116 seconds, Fetched: 3 row(s)
hive> desc lxw1234_b;
OK
id string
age int
Time taken: 0.159 seconds, Fetched: 2 row(s)
hive> select * from lxw1234_b;
OK
1 30
2 29
4 21
Time taken: 0.09 seconds, Fetched: 3 row(s)
1. 内关联（JOIN）
常规join，类似交集，只显示关联成功的行。
SELECT a.id, a.name, b.age
FROM lxw1234_a a
join lxw1234_b b
ON (a.id = b.id);
--执行结果
1 zhangsan 30
2 lisi 29
2. 左外关联（LEFT [OUTER] JOIN）
以LEFT [OUTER] JOIN关键字前面的表作为主表，和其他表进行关联，返回记录和主表的记录数一致，关联不上的字段置为NULL。
是否指定OUTER关键字，貌似对查询结果无影响。
SELECT a.id, a.name, b.age
FROM lxw1234_a a
left join lxw1234_b b
ON (a.id = b.id);
--执行结果：
1 zhangsan 30
2 lisi 29
3 wangwu NULL
3. 右外关联（RIGHT [OUTER] JOIN）
和左外关联相反，以RIGTH [OUTER] JOIN关键词后面的表作为主表，和前面的表做关联，返回记录数和主表一致，关联不上的字段为NULL。
是否指定OUTER关键字，貌似对查询结果无影响。
SELECT a.id, a.name, b.age
FROM lxw1234_a a
RIGHT OUTER JOIN lxw1234_b b
ON (a.id = b.id);
--执行结果：
1 zhangsan 30
2 lisi 29
NULL NULL 21
4. 全外关联（FULL [OUTER] JOIN）
以两个表的记录为基准，返回两个表的所有记录，类似并集，关联不上的字段为NULL。
是否指定OUTER关键字，貌似对查询结果无影响。
SELECT a.id, a.name, b.age
FROM lxw1234_a a
FULL OUTER JOIN lxw1234_b b
ON (a.id = b.id);
--执行结果：
1 zhangsan 30
2 lisi 29
3 wangwu NULL
NULL NULL 21
5. 左半连接（LEFT SEMI JOIN）
以LEFT SEMI JOIN关键字前面的表为主表，返回主表的KEY也在副表中的记录。相当于IN或EXISTS的作用。左半连接的限制是右侧的表只能出现在ON子句中，不能出现在WHERE或者SELECT子句中。
注意，left semi join和join并不是等价的，比如当左表1条数据可以关联右表2条数据时，此时left semi join只显示1条数据（因为只要左表的key在右表存在就行，不用管几条），join却显示两条数据（关联到的都显示），好好体会一下。
SELECT a.id, a.name
FROM lxw1234_a a
LEFT SEMI JOIN lxw1234_b b
ON (a.id = b.id);
--执行结果：
1 zhangsan
2 lisi
--等价于：
SELECT a.id, a.name
FROM lxw1234_a a
WHERE a.id IN (SELECT id FROM lxw1234_b);
三、Join语句需要注意的地方
- 首先是Hive中的连接查询只支持相等连接而不支持不等连接查询：
//有效的连接查询，相等连接查询
SELECT a.* FROM a JOIN b ON (a.id = b.id AND a.department = b.department)
//无效的连接查询，Hive不支持不等连接查询
SELECT a.* FROM a JOIN b ON (a.id <> b.id)
- 如果每个表都只使用相同的列join连接，Hive将只生成一个map/reduce作业；如果一个表使用了两个以上的字段，则会生成2个以上的mr任务：
//由于join子句中只使用了表b的key1列，该查询转换为一个作业
SELECT a.val, b.val, c.val FROM a JOIN b ON (a.key = b.key1) JOIN c ON (c.key = b.key1)
//由于表b的key1列用在第一个join子句中，key2列用在第二个join子句中，该查询被转换为两个作业，
//第一个作业执行表a和b的连接查询，第二个作业将第一个作业的结果与第二个join子句进行连接查询
SELECT a.val, b.val, c.val FROM a JOIN b ON (a.key = b.key1) JOIN c ON (c.key = b.key2)
- Join连接的顺序是不可以交换的，无论是LEFT还是RIGHT连接都是左结合的。
四、Hive数据倾斜的简单处理
数据倾斜症状：
任务长时间维持在99%（或100%）;
查看任务监控页面，发现只有少量（1个或几个）reduce子任务未完成；
本地读写数据量很大。
原因：
key分布不均匀；
业务数据本身特点。
导致数据倾斜的操作：
GROUP BY, COUNT DISTINCT()，join
这里列出几个常用解决办法：
参数hive.groupby.skewindata = true
似乎是解决数据倾斜的万能钥匙，查询计划会有两个 MR Job，第一个MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个Reduce做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce中，从而达到负载均衡的目的；第二个MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作。使用COUNT DISTINCT造成的数据倾斜
- 如果某一个值的记录特别多，可以先把该值过滤掉，在最后单独处理。比如某一天的IMEI值为’lxw1234’的特别多，当我要统计总的IMEI数，可以先统计不为’lxw1234’的，之后再加1。
SELECT CAST(COUNT(DISTINCT imei)+1 AS bigint)
FROM lxw1234 where pt = '2012-05-28'
AND imei <> 'lxw1234' ;
- 数据量大的情况下，由于COUNT DISTINCT操作需要用一个Reduce Task来完成，这一个Reduce需要处理的数据量太大，就会导致整个Job很难完成，一般COUNT DISTINCT使用先GROUP BY再COUNT的方式替换：
SELECT day, COUNT(DISTINCT id) AS uv
FROM lxw1234
GROUP BY day;
--可以转换成：
SELECT day, COUNT(id) AS uv
FROM (SELECT day,id FROM lxw1234 GROUP BY day,id) a
GROUP BY day;
虽然会多用一个Job来完成，但在数据量大的情况下，这个绝对是值得的。
- 使用JOIN引起的数据倾斜
表连接顺序优化：多表连接时，尽量小表在前，大表在后。因为JOIN前一阶段生成的数据会存在于Reducer的buffer中，小表数据量小，内存开销就小，与后面的大表进行连接时，只需要从buffer中读取缓存的Key，与大表中的指定Key进行连接，速度会更快，也可能避免内存缓冲区溢出。
where优化：Join出现在WHERR子句之前。因此如果想在join前进行过滤，限制条件应该出现在JOIN子句中，而不是where中：
SELECT a.val, b.val FROM a LEFT OUTER JOIN b
ON (a.key=b.key)
WHERE a.ds='2009-07-07' AND b.ds='2009-07-07';
//当该左外连接在a中发现key而在b中没有发现key时，b中的列将为null，包括分区列ds，后边的where就没有用了。
//下面的语句将会提前根据条件过滤：
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');
- 关联键存在大量空值：在SQL标准中，任何对NULL的操作(如数值比较，字符串操作等)结果都为NULL。Hive对NULL值的处理与其基本一致，除了JOIN时的特殊逻辑：Hive的JOIN中作为JOIN Key的字段比较，NULL=NULL是有意义的，且返回值为True。
所以需要改写查询手动过滤NULL值的情况，操作如下：
SELECT user.uid,count(user.uid) FROM class JOIN user
ON(class.uid = user.uid and class.uid IS NOT NULL and user.uid IS NOT NULL)
GROUP BY user.uid;
- 不同数据类型的字段关联：转换为同一数据类型之后再做关联。