五、大表 Join 大表优化
如果上述 mapjoin 中小表 dim_seller 很大呢？比如超过了 1GB 的大小？这种就是大表join 大表的问题 。
这类问题相对比较复杂，我们首先引入具体的问题场景，然后基于此介绍各种优化方案。
5.1 问题场景
我们先假设一个问题场景：
A 表为一个汇总表，汇总的是卖家买家最近 N 天交易汇总信息，即对于每个卖家最近 N 天，其每个买家共成交了多少单、总金额是多少，我们这里 N 先只取 90 天，汇总值仅取成交单数 。A 表的字段有：buyer_id 、seller_id 和 pay_cnt_90d 。
B 表为卖家基本信息表，其中包含卖家的一个分层评级信息，比如把卖家分为 6 个级别： S0、S1、S2、S3、S4、S5、S6 。
要获得的结果是每个买家在各个级别卖家的成交比例信息，比如：
某买家 S0:10%; S1:20%; S2:20%; S3:10%; S4:20%; S4:10%; S5:10%。
B表的字段有： seller_id 和 s_level。
正如 mapjoin 中的例子一样，我们的第一反应是直接 join 表并统计：
select m.buyer_id ,sum(pay_cnt_90d) as pay_cnt_90d ,sum(case when m.s_level=O then pay_cnt_90d end) as pay_cnt_90d_s0 ,sum(case when m.s_level=l then pay_cnt_90d end) as pay_cnt_90d_sl ,sum(case when m.s_level=2 then pay_cnt_90d end) as pay_cnt_90d_s2 ,sum(case when m.s level=3 then pay cnt 90d end) as pay_cnt_90d_s3 ,sum(case when m.s_level=4 then pay_cnt_90d end) as pay_cnt_90d_s4 ,sum(case when m.s_level=S then pay_cnt_90d end) as pay_cnt_90d_s5 from ( select a.buyer_id,a.seller_id,b.s_level,a.pay_cnt_90d from ( select buyer_id ,seller_id,pay_cnt_90d from table A ) a join ( select seller_id,s_level from table B ) b on a.seller_id=b.seller_id ) m group by m.buyer_id
但是此 SQL 会引起数据倾斜，原因在于卖家的二八准则。某些卖家 90 天内会有几百万甚至上千万的买家，但是大部分卖家 90 天内的买家数目并不多， join table_A 和table_B 的时候 ODPS 会按照 Seller_id 进行分发， table_A 的大卖家引起了数据倾斜。
但是本数据倾斜问题无法用 mapjoin table_B 解决，因为卖家有超过千万条、文件大小几个GB ，超过了 mapjoin 表最大 1GB 的限制。
方案 1：转化为 mapjoin
大表无法直接mapjoin，那么是否可以间接呢？实际上此思路有两种途径：限制行和限制列。
限制行: 不需要join B全表，只需要join其在A表中存在的。对于本问题场景，就是过滤掉 90 天内没有成交的卖家。
限制列: 只取需要的字段。
select m.buyer_id ,sum(pay_cnt_90d) as pay_cnt_90d ,sum(case when m.s_level=O then pay_cnt_90d end) as pay_cnt_90d_s0 ,sum(case when m.s_level=l then pay_cnt_90d end) as pay_cnt_90d_sl ,sum(case when m.s_level=2 then pay_cnt_90d end) as pay_cnt_90d_s2 ,sum(case when m.s level=3 then pay cnt 90d end) as pay_cnt_90d_s3 ,sum(case when m.s_level=4 then pay_cnt_90d end) as pay_cnt_90d_s4 ,sum(case when m.s_level=S then pay_cnt_90d end) as pay_cnt_90d_s5 from ( select /*+mapjoin(b)*/ a.buyer_id,a.seller_id,b.s_level,a.pay_cnt_90d from ( select buyer_id ,seller_id,pay_cnt_90d from table_A ) a join ( select b0.seller id,s_level from table_B b0 join (select seller_id from table_A group by seller_id) a0 on b0.seller_id=a0.seller_id ) b on a.seller_id=b.seller_id ) m group by m.buyer_id
此方案在一些情况下可以起作用，但很多时候还是无法解决上述问题，因为大部分卖家尽管 90 买家不多 ，但还是有一些的，过滤后的 B
表仍然很大。
方案 2：join 时用 case when 语句
应用场景为: 倾斜的值是明确的而且数量很少，比如null值引起的倾斜。
将这些引起倾斜的值随机分发到Reduce，其主要核心逻辑在于 join 时对这些特殊值concat 随机数,从而达到随机分发的目的。核心逻辑如下：
Select a.user_id,a.order_id,b.user_id From table_a a Join table_b b On (case when a.user_id is null then concat ('hive' ,rand()) else a.user_id end)=b.user_id
Hive已对此进行了优化，不需要修改SQL，只需要设置参数；比如 table_B 的值 “0” 和 “1” 引起倾斜，只需要如下设置：
set hive.optimize.skewinfo=table_B:(seller_id)[("0")("1")]; set hive.optimize.skewjoin=true;
但是方案二还是不能解决上述问题，因为倾斜的卖家大量存在而且动态变化。
方案 3：倍数B表，再取模join
通用方案
是建立一个numbers表,其值只有一列int行，比如从1到10(具体根据倾斜程度确定)，然后放大B表10倍，再取模join。
select m,buer_id ,sum(pay_cnt_90d) as pay_cnt_90d ,sum(case when m.s_level=O then pay_cnt_90d end) as pay cnt 90d so ,sum(case when m.s_level=l then pay cnt 90d end) as pay cnt 90d_sl ,sum(case when m.s_level=2 then pay_cnt_90d end) as pay_cnt_90d s2 ,sum(case when m.s_level=3 then pay_cnt_90d end) as pay_cnt_90d_s3 ,sum(case when m.s_level=4 then pay_cnt_90d end) as pay cnt 90d s4 ,sum(case when m.s level=S then pay cnt 90d end) as pay cnt 90d s5 from ( select a.buyer_id,a.seller_id,b.s_level,a.pay_cnt_90d from ( select buyer_id,seller_id,pay_cnt_90d from table_A ) a JOin ( select /*+mapjoin(members)*/ seller_id,s_level,member from table_B join numbers ) b on a.seller_id=b.seller_id and mod(a.pay_cnt_90d,10)+1=b.number ) m group by m.buyer_id
思路核心在于：既然按照seller_id分发会倾斜，那么再人工增加一列进行分发，这样之前倾斜的值的倾斜程度会减少为原来的1/10。可以通过配置numbers表修改放大倍数来降低倾斜程度，但弊端就是B表会膨胀N倍。
专有方案
通用方案思路是把B表的每条数据都放大了相同的倍数，实际上只需要把大卖家放大倍数即可。
首先需要知道大卖家的名单，即先建立一个临时表动态存放每日最新的大卖家（比如dim_big_seller），同时此表的大卖家要膨胀预先设定的倍数（比如1000倍）。
在A表和 B表中分别新建一个 join 列，其逻辑为：如果是大卖家，那么 concat 一个随
机分配正整数（0到预定义的倍数之间，本例为0~1000 ）；如果不是，保持不变。
相比通用方案，专用方案的运行效率明显好了很多，因为只是将B表中大卖家的行数放大了 1000 倍，其他卖家的行数保持不变，但同时也可以看到代码也复杂了很多，而且必须首先建立大卖家表。
方案 4：动态一分为二
实际上方案 2 和 3 都用到了一分为二的思想，但是都不彻底，对于 mapjoin 不能解决的
问题，终极解决方案就是动态一分为 ，即对倾斜的键值和不倾斜的键值分开处理，不倾
斜的正常 join 即可，倾斜的把它们找出来然后做 mapjoin ，最后 union all 其结果即可。
但是此种解决方案比较麻烦，代码会变得复杂而且需要一个临时表存放倾斜的键值。
-- 对于 90 天买家数超过 10000 的卖家直接 map join ，对于其他卖家正常 join 即可 select m.buyer_id ,surn(pay_cnt_90d) as pay_cnt_90d ,surn(case when rn.s_level=O then pay_cnt_90d end) as pay_cnt_90d_s0 ,surn(case when rn.s_level=l then pay_cnt_90d end) as pay_cnt_90d_sl ,surn(case when rn.s_level=2 then pay_cnt_90d end) as pay_cnt_90d_s2 ,surn(case when rn.s_level=3 then pay_cnt_90d end) as pay_cnt_90d_s3 ,surn(case when rn.s_level=4 then pay_cnt_90d end) as pay_cnt_90d_s4 ,surn(case when rn.s_level=S then pay_cnt_90d end) as pay_cnt_90d_s5 from ( select a.buyer_id,a.seller_id,b.s_level,a.pay_cnt_90d from ( select buyer_id,seller_id,pay_cnt_90d from table_A ) a join ( select seller_id ,a.s_level from table_A a left outer join tmp_table_B b on a.user_id = b.seller_id where b.seller_id is null ) b on a.seller id=b.seller id union all select /*+mapjoin(b)*/ a.buyer_id,a.seller_id,b.s_level,a.pay_cnt_90d from select buyer_id,seller_id,pay_cnt_90d from table A ) a join select seller_id,s_level from table B ) b on a.seller id=b.seller id ) m group by m.buyer_id ) m group by m.byer_id
总结起来，方案 1、2 以及方案 3 中的通用方案不能保证解决大表 join 大表问题，因为它们都存在种种不同的限制和特定的使用场景。
而方案 3 的专用方案和方案 4 是比较推荐的优化方案，但是它们都需要新建一个临时表来存放每日动态变化的大卖家 。
相对方案 4 来说，方案 3 的专用方案不需要对代码框架进行修改，但是 B 表会被放大，所以一定要是维度表，不然统计结果会是错误的 。 方案 4 的解决方案最通用，自由度最高，但是对代码的更改也最大，甚至需要更改代码框架，可作为终极方案来使用。
我是「云祁」，一枚热爱技术、会写诗的大数据开发猿。
