数据量大不是问题，数据倾斜是个问题。

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

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

count（distinct），在数据量大的情况下，效率较低，如果是多count（distinct）效率更低，因为count（distinct）是按group by字段分组，按distinct字段排序，一般这种分布式是很倾斜的，比如男uv，女uv，淘宝一天30亿的pv，如果按性别分组，分配2个reduce，每个reduce处理15亿数据。+

 好的模型设计事半功倍。

场景： 有一张user表，为卖家每天收入表，user_id，ds(日期)为key，属性有主营类目，指标有交易金额，交易笔数。每天要取前10天的总收入，总笔数，和最近一天的主营类目

常规方法：

1.利用分析函数，取每个user_id最近一天的主营类目，存入临时表t1

2.汇总10天的总交易金额，交易比数，存入临时表t2

3.关联t1,t2，得到最终的结果

解决方法2：

select user_id,

  substr(max(concat(ds,cat)),9) as main_cat,

  sum(qty),

  sum(amt)

from users

where ds between 20101201 and 20101210

group by user_id;

解决方法2的开销等于常规方法的第二步的开销，整体性能提升2倍。这种方式在RAC上也适用。

场景：如日志中，常会有信息丢失的问题，比如全网日志中的 user_id，如果取其中的 user_id 和 bmw_users 关联，会碰到数据倾斜的问题。

解决方法1： user_id为空的不参与关联

select * from log a

  join bmw_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 bmw_users b

  on case when a.user_id is null then concat(‘dp_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上 ,解决数据倾斜问题。附上hadoop通用关联的实现方法（关联通过二次排序实现的，关联的列为parition key,关联的列 c1 和表的 tag 组成排序的 group key ,根据 parition key 分配 reduce 。同一 reduce 内根据 group key 排序）

场景：一张表 s8_log，每个商品一条记录，要和商品表关联。但关联却碰到倾斜的问题。s8_log 中有字符串商品 id，也有数字的商品 id。字符串商品 id 类型是 string 的，但商品中的数字 id 是 bigint 的。问题的原因是把 s8_log 的商品 id 转成数字 id 做 Hash（数字的 Hash 值为其本身，相同的字符串的 Hash 也不同）来分配 Reducer，所以相同字符串 id 的 s8_log，都到一个 Reducer 上了。

解决方法：把数字类型转换成字符串类型

select * from s8_log a

  left outer join r_auction_auctions b

  on a.auction_id = cast(b.auction_id as string);

多表 union all 会优化成一个 job。比如推广效果表要和商品表关联，效果表中的 auction id 列既有商品 id，也有数字 id，和商品表关联得到商品的信息。

以下的hive sql性能会比较好

select * from effect a

join (

  select auction_id as auction_id from auctions

  union all

  select auction_string_id as auction_id from auctions

) b

on a.auction_id = b.auction_id;

结论：比分别过滤数字 id，字符串 id 然后分别和商品表关联性能要好。 这样写的好处,1个 MR 作业,商品表只读取一次，推广效果表只读取一次。把这个 sql 换成 MR 代码的话，map 的时候，把 a 表的记录打上标签 a ，商品表记录每读取一条，打上标签 t，变成两个<key,value> 对，<t,数字id,value>，<t,字符串id,value>。所以商品表的 HDFS 读只会是一次。

hive 对UNION ALL的优化的特性：对union all优化只局限于非嵌套查询

对于如下 SQL：

select * from (

  select * from t1

    group by c1,c2,c3

  union all 

  select * from t2

    group by c1,c2,c3) t3

group by c1,c2,c3;

存在的问题:从业务逻辑上说，子查询内的 group by 功能与外层的 group by 重复,除非子查询内有 count(distinct)。 调整后的sql:

select * from (

  select * from t1

  union all

  select * from t2) t3

group by c1,c2,c3;

结论：经过测试，并未出现 union all 的 hive bug，数据是一致的。MR 的作业数由3减少到1。 t1 相当于一个目录，t2 相当于一个目录，对map reduce程序来说，t1，t2 可以做为 map reduce 作业的 mutli inputs。这可以通过一个 map reduce 来解决这个问题。Hadoop的计算框架，不怕数据多，怕作业数多。

select * from (select * from t1

  group by c1,c2,c3

  union all 

select c1,c2,c3,count(disintct c4) from t2

  group by c1,c2,c3) t3

  group by c1,c2,c3;

由于子查询里头有count(distinct )操作，直接去group by将达不到业务目标。这时采用临时表消灭count(distinct)作业不但能解决倾斜问题，还能有效减少jobs。

如以下例子:

insert t4 select c1,c2,c3,c4 from t2 group by c1,c2,c3;

select c1,

      c2,

      c3,

      sum(income),

      sum(uv)

from (select c1,c2,c3,income,0 as uv from t1

     union all

     select c1,c2,c3,0 as income,1 as uv from t2) t3 group by c1,c2,c3;

Job数是2，减少一半，而且两次mr比count(distinct)效率更高。

先生成临时表，再 union all 还是写嵌套查询，这是个问题。比如以下例子：

select * from (

  select * from t1

  union all

  select * from t4

  union all

  select * from t2

    join t3 on t2.id = t3.id) x

group by c1,c2;

这个会有 4 个 jobs。假如先 join 生成临时表的话 t5，然后 union all，会变成 2 个 jobs。

insert overwrite table t5

select *

  from t2

  join t3

  on t2.id = t3.id;

select * from (t1 union all t4 union all t5);

结论： Hive 在 union all 优化上可以做得更智能（把子查询当做临时表），这样可以减少开发人员的负担。如果写MR程序这不是问题，就是 multi inputs。

使用 map join 解决小表(记录数少)关联大表的数据倾斜问题，这个方法使用的频率非常高，但如果小表很大，大到map join会出现bug或异常，这时就需要特别的处理。云瑞和玉玑提供了非常给力的解决方案。 以下例子:

select * from log a

  left outer join members b

  on a.memberid = b.memberid;

members 表有 600w+ 的记录，把 members 分发到所有的 map 上也是个不小的开销，而且 map join 不支持这么大的小表。如果用普通的 join，又会碰到数据倾斜的问题。

解决方法：

select /*+mapjoin(x)*/* from log a

  left outer join (

    select  /*+mapjoin(c)*/d.*

      from ( select distinct memberid from log ) c

      join members d

      on c.memberid = d.memberid

    ) x

  on a.memberid = b.memberid;

假如，log里memberid有上百万个，这就又回到原来map join问题。所幸，每日的会员uv不会太多，有交易的会员不会太多，有点击的会员不会太多，有佣金的会员不会太多等等。所以这个方法能解决很多场景下的数据倾斜问题。

HIVE下通用的数据倾斜解决方法,double被关联的相对较小的表，这个方法在mr的程序里常用。

Select * from log a

Left outer join (select /*+mapjoin(e)*/

memberid, number 

  From members d

  Join num e

  ) b

On a.memberid= b.memberid

And mod(a.pvtime,30)+1=b.number;

Num表只有一列number，有30行，是1,30的自然数序列。把member表膨胀成N份(基于倾斜程度做一个合适的选择)，然后把log数据根据memberid和pvtime分到不同的reduce里去，这样可以保证每个reduce分配到的数据可以相对均匀。就目前测试来看，使用mapjoin先过滤的方案性能稍好。后面的方案适合在map join无法解决问题的情况下。

设想，把如下的优化方案做成通用的hive优化方法

采样log表，哪些memberid比较倾斜，得到一个结果表tmp1。由于对计算框架来说，所有的数据过来，他都是不知道数据分布情况的，所以采样是并不可少的。Stage1

数据的分布符合社会学统计规则，贫富不均。倾斜的key不会太多，就像一个社会的富人不多，奇特的人不多一样。所以tmp1记录数会很少。把tmp1和members做map join生成tmp2,把tmp2读到distribute file cache。这是一个map过程。Stage2

map读入members和log，假如记录来自log,则检查memberid是否在tmp2里，如果是，输出到本地文件a,否则生成<memberid,value>的key,value对，假如记录来自member,生成<memberid,value>的key,value对，进入reduce阶段。Stage3

最终把a文件，把Stage3 reduce阶段输出的文件合并起写到hdfs。

这个方法在hadoop里应该是能实现的。Stage2是一个map过程，可以和stage3的map过程可以合并成一个map过程。 这个方案目标就是：倾斜的数据用mapjoin，不倾斜的数据用普通的join，最终合并得到完整的结果。用hive sql写的话，sql会变得很多段，而且log表会有多次读。倾斜的key始终是很少的，这个在绝大部分的业务背景下适用。那是否可以作为hive针对数据倾斜join时候的通用算法呢？

场景:比如要计算店铺的uv，还有要计算页面的uv,pvip。

方案一：

insert overwrite table shop_uv

Select shopid,count(distinct uid)

From log group by shopid;

insert overwrite table page_uv

Select pageid, count(distinct uid),

From log group by pageid;

由于存在数据倾斜问题，这个结果的运行时间是非常长的。

方案二：

From log

Insert overwrite table t1 (type='1')

Select shopid as dim

Group by shopid ,acookie

Insert overwrite table t1 (type='2')

select pageid as dim Group by pageid,acookie;

店铺uv:

insert overwrite table shop_uv Select dim,sum(1)

From t1

Where type ='1'

Group by dim ;

页面uv:

insert overwrite table page_uv Select dim,sum(1)

From t1

Where type ='2'

Group by dim ;

这里使用了multi insert的方法，有效减少了hdfs读，但multi insert会增加hdfs写，多一次额外的map阶段的hdfs写。使用这个方法，可以顺利的产出结果。缺点是 存在重复IO，并且作业数据较多。

方案三：

Insert into t1

Select type,type_name

From (

Select 'page' as type,

  Pageid as type_name,

  Uid

From log 

Union all 

Select 'shop' as type,

  Shopid as type_name,

  Uid

From log ) y

Group by type,type_name,uid;

汇总得到page和店铺uv

Insert into t2

Select type,type_name,sum(1)

From t1

Group by type,type_name;

分出page和店铺的uv

From t2

Insert into t3

Select type,type_name,uv

Where type='page'

Select type,type_name,uv

Where type='shop';

比较: 最终得到两个结果表t3,页面uv表，t4,店铺结果表。从io上来说，log一次读。但比方案2少次hdfs写（multi insert有时会增加额外的map阶段hdfs写）。作业数减少1个到3，有reduce的作业数由4减少到2，第三步是一个小表的map过程，分下表，计算资源消耗少。但方案2每步都是大规模的去重汇总计算。

这个优化的主要思路是，map reduce作业初始化的时间比较长，既然起来了，让他多干点活，顺便把页面按uid去重的活也干了，省下log的一次读和作业的初始化时间，但增加了本地磁盘读写。效率提升较多。 这个方案适合平级的不需要逐级向上汇总的多粒度uv计算，粒度越多，比较节省资源，比较通用。如果说，把两个uv分成两个脚本，分别并行计算，那就不敢保证这个方案比分脚本的方案快了，但这个更省资源。

场景:

比如4个维度，a,b,c,d，分别计算a,b,c,d,uv;a,b,c,uv;a,b,uv;a;uv,total uv4个结果表。这可以用多粒度uv计算的方法，这里由于uv场景的特殊性，多粒度，逐层向上汇总，就可以使用一次排序，所有uv计算受益的计算方法。 目前mm_log日志一天有25亿+的pv数，要从mm日志中计算uv，与ipuv,一共计算五个层次的结果表

(memberid,siteid,adzoneid,province,uv,ipuv） R_TABLE_4

(memberid,siteid,adzoneid,uv,ipuv） R_TABLE_3

(memberid,siteid,uv,ipuv) R_TABLE_2

(memberid,uv,ipuv) R_TABLE_1

(uv,ipuv) R_TABLE

第一步：

按memberid,siteid,adzoneid,province,使用group去重,产生临时表，对cookie,ip 打上标签放一起，一起去重，临时表叫T_4;

Select memberid,siteid,adzoneid,province,type,user

From(

Select memberid,siteid,adzoneid,province,'a' type ,cookie as user from mm_log where ds=20101205

Union all

Select memberid,siteid,adzoneid,province,'i' type ,ip as user from mm_log where ds=20101205

) x group by memberid,siteid,adzoneid,province,type,user ;

第二步：

排名,产生表T_4_NUM.Hadoop最强大和核心能力就是parition 和 sort.按type，acookie分组， Type，acookie，memberid,siteid,adzoneid,province排名。

Select * ,

row_number(type,user,memberid,siteid,adzoneid ) as adzone_num ,

row_number(type,user,memberid,siteid ) as site_num,

row_number(type,user,memberid ) as member_num, 

row_number(type,user ) as total_num

from (select * from T_4 distribute by type,user sort by type,user, memberid,siteid,adzoneid ) x;

这样就可以得到不同层次粒度上user的排名，相同的user id在不同的粒度层次上，排名等于1的记录只有1条。取排名等于1的做sum，效果相当于Group by user去重后做sum操作。

第三步：

不同粒度uv统计，先从最细粒度的开始统计，产生结果表R_TABLE_4,这时，结果集只有10w的级别。

如统计memberid,siteid,adzoneid,provinceid粒度的uv使用的方法就是

Select memberid,siteid,adzoneid, provinceid,

sum(case when type ='a' then cast(1) as bigint end ) as province_uv ,

sum(case when type ='i' then cast(1) as bigint end ) as province_ip ,

sum(case when adzone_num =1 and type ='a' then cast(1) as bigint end ) as adzone_uv ,

sum(case when adzone_num =1 and type ='i' then cast(1) as bigint end ) as adzone_ip ,

sum(case when site_num =1 and type ='a' then cast(1) as bigint end ) as site_uv ,

sum(case when site_num =1 and type ='i' then cast(1) as bigint end ) as site_ip ,

sum(case when member_num =1 and type ='a' then cast(1) as bigint end ) as member_uv ,

sum(case when member_num =1 and type ='i' then cast(1) as bigint end ) as member_ip ,

sum(case when total_num =1 and type ='a' then cast(1) as bigint end ) as total_uv ,

sum(case when total_num =1 and type ='i' then cast(1) as bigint end ) as total_ip ,

from T_4_NUM

group by memberid,siteid,adzoneid, provinceid ;

广告位粒度的uv的话，从R_TABLE_4统计，这是做10w级别记录数的统计

Select memberid,siteid,adzoneid,sum(adzone_uv),sum(adzone_ip)

From R_TABLE_4 

Group by memberid,siteid,adzoneid；

memberid,siteid的uv计算 ，

memberid的uv计算,

total uv 的计算也都从R_TABLE_4汇总。

场景:

经常遇到几个结果表都是出自一个源表，但是要取的字段不一样。

比如，需要从处罚表得到当天处罚的记录明细，和被处罚的会员的累积处罚积分。

如果采用通常的做法，得到明细数据查询一次处罚表，得到汇总数据再查一次处罚表，这样需要扫两次，当表很大的时候这个代价会非常大。 用multi_select可以扫描一次就得到这两个结果。

代码如下： set hive.merge.mapfiles = false;

from(select * from s_bmw_punish_award_result where pt='$env.lastPartition' and be_punished_type = 0) a

insert overwrite table m_mid_punish_result

partition (pt='$env.lastPartition',child='01')

select 

'$formatDate' as gmt_create,

id,

punish_award_rule_id,

be_punished_type,

from_id,

substr(create_date,1,10) as create_date,

point_range,

case when status=2 then substr(gmt_modified,1,10) end as cancel_date,

status,

be_punished_id,

operator

where dateCompare(create_date ,'$env.date',0)=0 

insert overwrite table m_mid_punish_result

partition (pt='$env.lastPartition',child='02')

select 

'$formatDate' as gmt_create,

id,

punish_award_rule_id,

be_punished_type,

from_id,

substr(create_date,1,10) as create_date,

point_range,

substr(gmt_modified,1,10)as cancel_date,

status,

be_punished_id,

operator

where dateCompare(gmt_modified,'$env.date',0)=0 

  and status=2

  and dateCompare(create_date,'$env.date',0)<>0 

insert overwrite table m_mid_punish_result_3

partition (pt='$env.lastPartition')

select be_punished_id,

sum(case when point_is_effect = 1 then abs(cast(point as double )) else 0.00 end ) as sum_point

where dateCompare(create_date,'20090915',0)>=0

and dateCompare(create_date,'$env.date',1)<0

group by be_punished_id

insert overwrite table t_afan_auc_prop3

partition(pt='20120717000000')

select auction_id,property_id,value_id 

from r_auctions_properties 

where pt='20120717000000' 

and 

(

(property_id = 1626130 and value_id = 46276) or

(property_id = 33510 and value_id = 31533293) or

(property_id = 33510 and value_id = 119834) or

(property_id = 33510 and value_id = 119831) or

(property_id = 33448 and value_id = 118432) or

(property_id = 33448 and value_id = 21039) or

 xxxx

)

and 里面的or 条件有170多个

这个分区有80G的数据，120亿条数据，单独扫描只需要不到5分钟，但是运行这个sql确需要1个多小时，主要原因是因为这个and条件hive在生成执行计划时产生了一个嵌套层次很多的算子。

property_id 和 value_id 都是string 类型的字段

解决方案：

（1）property_id、value_id 的值对搞成一个小表，然后通过一次mapjoin

（2）写个udf，把这些预设值读取进去，udf来完成这个and数据过滤操作

有些简单的sql语句，没有reduce产生，输入数据很大，有很多map，每个map都会有结果文件产生，输出文件数会很大。

 （1）容易导致quota超

 （2）会运行merge job，这个merge job用来合并小文件。

merge job有两种实现：

 （1）hive.mergejob.maponly为true，使用只有map的MR job来合并小文件，使用了org.apache.hadoop.hive.ql.io.CombineHiveInputFormat，

    问题是速度可能会很慢，1.1.4以后的版本包括1.1.4已经解决了CombineHiveInputFormat慢的问题。

 （2）hive.mergejob.maponly为false，使用有reduce的MR job来合并小文件。

规避方法：

（1）加上distribute by rand(12345)或者是distribute by 某个字段，如果结果很小，可以减少reduce数。

（2）没有加上distribute by，但是第二步的merge很慢，set hive.mergejob.maponly=false;

example:

 select * from r_auction_auctions 

 where  pt='20111205000000' and if_online=0

 and auction_id = 13033941674; 

改为

 set mapred.reduce.tasks=5;

 select * from r_auction_auctions 

 where  pt='20111205000000' and if_online=0

 and auction_id = 13033941674 

 distribute by rand(12345);

现象：开发人员写的很多sql很简单，只需要map无需reduce就能完成任务，hive解析sql生成map only的作业，map数很多时，每个map都有至少一个输出结果，在这个MapReduce作业完成后，会起一个合并小文件的Map only作业，它会有一些map产生，各个map使用CombineHiveInputFormat来读取多个小文件，写到一个大文件中。 

问题：合并小文件的job启动以及map运行很慢，从map的日志来看，比如job_201108301259_4048225

该job是一个合并小文件的作业，该作业输入输出数据量很小，如下：

Counter Map Reduce Total 

File Systems HDFS bytes read 94,056,452 0 94,056,452

HDFS bytes written 93,985,955 0 93,985,955

Job 运行的总的时间：Finished At: 8-Nov-2011 04:04:44 (45mins, 4sec)

该合并job有3个task，其中一个task运行了43分钟！！！！

task_201108301259_4048225_m_000000

8/11 03:20:46 8/11 04:04:38 (43mins, 51sec)

Task的日志，截取了一部分：

2011-11-08 03:21:30,764 INFO org.apache.hadoop.hive.ql.exec.TableScanOperator: 0 forwarding 100000 rows

2011-11-08 03:21:30,764 INFO ExecMapper: ExecMapper: processing 100000 rows: used memory = 14294320

2011-11-08 03:44:49,437 INFO org.apache.hadoop.hive.ql.exec.MapOperator: 2 forwarding 1000000 rows

2011-11-08 03:44:49,457 INFO org.apache.hadoop.hive.ql.exec.TableScanOperator: 0 forwarding 1000000 rows

2011-11-08 03:44:49,457 INFO ExecMapper: ExecMapper: processing 1000000 rows: used memory = 42485208

2011-11-08 04:04:34,807 INFO org.apache.hadoop.hive.ql.exec.MapOperator: 2 forwarding 2000000 rows

2011-11-08 04:04:34,825 INFO org.apache.hadoop.hive.ql.exec.TableScanOperator: 0 forwarding 2000000 rows

2011-11-08 04:04:34,826 INFO ExecMapper: ExecMapper: processing 2000000 rows: used memory = 33694880

2011-11-08 04:04:38,144 INFO org.apache.hadoop.hive.ql.exec.MapOperator: 2 finished. closing... 

其中有两段，前后两条日志的打印时间差在20分钟以上！！！

原因：这个map需要读取很多小文件，CombineHiveInputFormat之前的做法是依次从namenode获取每个小文件的location信息，对每个小文件跟namenode来一次RPC调用，namenode现在数据量很大，请求很多，所以有时候会花费比较长的时间。

结论：大量小文件对namenode来说是一种硬伤，对MapReduce job来说也是一个噩梦。

解决方法：

（1）业务层面，即sql语句，加入强制distribute by，让它产生reduce，并且把reduce数设小，对namenode和MapReduce job来说都是非常nice的。

未修改前的sql，比如是：

from (select thedate ,toprank_id ,if_test ,object_id ,idx ,idx_rank ,idx_last from rpt_extended_toprank t1 where ds = 20111107 and   toprank_conf_type_id = 5+0 ) t1 

insert overwrite table rpt_topranks_v3_tz partition(ds = 20111107,toprank_id_pt = 11) 

select thedate ,toprank_id ,if_test ,object_id ,idx ,idx_rank ,idx_last where toprank_id = 11+0;

修改后的sql是：

set mapred.reduce.tasks=5;

from (select thedate ,toprank_id ,if_test ,object_id ,idx ,idx_rank ,idx_last from rpt_extended_toprank t1 where ds = 20111107 and  toprank_conf_type_id = 5+0 ) t1 

insert overwrite table rpt_topranks_v3_tz partition(ds = 20111107,toprank_id_pt = 11) 

select thedate ,toprank_id ,if_test ,object_id ,idx ,idx_rank ,idx_last where toprank_id = 11+0 

distribute by rand();

（2）hive 层面，由于目前这些修改没办法一下子上线，所以推荐在业务层面改动。

 （2.1）修改CombineHiveInputFormat获取文件和split信息，一次RPC调用获取整个目录下面的所有文件信息，已完成，待上线。

 （2.2）如果没有数据，不产生空文件，未完成。

两种解决方法对比：在sql加入distribute by设置合适reduce数相对第二种要好些，因为前者只需要起一道MapReduce作业，调度等待的时间要小。

如果分组很多，第二个MR的Reduce只有一个，需要运行很长时间才能得到结果。

select mid,url,count(1) as cnt from (

  select * from r_atpanel_log where pt='20111017000000' and pagetype='normal'

) subq

group by mid,url

order by cnt desc

limit 15;

可以优化一下，先过滤很多小分组

select * from (

  select mid,url,count(1) as cnt from (

      select * from r_atpanel_log where pt='20111017000000' and pagetype='normal'

  ) subq

  group by mid,url

) subq2

where cnt >100

order by cnt desc

limit 15;