优化HIVE，

优化HIVE，

1、提前过滤数据，减少中间数据依赖：
尽量尽早的过滤数据，减少每个阶段的数据量，对于分区表要加分区，同时只选择需要使用到的数据。
如下，改写后的写法将会大大减少join的数据量
select ... from A
join B
on A.key = B.key
where A.userid>10
     and B.userid<10
        and A.dt='20120417'
        and B.dt='20120417';
应该改写为：
select .... from (select .... from A
                  where dt='201200417'
                                    and userid>10
                              ) a
join ( select .... from B
       where dt='201200417'
                     and userid < 10  
     ) b
on a.key = b.key;
 
2、慎重使用mapjoin，小表注意放在join的左边，否则易导致磁盘及内存的大量消耗
3、避免使用笛卡尔积，笛卡尔积只有1个reduce task，会导致计算超慢，甚至计算不出结果或节点挂掉
如select * from gbk， utf8 where gbk.key= utf8.key and gbk.key > 10;
select * from gbk join utf8 where gbk.key= utf8.key and gbk.key > 10;
上面两种写法就会产生笛卡尔积，可改写为如下形式：
select * from gbk join utf8 on gbk.key= utf8.key where gbk.key > 10;
 
         此外，当多个数据表做连接时，有些童靴喜欢写成以下形式：
 tablea join tableb join tablec join ... on tablea.col1 = tableb.col2 and ...
这种形式也会导致笛卡尔积。这种情况可改写为：
 tablea join tableb on （ tablea.col1 = tableb.col2 and ... ） join tablec on  ...  join  ... on  ...
至于连接条件中有or这种情况，可以考虑使用join + union all或者in等方式来改写。
 
4、列裁剪（Column Pruning）
在读数据的时候，只读取查询中需要用到的列，而忽略其他列。例如，对于查询：
SELECT a,b FROM T WHERE e < 10;
其中，T 包含 5 个列 (a,b,c,d,e)，列 c，d 将会被忽略，只会读取a, b, e 列
这个选项参数默认为真： hive.optimize.cp = true
分区裁剪（Partition Pruning）
在查询的过程中减少不必要的分区。例如，对于下列查询：
SELECT * FROM (SELECT c1, COUNT(1)
       FROM T GROUP BY c1) subq
       WHERE subq.prtn = 100;
 
SELECT * FROM T1 JOIN
       (SELECT * FROM T2) subq ON (T1.c1=subq.c2)
       WHERE subq.prtn = 100;
会在子查询中就考虑 subq.prtn = 100 条件，从而减少读入的分区数目。
此选项参数默认为真：hive.optimize.pruner=true
 
5、explain命令对于我们优化查询语句很重要，针对某些查询语句，我们可以通过它查看各个执行计划，针对耗时的地方，优化之。
语法如下：EXPLAIN [EXTENDED] query
参考使用方法：http://www.cnblogs.com/ggjucheng/archive/2013/01/16/2861665.html
 
6、union all改写为join
当需要把几个数据集的结果合并时，能使用join的话就不要使用union all。因为使用union all时通常需要加入大量的0，这会导致中间结果膨胀，增加系统负担。
 
7、大数据量时全局conut/sum distinct优化
         以下两个SQL，有人可能认为例句1快于例句2。
例句1：
SELECT
                 COUNT(DISTINCT uin) login_uins   
        FROM
                 tabled
        WHERE
                 ftime >= 20121001
                 AND ftime <= 20121001
例句2：      
SELECT
            COUNT(uin) login_uins
         FROM
         ( select distinct uin from tabled  WHERE
                 ftime >= 20121001
         AND ftime <= 20121001   ) subq
实际上，例句2远远快于例句1。
例句1虽然只有1个mr，但是，这个mr却只有1个reduce任务，导致这个reduce任务需要读取和处理大量的数据，这不仅导致执行慢，而且，在如果tabled中的数据量太大，可能导致执行reduce任务的节点down掉。
       例句2先去重,后全局统计，没有像例句1那样明显的瓶颈。所以，请牢记当对某个数据集进行全局count distinct操作时，尽量用例句2的形式。
在其他情况下，即使有group by，但是group by的维度太低，也可以考虑这种改写方法。也就是说，一定要避免group by的维度太低。
 
8、有小表存在时，考虑使用mapjoin
         使用Map Join的必要条件：
a.参与连接的小表的行数，以不超过2万条为宜。
b.连接类型是inner join、right outer join（小表不能是右表）、left outer join（小表不能是左表）、left semi join。
使用方法示例：
例如：
INSERT OVERWRITE 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会极大加速计算。
 
9、数据倾斜（转载）
 
数据倾斜的外在表现是执行时间超长。查看任务的监控页面可以发现，除了一个或几个任务外，其他reduce任务都执行的很快。这种情况下，只有一个解释：数据倾斜。
在做Shuffle阶段的优化过程中，遇到了数据倾斜的问题，造成了对一些情况下优化效果不明显。主要是因为在Job完成后的所得到的Counters是整个Job的总和，优化是基于这些Counters得出的平均值，而由于数据倾斜的原因造成map处理数据量的差异过大，使得这些平均值能代表的价值降低。Hive的执行是分阶段的，map处理数据量的差异取决于上一个stage的reduce输出，所以如何将数据均匀的分配到各个reduce中，就是解决数据倾斜的根本所在。规避错误来更好的运行比解决错误更高效。在查看了一些资料后，总结如下。
数据倾斜的原因
操作：
关键词
情形
后果
Join
其中一个表较小，
但是key集中
分发到某一个或几个Reduce上的数据远高于平均值
大表与大表，但是分桶的判断字段0值或空值过多
这些空值都由一个reduce处理，灰常慢
group by
group by 维度过小，
某值的数量过多
处理某值的reduce灰常耗时
Count Distinct
某特殊值过多
处理此特殊值的reduce耗时
原因
a)、key分布不均匀
b)、业务数据本身的特性
c)、建表时考虑不周
d)、某些SQL语句本身就有数据倾斜
表现
任务进度长时间维持在99%（或100%），查看任务监控页面，发现只有少量（1个或几个）reduce子任务未完成。因为其处理的数据量和其他reduce差异过大。
单一reduce的记录数与平均记录数差异过大，通常可能达到3倍甚至更多。 最长时长远大于平均时长。
数据倾斜的解决方案
2.1) 参数调节
a) hive.map.aggr = true
Map 端部分聚合，相当于Combiner
b) hive.groupby.skewindata=true
有数据倾斜的时候进行负载均衡，当选项设定为 true，生成的查询计划会有两个 MR Job。第一个 MR Job 中，Map 的输出结果集合会随机分布到 Reduce 中，每个 Reduce 做部分聚合操作，并输出结果，这样处理的结果是相同的 Group By Key 有可能被分发到不同的 Reduce 中，从而达到负载均衡的目的；第二个 MR Job 再根据预处理的数据结果按照 Group By Key 分布到 Reduce 中（这个过程可以保证相同的 Group By Key 被分布到同一个 Reduce 中），最后完成最终的聚合操作。
2.2)  SQL语句调整
a) 如何Join：
关于驱动表的选取，选用join key分布最均匀的表作为驱动表
做好列裁剪和filter操作，以达到两表做join的时候，数据量相对变小的效果。
b) 大小表Join：
使用map join让小的维度表（1000条以下的记录条数） 先进内存。在map端完成reduce.
c) 大表Join大表：
把空值的key变成一个字符串加上随机数，把倾斜的数据分到不同的reduce上，由于null值关联不上，处理后并不影响最终结果。
d) count distinct大量相同特殊值
count distinct时，将值为空的情况单独处理，如果是计算count distinct，可以不用处理，直接过滤，在最后结果中加1。如果还有其他计算，需要进行group by，可以先将值为空的记录单独处理，再和其他计算结果进行union。
e) group by维度过小：
采用sum() group by的方式来替换count(distinct)完成计算。
f)特殊情况特殊处理：
在业务逻辑优化效果的不大情况下，有些时候是可以将倾斜的数据单独拿出来处理。最后union回去。
典型的业务场景
3.1)  空值产生的数据倾斜
场景：如日志中，常会有信息丢失的问题，比如日志中的 user_id，如果取其中的 user_id 和 用户表中的user_id 关联，会碰到数据倾斜的问题。
解决方法1： user_id为空的不参与关联
select * from log a
  join users b
  on a.user_id is not null
  and a.user_id = b.user_id
union all
select * from log a
  where a.user_id is null;
解决方法2 ：赋与空值分新的key值
select *
  from log a
  left outer join users b
  on case when a.user_id is null then concat(‘hive’,rand() ) else a.user_id end = b.user_id;
结论：方法2比方法1效率更好，不但io少了，而且作业数也少了。解决方法1中 log读取两次，jobs是2。解决方法2 job数是1 。这个优化适合无效 id (比如 -99 , ’’, null 等) 产生的倾斜问题。把空值的 key 变成一个字符串加上随机数，就能把倾斜的数据分到不同的reduce上 ,解决数据倾斜问题。
3.2)  不同数据类型关联产生数据倾斜
场景：用户表中user_id字段为int，log表中user_id字段既有string类型也有int类型。当按照user_id进行两个表的Join操作时，默认的Hash操作会按int型的id来进行分配，这样会导致所有string类型id的记录都分配到一个Reducer中。
解决方法：把数字类型转换成字符串类型
select * from users a
  left outer join logs b
  on a.usr_id = cast(b.user_id as string)
3.3)  小表不小不大，怎么用 map join 解决倾斜问题
使用 map join 解决小表(记录数少)关联大表的数据倾斜问题，这个方法使用的频率非常高，但如果小表很大，大到map join会出现bug或异常，这时就需要特别的处理。 以下例子:
select * from log a
  left outer join users b
  on a.user_id = b.user_id;
users 表有 600w+ 的记录，把 users 分发到所有的 map 上也是个不小的开销，而且 map join 不支持这么大的小表。如果用普通的 join，又会碰到数据倾斜的问题。
解决方法：
select /*+mapjoin(x)*/* from log a
  left outer join (
    select  /*+mapjoin(c)*/d.*
      from ( select distinct user_id from log ) c
      join users d
      on c.user_id = d.user_id
    ) x
  on a.user_id = b.user_id;
假如，log里user_id有上百万个，这就又回到原来map join问题。所幸，每日的会员uv不会太多，有交易的会员不会太多，有点击的会员不会太多，有佣金的会员不会太多等等。所以这个方法能解决很多场景下的数据倾斜问题。
总结
使map的输出数据更均匀的分布到reduce中去，是我们的最终目标。由于Hash算法的局限性，按key Hash会或多或少的造成数据倾斜。大量经验表明数据倾斜的原因是人为的建表疏忽或业务逻辑可以规避的。在此给出较为通用的步骤：
1、采样log表，哪些user_id比较倾斜，得到一个结果表tmp1。由于对计算框架来说，所有的数据过来，他都是不知道数据分布情况的，所以采样是并不可少的。
2、数据的分布符合社会学统计规则，贫富不均。倾斜的key不会太多，就像一个社会的富人不多，奇特的人不多一样。所以tmp1记录数会很少。把tmp1和users做map join生成tmp2,把tmp2读到distribute file cache。这是一个map过程。
3、map读入users和log，假如记录来自log,则检查user_id是否在tmp2里，如果是，输出到本地文件a,否则生成<user_id,value>的key,value对，假如记录来自member,生成<user_id,value>的key,value对，进入reduce阶段。
4、最终把a文件，把Stage3 reduce阶段输出的文件合并起写到hdfs。
如果确认业务需要这样倾斜的逻辑，考虑以下的优化方案：
 
1、对于join，在判断小表不大于1G的情况下，使用map join
 
2、对于group by或distinct，设定 hive.groupby.skewindata=true
 
10、数据膨胀导致reduce数不合理
         数据膨胀很多时候也会导致reduce任务个数过少。reduce数目是否合理，可以从任务的mr监控页面发现端倪。请注意job监控页面的一个参数：
 
Reduce shuffle bytes        reduce任务读取字节数
一般情况下，reduce任务的个数应大致等于上面这个参数的大小（请换算成GB），如果该参数大小是reduce任务数的数倍，那就意味着reduce任务数不合理。
解决方法：
通过设置reduce任务数提高并行度来加速执行：
set mapred.reduce.tasks=N; //执行语句之前
set mapred.reduce.tasks=-1; //执行语句之后恢复原状
注意：请合理设置N的大小，最好设置为上述参数的大小。不要超过999！
 
11、合并小文件
         文件数目过多，会给 HDFS 带来压力，并且会影响处理效率，可以通过合并 Map 和 Reduce 的结果文件来消除这样的影响：
hive.merge.mapfiles = true是否和并 Map 输出文件，默认为 True
hive.merge.mapredfiles = false是否合并 Reduce 输出文件，默认为 False
hive.merge.size.per.task = 256*1000*1000合并文件的大小
 
12、排序优化
         Order by 实现全局排序，是由一个reduce task实现，效率比较低
Sort by 实现部分有序，单个reduce输出的结果是有序的，效率高，通常和DISTRIBUTE BY关键字一起使用（DISTRIBUTE BY关键字 可以指定map 到 reduce端的分发key）
CLUSTER BY col1 等价于DISTRIBUTE BY col1 SORT BY col1
 
13、动态分区
         在hive中，有时候会希望根据输入的key，把结果自动输出到不同的目录中，这可以通过动态分区来实现，就是把每一个key当作一个分区。
如果要启动动态分区，则需要进行下面的设置
首先需要在hive语句中设置允许动态分区
set hive.exec.dynamic.partition=true;
set hive.exec.dynamic.partition.mode=nonstrict;
在动态分区有可能很大的情况下，还需要其他的调整
hive.exec.dynamic.partitions.pernode 参数指的是每个节点上能够生成的最大分区，这个在最坏情况下应该是跟最大分区一样的值
hive.exec.dynamic.partitions.partitions 参数指的是总共的最大的动态分区数
hive.exec.max.created.files 参数指的是能够创建的最多文件数（分区一多，文件必然就多了...）
最后要注意的是select语句中要把distribute的key也select出来
适当设置reduce个数 mapred.reduce.tasks
比如经常遇到的此问题：
org.apache.hadoop.hive.ql.metadata.HiveException: org.apache.hadoop.ipc.RemoteException: org.apache.hadoop.hdfs.server.namenode.LeaseExpiredException: No lease on /tmp/hive-maintain/hive_2012-11-28_22-39-43_810_1689858262130334284/_task_tmp.-ext-10002/part=33436268/_tmp.000004_0 File does not exist. Holder DFSClient_attempt_201211250925_9859_m_000004_0 does not have any open files.
可以设置
SET hive.exec.max.dynamic.partitions=100000;
SET hive.exec.max.dynamic.partitions.pernode=100000;


14.数据量比较大的时候请尽量避免使用count(distinct *)

HQL写的好坏对性能的影响还是很大的，分享一个今天对一个HQL的优化供大家参考：
 
一开发给的原始HQL：
SELECT COUNT(DISTINCT a.deviceid)
FROM gbd_mobile_safe.td_app_rym a
WHERE a.appkey='97750A70867132C5E0CFBA22762BEA7E'
AND a.deviceid NOT IN
(SELECT DISTINCT b.device_id FROM jt_mobile_dm_safe.cx_app_detail b)
 
数据量：td_app_rym有约14.8亿行数据，cx_app_detail有约700万行数据
 
此HQL运行了5个多小时仍未运行出结果，有可能因reduce task OOM而根本就跑不出结果，原因在与stage-3只有一个reduce，数据量大时会很慢
Hadoop job information for Stage-3: number of mappers: 2137; number of reducers: 1
此HQL使用了count(distinct ***)的写法，当数量比较大时，此种写法效率将会非常低效，主要是因为所有的数据都shuffle到一个reduce处理
 
 
优化1：改为如下方式进行查询时，38分钟即可查询出结果，效率大大提高，但此时还是会产生笛卡尔积，还有优化空间
SELECT COUNT( a.deviceid)
FROM  (
SELECT DISTINCT deviceid
FROM gbd_mobile_safe.td_app_rym
WHERE appkey='97750A70867132C5E0CFBA22762BEA7E') a
WHERE a.deviceid NOT IN
       (SELECT DISTINCT b.device_id
FROM jt_mobile_dm_safe.cx_app_detail b)
 
优化2：改为如下方式查询时，20分钟即可查询出结果，效率再次提升，主要是使用left outer join替代not in
SELECT COUNT(a.deviceid)
FROM  (
SELECT DISTINCT deviceid
FROM gbd_mobile_safe.td_app_rym
WHERE appkey='97750A70867132C5E0CFBA22762BEA7E') a
LEFT OUTER JOIN  (
SELECT DISTINCT b.device_id
FROM jt_mobile_dm_safe.cx_app_detail b ) c
ON a.deviceid = c.device_id
WHERE  c.device_id IS null;
 
优化3：改为如下方式查询时，13分钟即可查询出结果，效率继续提升，主要是使用group by替代distinct
SELECT COUNT(a.deviceid)
FROM  (
SELECT deviceid
FROM gbd_mobile_safe.td_app_rym
WHERE appkey='97750A70867132C5E0CFBA22762BEA7E'
GROUP BY deviceid ) a
LEFT OUTER JOIN  (
SELECT  b.device_id
FROM  jt_mobile_dm_safe.cx_app_detail b
GROUP BY device_id) c
ON a.deviceid = c.device_id
WHERE c.device_id IS null;
 
上面三种查询方式reduce的个数会根据输入数据量来确定，也可以手动set，至于设置为多少合适大家可以试试
 
15.防数据倾斜优化
原始语句如下，看似很简单的一个SQL，但却运行了2个小时仍未跑完
数据量：cif2_party_business_telephone表约1.4亿行数据，250G左右，未做分区
use gbd_mobile_safe;
          set mapred.job.queue.name=${queueName};
          set hive.exec.dynamic.partition=true;
          set hive.exec.dynamic.partition.mode=nonstrict;
 
          insert overwrite table yhy_party_business_telephone partition(y,m,d)
          select alias_party_no,
                 customer_type,
                 party_no,
                 business,
                 site_no,
                 tel_no,
                 formatted_tel_no,
                 is_ignored,
                 is_valid,
                 '2015',
                 '08',
                 '10'
            from crm1.cif2_party_business_telephone;
查看运行日志发现数据倾斜非常严重：




优化1：添加如下参数缓解数据倾斜
hive.groupby.skewindata=true
试验证明基本上没有效果，究其原因估计于此reduce发生在将map后的结果往对应的partition写阶段，故shuffle的key值就是partition字段，而此sql只是往y=2015,m=08,d=10这个分区写
所以基本所有数据都会被shuffle到同一个reduce
 
优化2：改写SQL，由于只向一个分区写数据，直接指定分区partition(y=’2015’,m=’08’,d=’10’)
原SQL使用了动态分区，且为了匹配动态分区认为的在select上增加了如下三列常量数据，这同时也会增加mapreduce task的数据量
  '2015',
                 '08',
                 '10'
 
而如果直接指定分区partition(y=’2015’,m=’08’,d=’10’)则不需要在select中添加上面三列，更重要的是整个job没有reduce操作，没有reduce也就根本不可能产生数据倾斜了
改写后的SQL如下：
INSERT OVERWRITE TABLE yhy_party_business_telephone PARTITION(y='2015',m='08',d='10')
          SELECT    alias_party_no,
                            customer_type,
                            party_no,
                             business,
                            site_no,
                             tel_no,
                            formatted_tel_no,
                             is_ignored,
                             is_valid
          FROM crm.cif2_party_business_telephone;
16.   set hive.cli.print.header=true; // 打印列名
  set hive.cli.print.row.to.vertical=true; // 开启行转列功能, 前提必须开启打印列名功能
  set hive.cli.print.row.to.vertical.num=1; // 设置每行显示的列数