奇思妙想的SQL|兼顾性能的数据倾斜处理新姿势

简介: 文章分享了在支付宝支付数据链路改造升级过程中,针对数据倾斜的优化实践新方法,在解决数据倾斜问题的同时,还能兼顾更优的计算性能!

本篇为系列第2篇,分享在支付宝支付数据链路改造升级过程中,针对数据倾斜的优化实践新方法,在解决数据倾斜问题的同时,还能兼顾更优的计算性能!


文章导读


SQL作为目前最通用的数据库查询语言,其功能和特性复杂度远不止大家常用的“SELECT * FROM tbl”这样简单,一段好的SQL和差的SQL,其性能可能有几十乃至上千倍的差距。而写出一个好的能兼顾性能和易用性的SQL,考验的不仅仅是了解到多少新特性新写法,而是要深入理解数据的处理过程,然后设计好数据的处理过程。

因此想推出本系列文章,并取名为《奇思妙想的SQL》,希望能以实际案例出发,和大家分享一些SQL处理数据的新方案新思路,并在过程中融入对问题本质的理解,希望大家能喜欢~

本篇为系列第2篇,分享下在支付宝支付数据链路改造升级过程中,针对数据倾斜的优化实践新方法,在解决数据倾斜问题的同时,还能兼顾更优的计算性能!


一、场景描述

数据倾斜的处理,作为校招/社招最经典的一道面试题,相信作为一名数据研发同学,多少都有些了解。数据倾斜可能发生在join、group by、Count Distinct等环节,但本质上其实都类似,即因为数据重分发或重分布等原因,导致大部分数据分发至少数几个计算节点上,闲着大伙儿,累死少数几个兄弟。问题表象也好识别,以ODPS场景为例,少数几个Fuxi Instance处理的数据量,远大于同一环节的其他Instance处理的数据量,并伴有明显的长尾现象。

典型的案例就是淘宝双十一场景中,交易订单明细大表需要关联商家信息维表以补全商家信息,在数据关联处理中,同一个商家对应的交易订单和维表对应商家信息,将根据卖家ID shuffle至同一个数据处理节点上。由于TOP商家在大促中产生的交易单量远大于普通商家,从而导致大量的数据集中到一台或者几台机器上计算,这些数据的计算速度将远远低于平均计算速度,导致整个计算过程被拖慢。

image.png


如上图所示,数据重分发过程中,按照Join Key(即卖家ID)进行Shuffle,大部分交易数据记录分发至处理节点1,导致三个并发处理节点中,处理节点1需要处理的数据量远大于其他两个处理节点,从而造成数据处理的不均匀,即数据倾斜。

二、常见的优化方法

2.1.Mapjoin

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_key

2.2.特殊值/空值打散

  • 特殊值/空值场景也比较普遍,比如主表中有个属性字段在某些场景下为空或为一些无业务含义的特殊字符串(如DEFAULT),然后此属性字段本身对应了一张数据量较大的维表,需要关联打宽补全。此时做数据关联,由于两张表需要按照关联key进行shuffle,就会导致主表中该字段为空/相同特殊字符串的数据记录shuffle到同一节点上,从而导致数据倾斜。
  • 此类场景也好解决,对特殊值/空值在关联时转为随机值就行。
SELECT * 
FROM (SELECT * FROM dwd_tbl) base 
LEFT OUTER JOIN (SELECT * FROM dim_tbl) dim
ON IF(COALESCE(base.dim_key,'')='',CONCAT('HIVE_',RAND()),base.dim_key) = dim.dim_key

2.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
) base 
LEFT OUTER JOIN (
    SELECT *
    FROM 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_b


2.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的核心思路就是上面提到的热点数据单独处理,只是做了平台级别的集成,方便用户一键解决数据倾斜问题。详细用法和详细原理可参考《阿里云-SKEWJOIN HINT》[1]。
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
;

image.png

2.5.方案总结

  • 不难发现,上面几种方案核心都是在围绕解决数据重分发(即shuffle)导致的热点问题,一种是想方设法采用Mapjoin的方式避免热点数据重分发,一种是让数据重分发过程中尽可能得均匀。
  • 不管是哪种思路,问题核心都还是在解决shuffle导致的数据分布不均匀的问题。所以,一切的“罪魁祸首”就是shuffle、shuffle、shuffle~

三、一种新的思路 WithDistmapjoin~

3.1.核心思路

数据倾斜的核心在于数据处理不均匀,而数据处理的不均匀往往又来自数据重分发,也就是shuffle。因此如果能解决好shuffle不均匀问题,或者在不需要对大表进行shuffle的同时就能完成数据关联计算的操作,就能避免数据倾斜问题。

在此我们联想到了Distmapjoin的能力,通过对中小规模的表(为便于理解,后文用维表进行替代)构建远程分布式查询节点,大表再通过网络远程查询相关维表数据,从而实现了类似于Mapjoin的方式,大表无须shuffle即能完成Join操作。在此,预估Distmapjoin可以非常好的解决大表shuffle导致的数据倾斜问题。但我们忽略了一个问题,热点问题其实还没消失,只是转移成了远程网络查询的IO热点问题。当然在技术实现细节上可以通过同一key的多次查询合并为一次等方案进一步削弱热点问题,但热点问题并没有完全消除。在此,我们可以返回去参考skewjoin的方案,将维表的热点记录和非热点记录分而治之,只不过此时我们使用的不是“热点Mapjoin+非热点shuffle”的方案,而是采用“热点Mapjoin+非热点Distmapjoin”的方案。Distmapjoin的方案及原理详见《阿里云-DISTRIBUTED MAPJOIN》[2]

Mapjoin用于处理热点数据,将维表热点记录广播至大表所在计算节点;Distmapjoin用于处理非热点数据,用于通过构建远程分布式查询节点,实现大表在无需移动的情况下完成数据关联操作。当前方案还额外实现了提效的收益,大表在全流程中均无需shuffle,躺着不动就能实现join操作~

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

image.png


3.3.真实效果

当前新方案在支付宝核心支付数据链路上线,给相关可优化节点带来了平均40%的计算耗时缩减和平均30%的计算资源缩减。方案主要应用于支付交易join商家维表、支付交易join合约维表等场景,方案将原本需要手动拆分热点利用“Mapjoin+shuffle进行热点数据处理”的过程,改为利用Distmapjoin或Mapjoin+Distmapjoin的方案,让支付交易大表在计算全过程中均无需移动,在解决数据倾斜问题的同时,也实现了降低计算资源和提升产出时效。

另外值得说的一点,我们对域内的支付交易数据链路进行了全链路HashCluster的处理,结合Distmapjoin的倾斜处理方案,可有效避免已经排好序的HC表再二次重分桶,全链路加工过程中都可以保持其原本已经设定好的HashCluster分桶策略~

四、方案总结

上面介绍了一种结合Mapjoin和Distmapjoin的数据倾斜处理方案,在有效解决数据倾斜问题的同时还可以避免大表的shuffle,提供了更优的性能表现。实际上如果数据倾斜情况不是特别严重(比如 热点数据行/平均单节点处理数据行 < 100),完全可以直接使用纯Distmapjoin的方案。综合我们基于Distmapjoin提出的两种方案,我们结合各种方案的优劣势进行方案分级,然后根据具体场景进行更优的方案选择。

image.png

参考链接:

  • [1]https://help.aliyun.com/zh/maxcompute/user-guide/skewjoin-hint-1?spm=5176.28426678.J_HeJR_wZokYt378dwP-lLl.1.3a415181D3BDr0&scm=20140722.S_help@@文档@@455433.S_BB2@bl+RQW@ag0+BB1@ag0+os0.ID_455433-RL_ODPS%20SKEWJOIN-LOC_search~UND~helpdoc~UND~item-OR_ser-V_3-P0_0
  • [2]https://help.aliyun.com/zh/maxcompute/user-guide/distributed-mapjoin?spm=a2c4g.11186623.0.i1#concept-2197457
相关实践学习
基于MaxCompute的热门话题分析
本实验围绕社交用户发布的文章做了详尽的分析,通过分析能得到用户群体年龄分布,性别分布,地理位置分布,以及热门话题的热度。
SaaS 模式云数据仓库必修课
本课程由阿里云开发者社区和阿里云大数据团队共同出品,是SaaS模式云原生数据仓库领导者MaxCompute核心课程。本课程由阿里云资深产品和技术专家们从概念到方法,从场景到实践,体系化的将阿里巴巴飞天大数据平台10多年的经过验证的方法与实践深入浅出的讲给开发者们。帮助大数据开发者快速了解并掌握SaaS模式的云原生的数据仓库,助力开发者学习了解先进的技术栈,并能在实际业务中敏捷的进行大数据分析,赋能企业业务。 通过本课程可以了解SaaS模式云原生数据仓库领导者MaxCompute核心功能及典型适用场景,可应用MaxCompute实现数仓搭建,快速进行大数据分析。适合大数据工程师、大数据分析师 大量数据需要处理、存储和管理,需要搭建数据仓库?学它! 没有足够人员和经验来运维大数据平台,不想自建IDC买机器,需要免运维的大数据平台?会SQL就等于会大数据?学它! 想知道大数据用得对不对,想用更少的钱得到持续演进的数仓能力?获得极致弹性的计算资源和更好的性能,以及持续保护数据安全的生产环境?学它! 想要获得灵活的分析能力,快速洞察数据规律特征?想要兼得数据湖的灵活性与数据仓库的成长性?学它! 出品人:阿里云大数据产品及研发团队专家 产品 MaxCompute 官网 https://www.aliyun.com/product/odps&nbsp;
相关文章
|
1月前
|
存储 SQL 关系型数据库
PolarDB这个sql行存和列存性能差别好大 ,为什么?
PolarDB这个sql行存和列存性能差别好大 ,为什么?
56 0
|
18天前
|
SQL 存储 关系型数据库
如何巧用索引优化SQL语句性能?
在 MySQL 中,添加合适的索引可以显著提升慢查询的速度,因为索引加快了数据检索。要优化 SQL 性能,首先需定位慢查询,可通过查看执行时间和执行计划。`EXPLAIN` 命令用于查看执行计划,分析如`type`(全表扫描最慢,索引扫描较快)、`key`(未使用索引为NULL)等字段。例如,全表扫描的查询可考虑为慢查询,并创建相应索引进行优化。此外,注意聚簇索引、索引覆盖和最左前缀原则等索引使用技巧,以提高查询效率。启用慢查询日志并设置阈值,有助于识别已运行的慢查询。
|
1月前
|
SQL Windows
安装SQL Server 2005时出现对性能监视器计数器注册表值执行系统配置检查失败的解决办法...
安装SQL Server 2005时出现对性能监视器计数器注册表值执行系统配置检查失败的解决办法...
20 4
|
1月前
|
SQL 分布式计算 数据处理
奇思妙想的SQL|兼顾性能的数据倾斜处理新姿势
本篇为系列第2篇,分享在支付宝支付数据链路改造升级过程中,针对数据倾斜的优化实践新方法,在解决数据倾斜问题的同时,还能兼顾更优的计算性能!
|
1月前
|
SQL 存储 缓存
如何通过优化SQL查询提升数据库性能
SQL查询是数据库的核心功能之一,对于大型数据量的应用程序来说,优化SQL查询可以显著提升数据库的性能。本文将介绍如何通过优化SQL查询语句来提升数据库的性能,包括索引优化、查询语句优化以及其他一些技巧。
|
1月前
|
SQL 监控 关系型数据库
解密SQL性能异常事件及阿里云数据库的性能调优实践
作为开发者想必都知道数据库是现代应用的核心组件之一,而且在当今互联网时代之下,SQL查询的性能直接影响系统的整体性能,它的性能对于系统的稳定性和响应速度至关重要。那么本文就来讨论一下SQL性能异常的排查和优化方法,包括我个人印象深刻的SQL性能异常事件,以及分享一下使用阿里云数据库产品/工具进行SQL性能调优的经验和心得体会。
100 1
解密SQL性能异常事件及阿里云数据库的性能调优实践
|
1月前
|
SQL 监控 关系型数据库
常见的SQL优化和排查性能异常秘籍
常见的SQL优化和排查性能异常秘籍
48 1
|
SQL Oracle 关系型数据库
按图索骥:SQL中数据倾斜问题的处理思路与方法
 本文通过示例分享部分场景的处理方法  未使用绑定变量 使用绑定变量 几种特殊场景 1 测试环境说明 数据库版本:ORACLE 11.2.0.4 新建测试表tb_test: create tablescott.
2221 0
|
7天前
|
SQL DataWorks NoSQL
DataWorks产品使用合集之如何将SQL Server中的数据转存到MongoDB
DataWorks作为一站式的数据开发与治理平台,提供了从数据采集、清洗、开发、调度、服务化、质量监控到安全管理的全套解决方案,帮助企业构建高效、规范、安全的大数据处理体系。以下是对DataWorks产品使用合集的概述,涵盖数据处理的各个环节。
|
1月前
|
SQL API 流计算
实时计算 Flink版产品使用合集之在Mac M1下的Docker环境中开启SQL Server代理的操作步骤是什么
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
156 1

热门文章

最新文章