一文解析 ODPS SQL 任务优化方法原理

• ODPS接入层的最上层是通过LVS实现负载均衡，把请求发送给HTTP Server，该请求包括用户的AccessID和MD5签名信息，HTTP Server在接收到请求后，会把AccessID和MD5签名发给云账号服务进行用户认证，认证通过后，云账号服务会返回该用户的唯一AccountID，在后续执行逻辑中，发送的请求都是包含该AccountID，而不是AccessID。
• 逻辑层又称作控制层，是MaxCompute的核心部分。实现用户空间和对象的管理、命令的解析与执行逻辑、数据对象的访问控制与授权等功能。在逻辑层有Worker、Scheduler和Executor三个角色：

• Worker处理所有的RESTful请求，它可以本地处理一些作业，如对用户空间、表、资源、作业等的管理；而对于需要执行分布式计算的作业，如SQL、MR等，Worker会进一步把它提交给Scheduler处理；
• Scheduler负责instance的调度，它会维护一个Instance列表，并把Instance分解成各个Task，生成这些Task的工作流——DAG图（Directed Acyclic Graph，有向无环图），把可以运行的Task放到TaskPool中，TaskPool是个优先级队列，后台线程会定时对该优先级队列进行排序；此外，Scheduler还会查询计算集群的资源状况，向计算集群的Fuxi master询问资源占用情况以进行流控（Fuxi slot满的时候，停止响应Executor的task申请）。

• Executor会判断自身资源情况，如CPU、内存、正在运行的Task数（不能超过上限），如果资源满足，则会主动轮询Scheduler的TaskPool请求获取下一个Task，TaskPool会根据Task的优先级和计算集群的资源情况，把相应Task提交给Executor，Executor获取到Task后，会生成计算层的分布式作业描述文件，提交给计算层，监控这些任务的运行状态，并定时把状态汇报给Scheduler。

• 计算层就是飞天内核（Apsara Core),运行在和控制层相互独立的计算集群上。包括Pangu（分布式文件系统）、Fuxi（资源调度系统）、Nuwa/ZK（Naming服务）、Shennong（监控模块）等。MaxCompute中的元数据存储在阿里云计算的另一个开放服务OTS（Open Table Service，开放结构化数据服务）中，元数据内容主要包括用户空间元数据、Table/Partition Schema、ACL、Job元数据、安全体系等。

5.1.1SQL执行顺序

5.1.2SHUFFLE概念

5.1.3ODPS SQL逻辑执行计划算子

• ODPS SQL Task Operator结构(截取自ODPS官方文档)

注：可在ODPS SQL前添加EXPLAIN 执行流程得到，EXPLAIN主要有以下的作用：

1、检查SQL语法；

2、检查读取的表和分区是否符合预期，这样可以排除掉很多分区读错的尴尬；

• 各operator算子含义

5.2.1Multi Distinct优化分析

CASE1：不带Distinct的Count算子使用

EXPLAIN SELECT  app_id        ,count(user_id)FROM    xxx.table_vst_user_testWHERE   dt = '20230816'GROUP BY app_id;

分析：可以看到在Map Task输出阶段，会以app_id字段进行Hash分区传输，输出的临时结果是 app_id和__agg_0_count字段，数据已经预聚合，不存在带有user_id的明细数据shuffle传输，所以任务运行速度较快。

CASE2：带Distinct的Count算子使用

EXPLAIN SELECT  app_id        ,count(DISTINCT user_id)FROM    xxx.table_vst_user_testWHERE   dt = '20230816'GROUP BY app_id;

CASE3：带多Distinct的Count算子使用

EXPLAIN SELECT  app_id        ,count(DISTINCT user_id)        ,count(DISTINCT cy23_source_name_l1)        ,count(DISTINCT cy23_source_name_l2)        ,count(DISTINCT cy23_source_name_l3)        ,count(DISTINCT cy23_source_name_l4)FROM    xxx.table_vst_user_testWHERE   dt = '20230816'GROUP BY app_id;

CASE4：带Distinct的Count算子的优化代码(该CASE是对CASE2的代码优化)

EXPLAIN SELECT  app_id        ,COUNT(user_id)FROM(    SELECT  app_id            ,user_id    FROM    xxx.table_vst_user_test    WHERE   dt = '${bizdate}'    GROUP BY app_id            ,user_id)tGROUP BY app_id

5.2.2系统参数odps.sql.groupby.skewindata=True分析

CASE1：带Distinct的Count算子使用

EXPLAIN SELECT  app_id        ,COUNT(DISTINCT user_id)FROM    xxx.table_vst_user_testWHERE   dt = '${bizdate}'GROUP BY app_id

CASE2：Case1代码前加入系统优化参数

SET odps.sql.groupby.skewindata = true;
EXPLAIN SELECT  app_id        ,COUNT(DISTINCT user_id)FROM    xxx.table_vst_user_testWHERE   dt = '${bizdate}'GROUP BY app_id

5.2.3.Join(Map Join/Inner Join/Left Join)

CASE1：大小表关联(SortMergeJoin)

EXPLAIN SELECT  mini_cat_name_l1        ,COUNT(DISTINCT user_id)FROM    (            --主表            SELECT  app_id                    ,user_id            FROM    xxx.table_vst_user_test            WHERE   dt = '20230816'            GROUP BY app_id                    ,user_id        ) t1LEFT JOIN   (                --维表                SELECT  app_id                        ,mini_cat_name_l1                FROM    xxx.dim_category                WHERE   dt = '20230816'            ) t2ON      t1.app_id = t2.app_idGROUP BY mini_cat_name_l1;

下图来自Logview中的执行计划：

分析：逻辑执行计划中，M1阶段，主要针对右表小程序维表xxx.dim_category进行数据加工提取，由于左右表关联的Key是app_id，所以Hash分区的key也是app_id，输出的中间结果是app_id&mini_cate_name_l1。M2阶段，针对访问事件表xxx.table_vst_user_test进行数据加工，Hash分区的key是app_id&user_id，输出的中间结果是是app_id&user_id，因为Hash分区的key是app_id&user_id，所以在R3_2阶段执行时，不存在热点数据聚集导致的数据倾斜。但参看逻辑执行计划，R3_2的输出会以app_id作为Hash key进行数据传输，数据会在J4_1_3阶段进行整合，并跟M1阶段的小程序维表数据进行MergeJoin，存在数据倾斜的可能。同时在R5_4阶段，Hash分区key是mini_cate_name_l1,不同的行业类目下的用户量差异较大，也会存在可能的数据倾斜。基于Logview的执行计划，可以看到两表关联使用的是MergeJoin的算法(参考上图)。

CASE2：大小表关联使用mapjoin hint(BroadcastHashJoin)

EXPLAIN SELECT  /*+mapjoin(t2)*/mini_cat_name_l1        ,COUNT(DISTINCT user_id)FROM    (            SELECT  app_id                    ,user_id            FROM    xxx.table_vst_user_test            WHERE   dt = '20230816'            GROUP BY app_id                    ,user_id        ) t1LEFT JOIN   (                SELECT  app_id                        ,mini_cat_name_l1                FROM    xxx.dim_category                WHERE   dt = '20230816'            ) t2ON      t1.app_id = t2.app_idGROUP BY mini_cat_name_l1;

Broadcast Hash Join算法：

CASE3：大小表关联使用distributed mapjoin hint

XPLAIN SELECT  /*+ DISTMAPJOIN(t2 (shard_count = 2)) */        mini_cat_name_l1        ,COUNT(DISTINCT user_id)FROM    (            SELECT  app_id                    ,user_id            FROM    xxx.table_vst_user_test            WHERE   dt = '20230816'            GROUP BY app_id                    ,user_id        ) t1LEFT JOIN   (                SELECT  app_id                        ,mini_cat_name_l1                FROM    xxx.dim_category                WHERE   dt = '20230816'            ) t2ON      t1.app_id = t2.app_idGROUP BY mini_cat_name_l1;

• 小表一侧分为M1，R2_1 两个Stage。M1阶段读表并进行Shuffle，Shuffle的过程将数据分片(shard=2)，使得具有相同hash value的数据分发到同一个worker。R2_1（HashTableBuilder1）作为server端，完成HashTable的构建并常驻内存，接受client端（J4_2_3 DistributedMapJoin1）请求完成Lookup查询并返回values。多个shard共同组合成一个分布式的hash table services，shard数量可以手动调整。各shard的service一旦启动，需要等待client端（DistributedMapJoin1）完成所有的request请求后才stop。

• 大表一侧为Stage M3和J4_2_3。J4_2_3（DistributedMapJoin1）作为client端，通过网络传输方式将大表端的join keys，分batches往server端（HashTableBuilder1）发起request请求并获取返回values。由于server端的数据已经按照hash value分shard，client端可以根据数据的特征只请求特定的shard。

参考资料：

• Sort-Merge Join：https://www.sparkcodehub.com/spark-what-is-a-sort-merge-join-in-spark-sql

• Join实现原理：https://www.jianshu.com/p/97e76dddcbfb

• SparkSQL中的三种Join及实现：https://blog.csdn.net/wlk_328909605/article/details/82933552