Hive知识点

hive介绍

Hive通常意义上来说，是把一个SQL转化成一个分布式作业，如MapReduce，Spark或者Tez。无论Hive的底层执行框架是MapReduce、Spark还是Tez，其原理基本都类似。

而目前，由于MapReduce稳定，容错性好，大量数据情况下使用磁盘，能处理的数据量大，所以目前Hive的主流执行框架是MapReduce，但性能相比Spark和Tez也就较低。

Hive的基本组成

用户接口：包括 CLI、JDBC/ODBC、WebGUI。
CLI：command line interface，命令行接口。
Hive WEB Interface（HWI）：hive客户端提供了一种通过网页的方式访问hive所提供的服务。
ThriftServers：提供JDBC和ODBC接入的能力,它用来进行可扩展且跨语言的服务的开发，hive集成了该服务，能让不同的编程语言调用hive的接口。
元数据存储：通常是存储在关系数据库如 mysql , derby中。
语句转换：解释器、编译器、优化器、执行器。

基本组成
查询语言。由于 SQL 被广泛的应用在数据仓库中，因此，专门针对 Hive 的特性设计了类 SQL 的查询语言 HQL。熟悉 SQL 开发的开发者可以很方便的使用 Hive 进行开发。
数据存储位置。Hive 是建立在Hadoop 之上的，所有 Hive 的数据都是存储在HDFS 中的。
数据格式。Hive 中没有定义专门的数据格式，数据格式可以由用户指定，用户定义数据格式需要指定三个属性：列分隔符（通常为空格、”\t”、”\x001″）、行分隔符（”\n”）以及读取文件数据的方法（Hive 中默认有三个文件格式 TextFile，SequenceFile 以及 RCFile）。由于在加载数据的过程中，不需要从用户数据格式到 Hive 定义的数据格式的转换，因此，Hive 在加载的过程中不会对数据本身进行任何修改，而只是将数据内容复制或者移动到相应的 HDFS 目录中。
数据更新。由于 Hive 是针对数据仓库应用设计的，而数据仓库的内容是读多写少的。因此，Hive 中不支持对数据的改写和添加，所有的数据都是在加载的时候中确定好的。
索引。之前已经说过，Hive 在加载数据的过程中不会对数据进行任何处理，甚至不会对数据进行扫描，因此也没有对数据中的某些 Key 建立索引。Hive 要访问数据中满足条件的特定值时，需要暴力扫描整个数据，因此访问延迟较高。由于 MapReduce 的引入， Hive 可以并行访问数据，因此即使没有索引，对于大数据量的访问，Hive 仍然可以体现出优势。

hive的四种表类型

内部表(受控表)，就是一般的表，前面讲到的表都是内部表，当表定义被删除的时候，表中的数据随之一并被删除。
外部表，数据存在与否和表的定义互不约束，仅仅只是表对hdfs上相应文件的一个引用，当删除表定义的时候，表中的数据依然存在。

创建外部表，external是外部表的关键字，也是和内部表有区别的地方
create external table tblName(colName colType...);
加载数据
alter table tblName set location 'hdfs_absolute_uri';

外部表还可以在创建表的时候指定数据的位置，引用当前位置的数据。
create external table tblName(colName colType...) location 'hdfs_absolute_uri';

内部表和外部表的转换：
内——>外
alter table tblName set tblproperties('EXTERNAL'='TRUE');
外——>内
alter table tblName set tblproperties('EXTERNAL'='FALSE');

分区表，表对应一个目录，分区也对应一个目录，分区中的数据对应文件。
如何创建一张分区表？只需要在之前的创建表后面使用partition by加上分区字段就可以了，eg.

create table tblName (
　　 id int comment 'ID',
　　 name string comment 'name' 
　　) partitioned by (dt date comment 'create time')
row format delimited
fields terminated by '\t';

向分区表中插入数据(要指定分区)
load data local inpath linux_fs_path into table tblName partition(dt='2015-12-12');

多个分区如何创建？
和单分区表的创建类似：

create table tblName (
　　 id int comment 'ID',
　　 name string comment 'name' 
　　) partitioned by (year int comment 'admission year', school string comment 'school name')
　　row format delimited
　　fields terminated by '\t';

同时也可以从hdfs上引用数据：
alter table tblName partition(year='2015', school='crxy') set 'hdfs_absolute_uri';
注意：
必须得现有分区,必须要使用hdfs绝对路径。

桶表
表对应目录，桶对应文件，桶表是对存入的数据按照某个字段进行哈希取值，值相同的放到同一个文件存储。对于每一个表或者是分区，Hive可以进一步组织成桶，也就是说桶是更为细粒度的数据范围划分。Hive是针对某一列进行分桶。Hive采用对列值哈希，然后除以桶的个数求余的方式决定该条记录存放在哪个桶中。分桶的好处是可以获得更高的查询处理效率。使取样更高效。

创建 emp_bucket 表，字段如下，按照 job 字段分成四个桶

create table bucketed_user(
  id int,
  name string
)
clustered by(id) sorted by(name) into 4 buckets
row format delimited fields terminated by '\t'
stored as textfile;

我们使用用户id来确定如何划分桶(Hive使用对值进行哈希并将结果除于桶的个数取余数的方式进行分桶)

如何加载数据？不能使用load data这种方式，需要从别的表来引用
insert into bucketed_user select * from user;
注意:在插入数据之前需要先设置开启桶操作，不然插入数据不会设置为桶!

set hive.enforce.bucketing=true;

当从桶表中进行查询时，hive会根据分桶的字段进行计算分析出数据存放的桶中，然后直接到对应的桶中去取数据，这样做就很好的提高了效率。

Hive 表默认的仓库路径：user/hive/warehouse/tablename

Group By和Join

首先是Group By

例如我们有一条SQL语句：

INSERT INTO TABLE pageid_age_sum 
SELECT pageid, age, count(1) 
FROM pv_users 
GROUP BY pageid, age;

把每个网页的阅读数按年龄进行分组统计。由于前面介绍了，MapReduce就是一个Group By的过程，这个SQL翻译成MapReduce就是相对简单的。

我们在Map端，每一个Map读取一部分表的数据，通常是64M或者128M，然后按需要Group By的Key分发到Reduce端。经过Shuffle Sort，每一个Key再在Reduce端进行聚合（这里是Count)，然后就输出了最终的结果。值得一提的是，Distinct在实现原理上与Group By类似。当Group By遇上 Distinct……例如：

SELECT pageid, COUNT(DISTINCT userid) FROM page_view GROUP BY pageid

Hive 实现成MapReduce的原理如下：

也就是说Map分发到Reduce的时候，会使用pageid和userid作为联合分发键，再去聚合（Count），输出结果。

介绍了这么多原理，重点还是为了使用，为了适应场景和业务，为了优化。从原理上可以看出，当遇到Group By的查询时，会按Group By 键进行分发？如果键很多，撑爆了机器会怎么样？

对于Impala，或Spark，为了快，key在内存中，爆是经常的。爆了就失败了。对于Hive，Key在硬盘，本身就比Impala, Spark的处理能力大上几万倍。但……不幸的是，硬盘也有可能爆。

当然，硬盘速度也比内存慢上不少，这也是Hive总是被吐槽的原因，场景不同，要明白自己使用的场景。当Group By Key大到连硬盘都能撑爆时……这个时候可能就需要优化了。

Ｇroup By优化通常有Map端数据聚合和倾斜数据分发两种方式。Map端部分聚合，配置开关是hive.map.aggr

也就是执行SQL前先执行 set hive.map.aggr=true;它的原理是Map端在发到Reduce端之前先部分聚合一下。来减少数据量。因为我们刚才已经知道，聚合操作是在Reduce端完成的，只要能有效的减少Reduce端收到的数据量，就能有效的优化聚合速度，避免爆机，快速拿到结果。

另外一种方式则是针对倾斜的key做两道作业的聚合。什么是倾斜的数据？比如某猫双11交易，华为卖了1亿台，苹果卖了10万台。华为就是典型的倾斜数据了。如果要统计华为和苹果，会用两个Reduce作Group By，一个处理1亿台，一个处理10万台，那个1亿台的就是倾余。

由于按key分发，遇到倾斜数据怎么办？

可以使用hive.groupby.skewindata选项，通过两道MapReduce作业来处理。当选项设定为 true，生成的查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照 Group ByKey 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce中），最后完成最终的聚合操作。

第一道作业：Map随机分发，按gby key部分聚合
第二道作业：第一道作业结果Map倾斜的key分发，按gbk key进行最终聚合

无论你使用Map端，或者两道作业。其原理都是通过部分聚合来来减少数据量。能不能部分聚合，部分聚合能不能有效减少数据量，通常与UDAF，也就是聚合函数有关。也就是只对代数聚合函数有效，对整体聚合函数无效。

所谓代数聚合函数，就是由部分结果可以汇总出整体结果的函数，如count，sum。所谓整体聚合函数，就是无法由部分结果汇总出整体结果的函数，如avg，mean。比如，sum, count，知道部分结果可以加和得到最终结果。而对于，mean，avg，知道部分数据的中位数或者平均数，是求不出整体数据的中位数和平均数的。

在遇到复杂逻辑的时候，还是要具体问题具体分析，根据系统的原理，优化逻辑。刚才说了，Hive最重要的是Group By和Join，所以下面我们讲Join.

JOIN
例如这样一个查询：

INSERT INTO TABLE pv_users 
SELECT pv.pageid, u.age 
FROM page_view pv JOIN user u ON (pv.userid = u.userid);

image

把访问和用户表进行关联，生成访问用户表。Hive的Join也是通过MapReduce来完成的。

image

就上面的查询，在ＭapReduce的Join的实现过程如下：

image

Ｍap端会分别读入各个表的一部分数据，把这部分数据进行打标，例如pv表标1，user表标2.

Map读取是分布式进行的。标完完后分发到Reduce端，Reduce 端根据Join Key，也就是关联键进行分组。然后按打的标进行排序，也就是图上的Shuffle Sort。

在每一个Reduce分组中，Key为111的在一起，也就是一台机器上。同时，pv表的数据在这台机器的上端，user表的数据在这台机器的下端。

这时候，Reduce把pv表的数据读入到内存里，然后逐条与硬盘上user表的数据做Join就可以了。

从这个实现可以看出，我们在写Hive Join的时候，应该尽可能把小表（分布均匀的表）写在左边，大表（或倾斜表）写在右边。这样可以有效利用内存和硬盘的关系，增强Hive的处理能力。

同时由于使用Join Key进行分发， Hive也只支持等值Join，不支持非等值Join。由于Join和Group By一样存在分发，所以也同样存在着倾斜的问题。所以Join也要对抗倾斜数据，提升查询执行性能。

通常，有一种执行非常快的Join叫Map Join 。

Map Join 优化

手动的Map Join SQL如下：

INSERT INTO TABLE pv_users 
SELECT /*+ MAPJOIN(pv) */ pv.pageid, u.age 
FROM page_view pv JOIN user u 
ON (pv.userid = u.userid);

还是刚才的例子，用Map Join执行

Map Join通常只适用于一个大表和一个小表做关联的场景，例如事实表和维表的关联。

原理如上图，用户可以手动指定哪个表是小表，然后在客户端把小表打成一个哈希表序列化文件的压缩包，通过分布式缓存均匀分发到作业执行的每一个结点上。然后在结点上进行解压，在内存中完成关联。

Map Join全过程不会使用Reduce，非常均匀，不会存在数据倾斜问题。默认情况下，小表不应该超过25M。在实际使用过程中，手动判断是不是应该用Map Join太麻烦了，而且小表可能来自于子查询的结果。

Hive有一种稍微复杂一点的机制，叫Auto Map Join

还记得原理中提到的物理优化器？Physical Optimizer么？它的其中一个功能就是把Join优化成Auto Map Join

图上左边是优化前的，右边是优化后的

优化过程是把Join作业前面加上一个条件选择器ConditionalTask和一个分支。左边的分支是MapJoin，右边的分支是Common Join(Reduce Join)

看看左边的分支是不是和我们上上一张图很像？

这个时候，我们在执行的时候，就由这个Conditional Task 进行实时路径选择，遇到小于25兆走左边，大于25兆走右边。

在比较新版的Hive中，Auto Mapjoin是默认开启的。如果没有开启，可以使用一个开关， set hive.auto.convert.join=true 开启。
当然，Join也会遇到和上面的Group By一样的倾斜问题。

Ｈive 也可以通过像Group By一样两道作业的模式单独处理一行或者多行倾斜的数据。

hive 中设定

set hive.optimize.skewjoin = true; 
set hive.skewjoin.key = skew_key_threshold （default = 100000）

其原理是就在Reduce Join过程，把超过十万条的倾斜键的行写到文件里，回头再起一道Join单行的Map Join作业来单独收拾它们。最后把结果取并集就是了。如上图所示。