MaxCompute JOIN优化小结

摘要： Join是MaxCompute中最基本的语法，但由于数据量和倾斜问题，非常容易出现性能问题。一般情况下，join产生的问题有两大类： 数据倾斜问题：join会将key相同的数据分发到同一个instance上处理，如果某个key上的数据量特别多则会导致该instance处理时间比其他instance...

原文地址：http://click.aliyun.com/m/43804/

Join是MaxCompute中最基本的语法，但由于数据量和倾斜问题，非常容易出现性能问题。一般情况下，join产生的问题有两大类：
  ● 数据倾斜问题：join会将key相同的数据分发到同一个instance上处理，如果某个key上的数据量特别多则会导致该instance处理时间比其他instance处理时间长，这就是我们常说的数据倾斜，这也是join计算性能问题的罪魁祸首；
  ● 数据量问题：关联的两表基本没有热点问题，但两个表数据量都非常大同样会影响性能，比如记录数达几十亿条，如商品表、库存表等；
      
虽然MaxCompute中提供了一些通用的优化算法，但从业务角度解决性能问题往往更精确，更有效。对于MaxCompute sql优化，在云栖社区上已经有比较多的经验积累，本文主要对join产生的性能问题以及解法做些总结。

不同数据类型key关联

例子

       浏览IPV日志以商品id关联商品表，假设日志表的商品id字段是string类型，商品表中的商品id是bigint类型，那么在关联中，关联key会全部转换成double类型进行比较，设想由于埋点问题日志表中的商品id存在很多非数值的脏数据，那么转换成double后值都变为NULL或者截取前面的数值，关联时就会产生数据倾斜问题，更严重的会造成数据错误。

解法

      关联时手工进行数据格式转换，在这种情况下一般将bigint类型key转换成string类型。
select a.*
from ipv_log_table a
left outer join item_table b
on a.item_id = cast(b.item_id as string)
      
思考下，假如反过来将string类型转换成bigint、假如IPV日志表中的商品id大部分为无效值（比如0）、又假如IPV日志表中没有无效值但是有热点key会有什么问题呢？下面的例子会解答这些问题。

小表join大表

         Join中存在小表，一般这个小表在100M以内，可以用mapjoin，避免分发引起的长尾。拿上面的例子来说，假如商品表数据量只有几万条记录（这里只是打个比方，现实业务中商品表一般都是非常庞大的），但是IPV日志表中的商品id 80%值为0的无效值，且记录数有几十亿，如果采用上述SQL写法，数据倾斜是显而易见的，但利用mapjoin可以有效解决这个问题：
select /*+ MAPJOIN(b) */a.*
from ipv_log_table a
left outer join item_table b
on a.item_id = cast(b.item_id as string)

mapjoin原理 
    
       把小表广播传递到所有的Join Task Instance上面，然后直接和大表做Hash Join，简单的说就是将join操作提前到map端，而非reduce端。

mapjoin使用注意点
  ● 小表在left outer join 时只能是右表， right outer join 时只能是左表， inner join 时无限制，full outer join不支持mapjoin；
  ●  mapjoin最多只支持8张小表，否则会报语法错误；
  ● mapjoin所有小表内存限制不能超过2GB，默认为512M；
  ● mapjoin支持小表为子查询；
  ● mapjoin支持不等值连接或者使用or连接多个条件；

大表join大表存在无效值
   
   在小表join大表时我们已经了解到通过mapjoin将小表全部加载到map端可以解决倾斜问题，但假如‘小表‘不够小，mapjoin失效的时候该怎么办呢？同样以本文第一个场景为例，IPV日志表中80%商品id都为无效值0（目前MaxCompute底层已经针对NULL值进行优化，已经不存在倾斜问题了），这时关联十几亿量级商品表那就是个灾难。

解法1-分而治之：
      
我们可以事先知道无效值是不可能关联出结果的，而且完全不需要参与关联，所以可以将无效值与有效值数据分开处理：

select a.visitor_id
      ,b.seller_id
from (
      select
      from ipv_log_table
      where item_id > 0
) a
left outer join item_table b
on a.item_id = b.item_id

union all

select a.visitor_id
      ,cast(null as bigint) seller_id
from ipv_log_table
where item_id = 0

解法2-随机值打散：
      
我们也可以以随机值代替NULL值作为join的key，这样就从原来一个reduce来处理倾斜数据变成多reduce并行处理，因为无效值不参与关联，即使分发到不同reduce，也不会影响最终计算结果：

select a.visitor_id
      ,b.seller_id
from ipv_log_table a
left outer join item_table b
on if(a.item_id > 0, cast(a.item_id as string), concat('rand',cast(rand() as string))) = cast(b.item_id as string)

解法3-转化为mapjoin：

      虽然商品表有十几亿条记录，不能直接通过mapjoin来处理，但在实际业务中，我们知道一天内用户访问的商品数是有限的，在业务中尤为明显，基于此我们可以通过一些处理转换成
mapjoin：

select /*+ MAPJOIN(b) */
       a.visitor_id
      ,b.seller_id
from ipv_log_table a
left outer join (
   select /*+ MAPJOIN(log) */
         itm.seller_id
        ,itm.item_id
   from (
         select item_id
         from ipv_log_table
         where item_id > 0
         group by item_id
   ) log join item_table itm
   on log.item_id = itm.item_id
) b
on a.item_id = b.item_id

解法对比

       解法1和解法2是通用解决方案，对于解法1，日志表被读取两次，而解法2中只需读取一次，另外任务数解法2也是少于解法1的，所以总的来看解法2是优于解法1的。解法3是基于一定的假设，随着业务发展或者某些特殊情况下假设可能失效（比如一些爬虫日志，可造成访问商品数接近全量），这会导致mapjoin失效，所以在使用过程中要根据具体情况来评估。

一个古老的例子

       最后要讲一个古老的优化case，虽然历史比较久远，目前已没有相关问题，但优化思路值得借鉴。情况是这样的，历史上并存过两套商品维表，一份主键是字符串id，新的商品表也就是目前在使用的主键是数字id，字符串id和数字id做了映射，存在商品表中的两个字段中，所以在使用中需要分别过滤数字id、字符串id然后分别和商品表关联，最后union起来得到最终结果。
      思考下如果换成下面的优化思路是不是更优呢？

select ...
from ipv_log_table a
join (
      select auction_id as auction_id
      from auctions
      union all
      select auction_string_id as auction_id
      from auctions
      where auction_string_id is not null
) b
on a.auction_id = b.auction_id


答案是肯定的，可以看到优化后商品表读取从2次降为1次，IPV日志表同样，另外MR作业数也从2个变为1个。

总结

       对于MaxCompute sql优化最有效的方式是从业务的角度切入，并能够将MaxCompute sql转化为mapreduce程序来解读。本文针对join中各种场景优化都做了一些梳理，现实情况很可能是上述多场景的组合，这时候就需要灵活运用相应的优化方法，举一反三。

识别以下二维码，阅读更多干货