Hive入门到剖析（三）

8 Hive Shell操作

8.1 Hive bin下脚本介绍

8.2 Hive Shell 基本操作

1、Hive 命令行

hive [-hiveconf x=y]* [<-ifilename>]* [<-f filename>|<-e query-string>] [-S]

-i  从文件初始化HQL

-e  从命令行执行指定的HQL

-f  执行HQL脚本

-v  输出执行的HQL语句到控制台

-p <port> connect to HiveServer on port number -hiveconf x=y Use this to set hive/hadoop configurationvariables.

Hive 命令行示例

从命令行执行指定的sql语句

$HIVE_HOME/bin/hive -e 'select a.colfrom tab1 a'

以指定的hive环境变量执行指定的sql语句

$HIVE_HOME/bin/hive -e 'select a.colfrom tab1 a' -hiveconf hive.exec.scratchdir=/home/my/hive_scratch -hiveconfmapred.reduce.tasks=32

以沉默模式执行指定的sql语句，并将执行结果导出到指定文件:

HIVE_HOME/bin/hive  -e'select a.col from tab1 a' > a.txt

以非交互式模式执行sql文件

HIVE_HOME/bin/hive -f/home/my/hive-script.sql

在进入交互模式之前，执行初始化sql文件

HIVE_HOME/bin/hive -i/home/my/hive-init.sql

Hive 交互式Shell命令

当命令 $HIVE_HOME/bin/hive以不带 -e/-f 选项的方式运行时， hive将进入到交互模式

以（;）冒号结束命令行

8.3  日志

Hive使用Log4J来处理日志

我们可以通过下面的命令设计Hive的日志级别

$HIVE_HOME/bin/hive -hiveconfhive.root.logger=INFO,console

hive.root.logger的有INFO,DEBUG, 等

8.4  资源

Hive添加资源

Hive可以动态的添加资源，如文件

一般情况下，我们是在与Hive进行交互时添加文件

实际上是使用Hadoop的 Distributed Cache来控制的

例子

ADD { FILE[S] | JAR[S] | ARCHIVE[S]} <filepath2> [<filepath3>]*

LIST { FILE[S] | JAR[S] | ARCHIVE[S]} [<filepath2> <filepath3> ..]

DELETE { FILE[S] | JAR[S] |ARCHIVE[S] } [<filepath2> <filepath3> ..]

9 Hive优化

9.1  Hadoop 计算框架的特性

1、什么是数据倾斜

由于数据的不均衡原因，导致数据分布不均匀，造成数据大量的集中到一点，造成数据热点。

2、Hadoop框架的特性

不怕数据大，怕数据倾斜

jobs数比较多的作业运行效率相对比较低，比如即使有几百行的表，如果多次关联多次汇总，产生十几个jobs，耗时很长。原因是map reduce作业初始化的时间是比较长的

sum,count,max,min等UDAF，不怕数据倾斜问题,hadoop在map端的汇总合并优化，使数据倾斜不成问题

count(distinct ),在数据量大的情况下，效率较低，因为count(distinct)是按group by 字段分组，按distinct字段排序，一般这种分布方式是很倾斜的。

9.2  优化的常用手段

解决数据倾斜问题

减少job数

设置合理的map reduce的task数，能有效提升性能。

了解数据分布，自己动手解决数据倾斜问题是个不错的选择

数据量较大的情况下，慎用count(distinct)。

对小文件进行合并，是行至有效的提高调度效率的方法。

优化时把握整体，单个作业最优不如整体最优。

9.3  Hive的数据类型方面的优化--优化原则

按照一定规则分区（例如根据日期）。通过分区，查询的时候指定分区，会大大减少在无用数据上的扫描, 同时也非常方便数据清理。

合理的设置Buckets。在一些大数据join的情况下，map join有时候会内存不够。如果使用Bucket Map Join的话，可以只把其中的一个bucket放到内存中，内存中原来放不下的内存表就变得可以放下。这需要使用buckets的键进行join的条件连结，并且需要如下设置

 set hive.optimize.bucketmapjoin = true

9.4  Hive的操作方面的优化

全排序

怎样做笛卡尔积

怎样决定map个数

怎样决定reducer个数

合并MapReduce操作

Bucket 与sampling

Partition

JOIN

Group By

合并小文件

1、全排序

Hive的排序关键字是SORTBY，它有意区别于传统数据库的ORDER BY也是为了强调两者的区别–SORT BY只能在单机范围内排序。

2、怎样做笛卡尔积

当Hive设定为严格模式（hive.mapred.mode=strict）时，不允许在HQL语句中出现笛卡尔积

MapJoin是的解决办法

MapJoin，顾名思义，会在Map端完成Join操作。这需要将Join操作的一个或多个表完全读入内存

MapJoin的用法是在查询/子查询的SELECT关键字后面添加/*+MAPJOIN(tablelist) */提示优化器转化为MapJoin（目前Hive的优化器不能自动优化MapJoin）

其中tablelist可以是一个表，或以逗号连接的表的列表。tablelist中的表将会读入内存，应该将小表写在这里

在大表和小表做笛卡尔积时，规避笛卡尔积的方法是，给Join添加一个Join key，原理很简单：将小表扩充一列join key，并将小表的条目复制数倍，join key各不相同；将大表扩充一列join key为随机数

3、控制Hive的Map数

通常情况下，作业会通过input的目录产生一个或者多个map任务

主要的决定因素有： input的文件总个数，input的文件大小，集群设置的文件块大小(目前为128M, 可在hive中通过setdfs.block.size;命令查看到，该参数不能自定义修改)

是不是map数越多越好

答案是否定的。如果一个任务有很多小文件（远远小于块大小128m）,则每个小文件也会被当做一个块，用一个map任务来完成，而一个map任务启动和初始化的时间远远大于逻辑处理的时间，就会造成很大的资源浪费。而且，同时可执行的map数是受限的

是不是保证每个map处理接近128m的文件块，就高枕无忧了？

     答案也是不一定。比如有一个127m的文件，正常会用一个map去完成，但这个文件只有一个或者两个小字段，却有几千万的记录，

如果map处理的逻辑比较复杂，用一个map任务去做，肯定也比较耗时。 

针对上面的问题3和4，我们需要采取两种方式来解决：即减少map数和增加map数；

是不是保证每个map处理接近128m的文件块，就高枕无忧了？

     答案也是不一定。比如有一个127m的文件，正常会用一个map去完成，但这个文件只有一个或者两个小字段，却有几千万的记录，

如果map处理的逻辑比较复杂，用一个map任务去做，肯定也比较耗时。 

针对上面的问题3和4，我们需要采取两种方式来解决：即减少map数和增加map数；

举例

    a) 假设input目录下有1个文件a,大小为780M,那么hadoop会将该文件a分隔成7个块（6个128m的块和1个12m的块），从而产生7个map数

    b)    假设input目录下有3个文件a,b,c,大小分别为10m，20m，130m，那么hadoop会分隔成4个块（10m,20m,128m,2m）,从而产生4个map数

即，如果文件大于块大小(128m),那么会拆分，如果小于块大小，则把该文件当成一个块

4、怎样决定reducer个数

Hadoop MapReduce程序中，reducer个数的设定极大影响执行效率

不指定reducer个数的情况下，Hive会猜测确定一个reducer个数，基于以下两个设定：

     参数1：hive.exec.reducers.bytes.per.reducer（默认为1G)

     参数2 ：hive.exec.reducers.max（默认为999）

计算reducer数的公式

N=min(参数2，总输入数据量/参数1)

依据Hadoop的经验，可以将参数2设定为0.95*(集群中TaskTracker个数)

reduce个数并不是越多越好

同map一样，启动和初始化reduce也会消耗时间和资源；

另外，有多少个reduce,就会有多少个输出文件，如果生成了很多个小文件，那么如果这些小文件作为下一个任务的输入，则也会出现小文件过多的问题

什么情况下只有一个reduce

    很多时候你会发现任务中不管数据量多大，不管你有没有设置调整reduce个数的参数，任务中一直都只有一个reduce任务；

其实只有一个reduce任务的情况，除了数据量小于

      hive.exec.reducers.bytes.per.reducer参数值的情况外，还有以下原因：

        a)    没有group by的汇总

        b)    用了Order by

5、合并 MapReduce操作

Multi-group by

Multi-group by是Hive的一个非常好的特性，它使得Hive中利用中间结果变得非常方便

FROM log

 insert overwrite table test1 select log.id group by log.id

  insert  overwrite table test2select log.name group by log.name

上述查询语句使用了Multi-group by特性连续group by了2次数据，使用不同的groupby key。这一特性可以减少一次MapReduce操作。

6、Bucket 与 Sampling

Bucket是指将数据以指定列的值为key进行hash，hash到指定数目的桶中。这样就可以支持高效采样了

Sampling可以在全体数据上进行采样，这样效率自然就低，它还是要去访问所有数据。而如果一个表已经对某一列制作了bucket，就可以采样所有桶中指定序号的某个桶，这就减少了访问量。

如下例所示就是采样了test中32个桶中的第三个桶。

SELECT * FROM test 、、、TABLESAMPLE(BUCKET 3 OUT OF 32);

7、JOIN 原则

在使用写有 Join 操作的查询语句时有一条原则：应该将条目少的表/子查询放在 Join 操作符的左边

原因是在 Join 操作的 Reduce 阶段，位于 Join 操作符左边的表的内容会被加载进内存，将条目少的表放在左边，可以有效减少发生 OOM 错误的几率。

8、Map Join

Join 操作在 Map阶段完成，不再需要Reduce，前提条件是需要的数据在 Map的过程中可以访问到

例如：

INSERT OVERWRITE TABLE phone_traffic
SELECT /*+MAPJOIN(phone_location) */ l.phone,p.location,l.traffic from phone_location p join log l on(p.phone=l.phone)

相关的参数为：

hive.join.emit.interval = 1000How many rows in the right-most join operand Hive should buffer before emittingthe join result.
hive.mapjoin.size.key = 10000
hive.mapjoin.cache.numrows =10000

9、Group By

Map 端部分聚合

并不是所有的聚合操作都需要在 Reduce 端完成，很多聚合操作都可以先在 Map 端进行部分聚合，最后在 Reduce 端得出最终结果

基于 Hash

参数包括：

hive.map.aggr = true 是否在 Map 端进行聚合，默认为 True

hive.groupby.mapaggr.checkinterval =100000 在 Map 端进行聚合操作的条目数目

有数据倾斜的时候进行负载均衡

hive.groupby.skewindata = false

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

10、合并小文件

文件数目过多，会给 HDFS 带来压力，并且会影响处理效率，可以通过合并 Map 和 Reduce 的结果文件来消除这样的影响：

hive.merge.mapfiles = true 是否和并 Map 输出文件，默认为 True

hive.merge.mapredfiles = false 是否合并 Reduce 输出文件，默认为 False

hive.merge.size.per.task =256*1000*1000 合并文件的大小