Hive QL详解

第一部分：Hadoop 计算框架的特性
什么是数据倾斜
•由于数据的不均衡原因，导致数据分布不均匀，造成数据大量的集中到一点，造成数据热点
Hadoop框架的特性
•不怕数据大，怕数据倾斜
•jobs数比较多的作业运行效率相对比较低，比如即使有几百行的表，如果多次关联多次汇总，产生十几个jobs，耗时很长。原因是map reduce作业初始化的时间是比较长的
•sum,count,max,min等UDAF，不怕数据倾斜问题,hadoop在map端的汇总合并优化，使数据倾斜不成问题
•count(distinct ),在数据量大的情况下，效率较低，因为count(distinct)是按group by 字段分组，按distinct字段排序，一般这种分布方式是很倾斜的
第二部分：优化的常用手段
优化的常用手段
•解决数据倾斜问题
•减少job数
•设置合理的map reduce的task数，能有效提升性能。
•了解数据分布，自己动手解决数据倾斜问题是个不错的选择
•数据量较大的情况下，慎用count(distinct)。
•对小文件进行合并，是行至有效的提高调度效率的方法。
•优化时把握整体，单个作业最优不如整体最优。


第三部分：Hive的数据类型方面的优化
优化原则
•按照一定规则分区（例如根据日期）。通过分区，查询的时候指定分区，会大大减少在无用数据上的扫描, 同时也非常方便数据清理。
•合理的设置Buckets。在一些大数据join的情况下，map join有时候会内存不够。如果使用Bucket Map Join的话，可以只把其中的一个bucket放到内存中，内存中原来放不下的内存表就变得可以放下。这需要使用buckets的键进行join的条件连结，并且需要如下设置
     set hive.optimize.bucketmapjoin = true
第四部分：Hive的操作方面的优化
•全排序
•怎样做笛卡尔积
•怎样决定map个数
•怎样决定reducer个数
•合并MapReduce操作
•Bucket 与 sampling
•Partition
•JOIN
•Group By
•合并小文件
全排序
•Hive的排序关键字是SORT BY，它有意区别于传统数据库的ORDER BY也是为了强调两者的区别–SORT BY只能在单机范围内排序
怎样做笛卡尔积
•当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为随机数

控制Hive的Map数
•通常情况下，作业会通过input的目录产生一个或者多个map任务
•主要的决定因素有： input的文件总个数，input的文件大小，集群设置的文件块大小(目前为128M, 可在hive中通过set dfs.block.size;命令查看到，该参数不能自定义修改)
•是不是map数越多越好
答案是否定的。如果一个任务有很多小文件（远远小于块大小128m）,则每个小文件也会被当做一个块，用一个map任务来完成，
而一个map任务启动和初始化的时间远远大于逻辑处理的时间，就会造成很大的资源浪费。
而且，同时可执行的map数是受限的
•是不是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),那么会拆分，如果小于块大小，则把该文件当成一个块

怎样决定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
合并 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 test2 select log.name group by log.name
• 上述查询语句使用了Multi-group by特性连续group by了2次数据，使用不同的group by key。这一特性可以减少一次MapReduce操作。
Bucket 与 Sampling
•Bucket是指将数据以指定列的值为key进行hash，hash到指定数目的桶中。这样就可以支持高效采样了
•Sampling可以在全体数据上进行采样，这样效率自然就低，它还是要去访问所有数据。而如果一个表已经对某一列制作了bucket，就可以采样所有桶中指定序号的某个桶，这就减少了访问量。
•如下例所示就是采样了test中32个桶中的第三个桶。
•SELECT * FROM test 、、、TABLESAMPLE(BUCKET 3 OUT OF 32);
JOIN 原则
•在使用写有 Join 操作的查询语句时有一条原则：应该将条目少的表/子查询放在 Join 操作符的左边
•原因是在 Join 操作的 Reduce 阶段，位于 Join 操作符左边的表的内容会被加载进内存，将条目少的表放在左边，可以有效减少发生 OOM 错误的几率
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 = 1000 How many rows in the right-most join operand Hive should buffer before emitting the join result.
hive.mapjoin.size.key = 10000
hive.mapjoin.cache.numrows = 10000
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 By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作。
合并小文件
•文件数目过多，会给 HDFS 带来压力，并且会影响处理效率，可以通过合并 Map 和 Reduce 的结果文件来消除这样的影响：
•hive.merge.mapfiles = true 是否和并 Map 输出文件，默认为 True
•hive.merge.mapredfiles = false 是否合并 Reduce 输出文件，默认为 False
•hive.merge.size.per.task = 256*1000*1000 合并文件的大小
私塾在线原创