1. 前言

MaxCompute（原ODPS）是阿里云自主研发的具有业界领先水平的分布式大数据处理平台, 尤其在集团内部得到广泛应用，支撑了多个BU的核心业务。 MaxCompute除了持续优化性能外，也致力于提升SQL语言的用户体验和表达能力，提高广大ODPS开发者的生产力。MaxCompute基于ODPS2.0新一代的SQL引擎，显著提升了SQL语言编译过程的易用性与语言的表达能力。我们在此推出MaxCompute(ODPS2.0)重装上阵系列文章第一弹 - 善用MaxCompute编译器的错误和警告
第二弹 - 新的基本数据类型与内建函数
第三弹 - 复杂类型
第四弹 - CTE，VALUES，SEMIJOIN
第五弹 - SELECT TRANSFORM
第六弹 - User Defined Type
第七弹 - Grouping Set, Cube and Rollup
第八弹 - 动态类型函数
第九弹 - 脚本模式与参数视图

第十弹 - IF ELSE分支语句

第十一弹 - QUALIFY

第十二弹 - PIVOT/UNPIVOT

本文将向您介绍MaxCompute 对Global Zorder的支持。

2. Global Zorder的背景

在大数据分析的典型场景中，查询一般都会带有一个或多个过滤条件，对于这种类型的查询，尤其是在高基数字段上的过滤查询，可以利用文件级别的Min/Max索引在读取数据的时候跳过所有不相关的数据，只读取极少部分需要的数据，这种技术一般称为Data Clustering以及Data Skipping。Data Clustering是指数据按照某种方式聚集，而Data Skipping则指任务在读取数据时跳过不相干数据，Data Clustering的方式以及查询中的过滤条件共同决定了Data Skipping的效果，从而影响查询的执行效率及运行时间。下图展现了一个普通表在不同Data Clustering下的Data Skipping效果，从右图可以看到在有合适的Data Clustering的情况下，任务可以过滤掉大部分文件：

目前在MaxCompute上不仅有传统的ETL作业，也有大量的分析型作业，在实际业务中经常面对的问题是：如何在查询时尽可能达到最有效的Data Skipping，而不是做全表的扫描，从而大大提高查询的执行效率？这里面的关键点就是Data Clustering。这种情况下，我们可以很轻松的想到可以针对过滤字段进行全局或者本地排序，这种做法对于1维的分析查询有效，但是对于多维的分析查询，这种做法效果并不好。比如说当使用a，b，c三个字段排序时，对a的数据聚集性是最好的，越往后聚集性越差。如果字段a的基数比较高，那么对于后面的字段可能完全没有聚集性，达不到过滤的效果。

针对这种多维分析查询，MaxCompute提供了聚簇索引（Hash Clustering和Range Clustering）来进行优化，但是这2种方式有各自的局限性：

• Hash Clustering：只有在过滤条件包含了所有的Cluster Key之后才能进行有效的数据过滤。
• Range Clustering：满足最左匹配原则，即只有出现了Cluster Key的前缀过滤条件才能有比较好的过滤效果，如果不包含前缀过滤条件则效果不佳。

针对上述情况，MaxCompute支持了Local Zorder，使得在多维分析场景下查询能跳过不相干数据，但是这种方式只是单个文件内部按照Zorder排序，并不是对全局数据做一个重分布，所以如果数据分散在各个文件，那么数据的聚集程度可能也不高，无法做到最有效的Data Skipping。鉴于这个问题，MaxCompute支持了Global Zorder，使得整个表或者分区的数据在全局上能按照指定字段进行ZORDER排序，以便数据能有更好的聚集性。

3. MaxCompute Global Zorder的使用

3.1 语法格式

set odps.sql.default.zorder.type=global;
-- 必选，不加这个flag则默认是local zorder
set odps.sql.zorder.add.noise=true; 
-- 可选，对于有倾斜的数据可以加上这个参数以便打散数据；打散之后可能影响数据聚集性的效果
INSERT OVERWRITE|INTOTABLE table_name [PARTITION (partcol1=val1, partcol2=val2 ...)] [(col1,col2 ...)]
select_statement
FROM from_statement
[ZORDER BY zcol1 [, zcol2 ...]]
[SORT BY zol1 [, zcol2 ...]]
;

3.1.1 语法上global zorder和local zorder没有区别，是需要通过flag来区分，默认是local zorder，所以要使用global zorder，需在脚本前加flag：set odps.sql.default.zorder.type=global。

3.1.2 对于分区表，一次只允许对1个分区进行zorder by排序。

3.1.3 zorder by字段数目只能在2-4之间，超过4个字段会报错。

3.1.4 sort by语句用于指定单个文件内部排序的方式，如果没有指定，则单个文件内部会按照指定的字段进行local zorder排序。

3.1.5 目标表为聚簇表时，不支持zorder by子句。

3.2 对新创建表进行zorder by排序示例

--创建目标表mf_src。
create table mf_src (key string, value string);
insert overwrite table mf_src
select a, b from values ('1', '1'),('3', '3'),('2', '2')
as t(a, b);
select * from mf_src;
--返回结果
+-----+-------+
| key | value |
+-----+-------+
| 1   | 1     |
| 3   | 3     |
| 2   | 2     |
+-----+-------+
--创建目标表mf_zorder_src，与mf_src有相同的结构。
create table mf_zorder_src like mf_src;
--使用global zorder模式排序。
set odps.sql.default.zorder.type=global;
insert overwrite table mf_zorder_src
select key, value from mf_src 
zorder by key, value;
select * from mf_zorder_src;
--返回结果
+-----+-------+
| key | value |
+-----+-------+
| 1   | 1     |
| 2   | 2     |
| 3   | 3     |
+-----+-------+

3.3 对存量表进行zorder by排序示例

-- target表是存量表
set odps.sql.default.zorder.type=global;
insert overwrite table target
select key, value from target 
zorder by key, value;

4. 技术解析

本节重点阐述业界和MaxCompute的global zorder实现。

4.1 实现调研

多维数据没有天然的有序性，需要将多维数据映射成一维数据进行比较，而Zorder是把天然没有有序性的多维数据以某种方式映射成一维数据进行比较。映射后的一维数据，能够保证各个原始维度按照同种程度去保证其聚集性。实现全局的zorder一般需要两个步骤：

4.1.1 step1. 计算zvalue，将多列的值拍平到一维空间。

4.1.2 step2. 按照zvalue进行全局排序。

大数据系统中一般使用range partition来进行全局排序，而计算zvalue一般有以下有2种方式：

方法1. 基于映射策略生成zvalue

1）直接将类型转成bits位表示，缺点是开销会非常大，而且可能由于数据分布不好，导致效果很差。比如x 字段取值(0, 1, 2)， y字段取值(100, 200, 300)， 用x, y生成的z-value只是完整z曲线的一部分，对其做z值排序的效果和直接用x排序的效果是一样的；再比如x的基数值远远低于y的基数值时采用上述策略排序效果基本和按y值排序是一样的，真实效果还不如先按x排序再按y排序。

2）使用全局字典的方式。这种方式的前提是需要取字段的Distinct值并进行全局排序，排序的序号值自然就是从0开始的连续正整数，然后使用排序序号来计算zvalue，这种方式因为涉及到取distinct值以及全局排序，开销非常大，实际不大可行。

方法2. 基于boundary index生成zvalue - 典型代表是Delta、Hudi

对每个参与zorder的字段采样若干个数值并对进行排序，从而计算出每个字段的boundary；实际映射过程中每个字段映射为该数据所在boundary的下标，然后使用下标进行z-value的计算。

4.2 MaxCompute Global Zorder的实现

在MaxCompute中采取的实现类似方法2，即将zorder字段映射为bounday的下标用于zvalue的计算，但是不像Spark能利用shuffle对字段进行采样并计算分区边界，MaxCompute根据自身引擎特点采取的是内建函数的形式来完成计算的，如下所示：

其中涉及的三个函数如下：

• boundary

函数声明 Array boundary(T ref, bigint bucketNum);

返回指定列的若干个boundary，在当前实现中bucketNum是1000。

• array_lower_bound_index

函数声明 Long ARRAY_LOWER_BOUND_INDEX(Array array, T value) ；

返回array中不小于value的第一个元素的index。

• zvalue

函数声明 Long ZVALUE(Integer a, Integer b.. Integer d) ；

返回给定列的zvalue值。

5. 经验总结

Zorder功能不是一个silver bullet，也没有统一的规则来指导是否该用和如何使用。很多时候都需要case by case去尝试改造，综合评估改造ZORDER后生成数据带来的额外计算成本，相对于存储成本的节省和下游消费计算成本的节省，是否有收益。以下是一些经过实践的经验上总结的建议供参考：

5.1 优先考虑Clustered Index而不是Zorder的场景

• 如果过滤条件基本都是某个前缀的组合，比如 a、a和b、a和b和c，那么使用Clustered Index（即ORDER BY a, b, c）更有效，此时不要使用ZORDER BY。因为ORDER BY对第一个字段有非常好的排序效果，而对后面字段影响较少；而ZORDER BY对每个字段给予了相同的权重，仅看某一列的排序是不如ORDER BY的第一个字段的。
• 如果某些字段经常出现在JOIN KEY上，这些字段使用Hash或Range Clustering更合适。因为MaxCompute Zorder的实现仅仅在文件内进行了排序，而SQL引擎对ZOrder的数据分布没有感知；但是SQL引擎是能够感知Clustered Index的，因此在做查询计划阶段能够更好地优化JOIN的性能。
• 如果某些字段经常需要进行GROUP BY和ORDER BY，那使用Clustered Index可以获得更好的性能。

5.2 Zorder使用建议

• 选取经常出现在过滤条件中的字段作为ZORDER字段，尤其是经常联合在一起过滤的。
• Zorder By的字段数越多，每个字段的排序性能会越差，因此不宜超过4个。如果只有一个字段，那就应该使用Clustered Index而不是Zorder。
• 表的数据量也不宜太小和太大。数据量太小的话，ZOrder基本看不到什么效果。而数据量太大的话，按ZOrder方式产出数据的代价比较高，尤其是基线任务会明显影响产出的时间。
• Global Zorder相对Local Zorder有更好的聚集性，但是代价也更大，因而需要根据任务特点选择适合的方式