目录
前言
一、场景描述
二、常见的优化方法
2.1 Mapjoin
2.2 特殊值/空值打散
2.3 热点值打散，副表呈倍数扩散
2.4 热点数据单独处理/SkewJoin
2.5 方案总结
三、Distmapjoin
3.1 核心思路
3.2 代码实现
3.3 真实效果
四、方案总结
文章主要是介绍在支付宝支付数据链路改造升级过程中，针对数据倾斜的优化实践方案，在解决数据倾斜问题的同时，还能兼顾更优的计算性能。
一、场景描述
数据倾斜可能发生在join，group by，count distinct等环节，但本质上其实都类似，即因为数据重分发或重分布等原因，导致大部分数据仅分发到少数几个计算节点上。以ODPS场景为例，少数几个Fuxi Instance处理的数据量，远大于同一环节的其他Instance处理的数据量，并伴有明显的长尾现象。
典型的案例是在淘宝双十一场景中，交易订单明细大表需要关联商家信息维表以补全商家信息，在数据关联处理中，同一个商家对应的交易订单个维表对应商家信息，将根据卖家ID shuffle至同一个数据处理节点上。由于TOP商家在大促中产生的交易单量远大于普通商家， 从而导致大量的数据集中到一台或者几台机器上计算，这些数据的计算速度将远远低于平均计算速度，导致整个计算过程被拖慢。
如上图所示，数据重分发过程中，按照Join Key（即卖家ID）进行shuffle，大部分交易数据记录分发至处理节点1，导致三个并发处理节点中，处理节点1需要处理的数据量远大于其他两个处理节点，从而造成数据处理的不均匀，即数据倾斜。
二、常见的优化方法
2.1 Mapjoin
实现原理是：通过把小表广播到大表所在计算节点上，有效避免了大表的shuffle，因而避免了数据重分布导致的数据倾斜。若大表数据的原始分布本身就有不均匀的情况，此时也可以通过增加随机重分布的临时打散方式，将数据打的散一些，再通过Mapjoin实现数据关联。
SELECT /*+MAPJOIN(dim)*/ *
FROM (SELECT * FROM dwd_tbl) base
LEFT OUTER JOIN (SELECT * FROM dim_tbl) dim
ON base.dim_key = dim.dim_key2.2 特殊值/空值打散
- 特殊值/空值场景也比较普遍，比如主表中有个属性字段在某些场景下为空或者为一些无业务含义的特殊字符串（如DEFAULT），然后此属性字段本身对应了一张数据量较大的维表，需要关联打宽补全。此时做数据关联，由于两张表按照关联key进行shuffle，就会导致主表中该字段为空/相同特殊字符串的数据记录shuffle到同一个节点上，从而导致数据倾斜。
- 此类场景好解决，对特殊值/空值在关联时转为随机值就行。
SELECT *
FROM (SELECT * FROM dwd_tbl) base
LEFT JOIN (SELECT * FROM dim_tbl) dim
ON IF(COALESCE(base.dim_key,'')='',CONCAT('HIVE_',RAND()),base.dim_key) = dim.dim_key2.3 热点值打散，副表呈倍数扩散
- 此类方法使用较少，实现原理是对主表附加一个随机值（例如1~10）字段，记为ext_a字段，然后对应被关联维表数据按照对应倍数进行复制膨胀，并依次赋予1~10的编号，记为ext_b字段，然后在关联两张表时把ext_a、ext_b两个字段也作为关联字段之一。
- 此方法适用于被关联表远远比主表小，但又因数据大小超过内存容量而无法使用Mapjoin，且主表的数据倾斜程度不大的情况下可以使用，但整体上此方案只能对数据热点成倍数的削弱
SELECT *
FROM (
SELECT *,CAST(RAND()*10 AS BIGINT) AS ext_a
FROM dwd_tbl
--主表dwd_tbl
) base
LEFT JOIN (
SELECT *
FROM dim_tbl
--被关联表 dim_tbl
LATERAL VIEW EXPLODE(SPLIT('0;1;2;3;4;5;6;7;8;9',';')) tt AS ext_b
-- 或者Join一个用于倍数膨胀的小表
) dim
ON base.dim_key = dim.dim_key
AND base.ext_a = dim.ext_b2.4 热点数据单独处理/SkewJoin
- 使用此方法通常也意味着被关联的维表数据大小较大，无法使用Mapjoin，只能走普通shuffle模式的join方案。此类场景最典型的案例就是双十一淘系交易大表去关联商家维表，此时的商家维表因记录数和数据大小都较大，所以无法放入到内存，此外部分商家的交易单量远超大盘平均，此时的数据倾斜就需要使用热点数据单独处理的方案。
- 将热点数据提取出来单独处理，可以用Mapjoin的方式完成关联维表热点记录行，非热点则使用普通的shuffle模式的join完成关联。
- 具体操作主要分为三个部分：基于主表统计获得Top热点的属性值：用热点属性值将关联维表拆成热点小表和非热点表，同时也将主表拆成热点主表和非热点主表；热点小表通过Mapjoin与热点主表join，非热点表与非热点主表join，最终两部分再union到一起，完成数据关联。
-- Step01:热点数据记录提取
INSERT OVERWRITE TABLE tmp_hot_list PARTITION (dt = '${bizdate}')
SELECT dim_shop_id AS hot_id
FROM main_table
WHERE dt = '${bizdate}'
GROUP BY dim_shop_id
HAVING COUNT(1) > 10000;
INSERT OVERWRITE TABLE final_result_table PARTITION (dt = '${bizdate}')
-- Step02:热点数据处理,使用MapJoin完成处理
SELECT /*+MAPJOIN(a2,a3)*/
a1.trade_no AS trade_no
,a1.dim_shop_id AS shop_id
,a3.shop_name AS shop_name
,a3.shop_type AS shop_type
FROM (SELECT * FROM main_table WHERE dt = '${bizdate}') a1
-- Step02-1:主表用JOIN关联热点表进行热点记录筛选
JOIN (SELECT * FROM tmp_hot_list WHERE dt = '${bizdate}') a2 -- 热点数据清单
ON a1.dim_shop_id = a2.dim_shop_id
-- Step02-2:热点维度数据处理
LEFT OUTER JOIN (
SELECT /*+MAPJOIN(b2)*/ b1.*
FROM (SELECT * FROM dim_table_info WHERE dt = '${bizdate}') b1
JOIN (SELECT * FROM tmp_hot_list WHERE dt = '${bizdate}') b2 -- 热点数据清单
ON b1.dim_shop_id = b2.dim_shop_id
) a3
ON a1.dim_shop_id = a3.dim_shop_id
UNION ALL
-- Step03:非热点数据处理,使用普通Join完成处理,两张表均需要进行Shuffle
SELECT /*+MAPJOIN(a12)*/
a11.trade_no AS trade_no
,a11.dim_shop_id AS shop_id
,a13.shop_name AS shop_name
,a13.shop_type AS shop_type
FROM (SELECT * FROM main_table WHERE dt = '${bizdate}') a11
-- Step03-1:主表用ANTI JOIN关联热点表进行剔除
LEFT ANTI JOIN (SELECT * FROM tmp_hot_list WHERE dt = '${bizdate}') a12
ON a11.dim_shop_id = a12.dim_shop_id
-- Step03-2:非热点维度数据处理
LEFT OUTER JOIN (
SELECT /*+MAPJOIN(b12)*/ b11.*
FROM (SELECT * FROM dim_table_info WHERE dt = '${bizdate}') b11
LEFT ANTI JOIN (SELECT * FROM tmp_hot_list WHERE dt = '${bizdate}') b12
ON b11.dim_shop_id = b12.dim_shop_id
) a13
ON a11.dim_shop_id = a13.dim_shop_id- 整个步骤稍有些复杂，这里也可以直接用平台的skewjoin参数完成倾斜处理， skew的核心思路就是上面提到的热点数据单独处理，只是做了平台级别的集成，方便用户一键解决数据倾斜问题。
INSERT OVERWRITE TABLE final_result_table PARTITION (dt = '${bizdate}')
SELECT /*+SKEWJOIN(a1)*/
a1.trade_no AS trade_no
,a1.dim_shop_id AS shop_id
,a2.shop_name AS shop_name
,a2.shop_type AS shop_type
FROM (SELECT * FROM main_table WHERE dt = '${bizdate}') a1
LEFT JOIN (SELECT * FROM dim_table_info WHERE dt = '${bizdate}') a2
ON a1.dim_shop_id = a2.dim_shop_id;
2.5 方案总结
总结，上面集中方案核心都是在围绕解决数据重分发（即为shuffle）导致的热点问题，一种是采用Mapjoin的方式避免热点数据重分发，一种是让数据充分发过程中尽可能的均匀。
三、Distmapjoin
3.1 核心思路
数据倾斜的核心在于数据处理不均匀，而数据处理的不均匀往往又来自于数据重分发，也就是shuffle。因此如果能解决好shuffle不均匀问题，或者在不需要对大表进行shuffle的同时就能完成数据关联计算的操作，就能避免数据倾斜的问题。
Mapjoin用于处理热点数据，将维表热点记录广播至大表所在计算节点；Distmapjoin用于处理非热点数据，用于通过构建远程分布式查询节点，实现大表在无需移动的情况下，完成数据关联操作。
https://help.aliyun.com/zh/maxcompute/user-guide/distributed-mapjoin?spm=a2c4g.11186623.0.i1#concept-2197457
https://help.aliyun.com/zh/maxcompute/user-guide/distributed-mapjoin?spm=a2c4g.11186623.0.i1#concept-2197457
3.2 代码实现
WITH
-- STEP01:热点Key采集
tmp_hot_pid AS (
SELECT dim_shop_id,'Y' AS is_hot
FROM main_table_detail
WHERE dt = '${bizdate}'
GROUP BY dim_shop_id
HAVING COUNT(1) > 100000
)
-- STEP02:维表热点数据打标
,tmp_dim_tbl AS (
SELECT /*+MAPJOIN(hot)*/
dim.*
,COALESCE(hot.is_hot,'N') AS is_hot
FROM (
SELECT *
FROM dim_table_info
WHERE dt = '${bizdate}'
) dim
LEFT OUTER JOIN tmp_hot_pid hot
ON dim.dim_shop_id = hot.dim_shop_id
)
-- STEP03:明细热点数据打标
,tmp_dwd_tbl AS (
SELECT /*+MAPJOIN(hot)*/
base.*
,COALESCE(hot.is_hot,'N') AS is_hot
FROM (
SELECT *
FROM main_table_detail
WHERE dt = '${bizdate}'
) base
LEFT OUTER JOIN tmp_hot_pid hot
ON base.dim_shop_id = hot.dim_shop_id
)
-- STEP04:数据合并处理,热点数据用Mapjoin,非热点数据用DISTMAPJOIN
INSERT OVERWRITE TABLE final_result_table PARTITION (dt = '${bizdate}')
SELECT *
FROM (
-- STEP04-1:非热点数据用DISTMAPJOIN
SELECT /*+ DISTMAPJOIN(dim(shard_count=77)) */
dwd_tbl.trade_no AS trade_no
,dwd_tbl.trade_date AS trade_date
,dwd_tbl.shop_id AS shop_id
,dim.shop_name AS shop_name
,dim.shop_type AS shop_type
FROM (SELECT * FROM tmp_dwd_tbl WHERE is_hot = 'N') dwd_tbl
LEFT OUTER JOIN (SELECT * FROM tmp_dim_tbl WHERE is_hot = 'N') dim
ON dwd_tbl.partner_id = dim.partner_id
UNION ALL
-- STEP04-1:热点数据用Mapjoin
SELECT /*+MAPJOIN(dim)*/
dwd_tbl.trade_no AS trade_no
,dwd_tbl.trade_date AS trade_date
,dwd_tbl.shop_id AS shop_id
,dim.shop_name AS shop_name
,dim.shop_type AS shop_type
FROM (SELECT *FROM tmp_dwd_tbl WHERE is_hot = 'Y') dwd_tbl
LEFT OUTER JOIN (SELECT *FROM tmp_dim_tbl WHERE is_hot = 'Y') dim
ON dwd_tbl.partner_id = dim.partner_id
) base ;
3.3 真实效果
当前新方案在支付宝核心支付数据链路上线，给相关可优化节点带来了平均40%的计算耗时缩减和平均30%的计算资源缩减。方案主要应用于支付交易join商家维表、支付交易join合约维表等场景，方案将原本需要手动拆分热点利用“Mapjoin+shuffle进行热点数据处理”的过程，改为利用Distmapjoin或Mapjoin+Distmapjoin的方案，让支付交易大表在计算全过程中均无需移动，在解决数据倾斜问题的同时，也实现了降低计算资源和提升产出时效。
四、方案总结
上面介绍了一种结合Mapjoin和Distmapjoin的数据倾斜处理方案，在有效解决数据倾斜问题的同时还可以避免大表的shuffle，提供了更优的性能表现。实际上如果数据倾斜情况不是特别严重（比如 热点数据行/平均单节点处理数据行 < 100），完全可以直接使用纯Distmapjoin的方案。
综合我们基于Distmapjoin提出的两种方案，我们结合各种方案的优劣势进行方案分级，然后根据具体场景进行更优的方案选择。
参考文章：
https://mp.weixin.qq.com/s/VGgT1faRKGnUuQ-HHQ3Q5g
