MaxCompute复杂数据分布的查询优化实践

摘要： 2017年中国大数据技术大会于12月7-9日在北京新云南皇冠假日酒店隆重举行, 大会就大数据时代社会各行业的智能化进程和行业实践展开深刻讨论。 在12月8日的“大数据分析与生态系统”分论坛上，来自阿里巴巴计算平台事业部的高级技术专家少杰，以“MaxCompute 复杂数据分布的查询优化实践”为题，为现场来宾分享了阿里云MaxCompute最新技术与实践的洞察与经验。

2017年中国大数据技术大会于12月7-9日在北京新云南皇冠假日酒店隆重举行, 大会就大数据时代社会各行业的智能化进程和行业实践展开深刻讨论。

在12月8日的“大数据分析与生态系统”分论坛上，来自阿里巴巴计算平台事业部的高级技术专家少杰，以“MaxCompute 复杂数据分布的查询优化实践”为题，为现场来宾分享了阿里云MaxCompute最新技术与实践的洞察与经验。

概述
数据分布的问题在大数据处理领域由来已久。很不幸，现在流行的大数据处理系统仍然没有很好地解决这个问题。在MaxCompute 2.0全新的优化器中，咱们引入了复杂数据分布，添加了分区剪枝、分布上拉、下推以及分布对齐等优化措施。本文将从数据分布的历史和原理开始，介绍咱们的思路和解决办法。

理解数据分布
提到数据分布，不少人会想到MPP DBMS。的确，咱们一般说只有MPP DBMS才须要考虑数据分布优化。先考虑一个流行的分布式数据库分类学：

Shared Everything: 区别于后两类，这一类基本不是分布式的。
Shared Disk: 数据库服务器能够横向扩展，他们自己没有存储器，经过SAN或NAS技术链接到后端一样能够横向扩展的统一存储。受限于这层网络链接，数据库服务器的扩展能力很是有限。Oracle RAC等商业分布式数据库属于这类。
Shared Nothing: 区别于Shared Disk，这种架构让数据库服务器和存储落在相同的物理节点上（co-located），使得物理节点之间不share任何信息，这大幅减小了网络IO。MPP DBMS和Hadoop属于这类。

显然，只有Shared Nothing的数据库才须要考虑数据分布，你须要预知怎样把数据分布到不一样的物理节点（而不是像Shared Disk那样放在统一存储），会使后续的操做代价更小。例如，在Greenplum中，必须在建表时指定partition key，系统会按照指定的key（哈希）分布数据。若是Join的两张表都按照join key来partition，这个Join就不须要网络IO。若是其中一张表使用了另外一组partition key，那么可能要作一次re-partition。
这就是为何要理解数据分布的缘由：它对应用优化和系统优化都是很是重要的。MPP DBMS在数据分布上都有比较深的积累。可是为何Hadoop这种大数据处理系统没有这类优化？是由于他们须要更强的扩展能力（以及非结构化数据支持，咱们不展开这个话题）。
区别于MPP，Hadoop并非在物理上强制数据和计算在相同节点，若是这么作，系统的横向扩展能力仍然受限。特别是动态扩展能力，考虑正在运行的一个50个节点的Greenplum集群，咱们基本没法作到快速地加入例如2个节点还能高效工做。Hadoop在这方面是很在行的，它的解决办法主要是：
一、存储计算分离
二、去中心化的设计支持高效的peer to peer读写（HDFS）
这就是为何你在Hive中建立一张表时，无须像Greenplum中那样指定partition key，同时Hive在Join的效率低于Greenplum的缘由。

数据分布优化的目的
如上文所述，大数据分布式系统在存储系统上一般倾向随机分布，这提高了扩展性，牺牲了性能。可是从新审视这个权衡，在存储系统上随机分布并不意味着咱们不能利用数据分布优化查询。分布优化的目的是但愿尽量的利用已经存在的分布，并尽量知足将来要求的分布。这种优化包括：

一、分区剪枝：利用数据分布特性，咱们能够作分区剪枝来减小数据读取。例如，哈希分布对于点查询，范围分布对于区间查询能够应用分区剪枝。
二、消除重分布：若是当前的分布知足后续算法的要求，咱们能够消除额外的重分布操做。众所周知，重分布（在Hadoop中叫作shuffle）是分布式算法最主要的消耗。
三、避免数据倾斜：可使用更好的数据分布算法避免数据倾斜。例如，某些单值重复率很高（end-biased）的数据集，使用范围分布而不是哈希分布可能会有效避免数据倾斜带来的性能影响。

定义
数据分布类型
数据分布类型和对应的意义和范例以下所示：

实现
在不破坏Volcano优化器语义的前提下，咱们把分布特性实现为一种physical property，称做distribution。和其余property同样，它有required property和delivered property成对的属性。例如，对于sorted merge join，它对全部输入会施加一个Partial Ordered的required property，同时自身会deliver一个Partial Ordered property，这使得它的后继操做有机会利用这个property，避免一次从新分布。考虑如下查询：

此时Join若是被实现为Sorted Merge Join，它可能会deliver一个Hash[uid]的property，这正好被Aggregate要求，那么这里咱们就能够省去一次没必要要的重分布操做。
要作到相似的优化效果，咱们须要关注下列问题：
一、收集分布特性
二、（局部关系代数编译）选择合适的分布特性
三、（所有代价计算上）规避不合适的分布特性
收集分布特性
产生数据分布有3种途径：
一、用户指定：就像MPP那样，能够在DDL中引入partition key，容许用户指定数据分布。固然区别于MPP，这种分布仅要求在分布式文件系统上的目录结构，并不能关联具体的物理节点。
二、SQL逻辑：SQL逻辑可能产生一次运行时的数据分布。例如distribute by字句声明了一次运行时的数据分布。
三、算法的反作用：每一个分布式算法可能产生一次运行时数据分布。例如，sorted merge join能够保证它的输出数据知足按join key的有序和哈希分布的特征。

有若干算法要求一种特殊的数据分布：
一、Aggregate：Sorted Aggregate要求grouping key的Hash分布。
二、Join：Sorted Merge Join和Hash Join都要求输入按照join key的相同Hash分布。
三、Sort：Order by要求sort key上的Range分布，或Singleton分布。
选择合适的分布特性
即便给定了一系列required和delivered distribution property， 肯定某个操做的分布仍然不是容易的事情。区别于ordering property（仅有排序列和升降序的属性），distribution property的变化不少，这些变化的缘由包括：
一、知足要求的分布有多种选择。例如group by a, b, c这个aggregate，对输入有按a, b, c的Partial Ordered的要求，它对ordering的要求是a, b, c有序，可是知足它的分布能够是Hash(a), Hash(b), Hash(a,b), Hash(a,b,c), RNG(a)等不一样的组合。
二、能利用的实现分布有多种选择。例如join a and b on a.id = b.id这个join，若是a服从Hashid, b服从Hashid，对于Sorted Merge Join，它能够选择要求Hashid，或Hashid，甚至任意Hash(id)。
这些复杂度加大了最优计划的搜索空间。事实上，最优计划是相对于关系代数数量的一个NPC问题。为了缩小搜索空间，咱们引入了启发式的分支选择算法。在编译一个关系代数时，不只须要知足后继操做的要求，还要考虑前序操做提供知足的分布的可能性，后者被实现为称做Pulled Up Property的模块。

Pulled Up Property猜想并筛选可能的前序delivered property，用于在编译时减小搜索宽度。考虑上图的查询，在Join编译时，由于Sink的需求下推，它被要求提供一个Hashc1。Pulled Up Property则从前序操做猜想可能会提供Hashc1和Hashc1，综合考虑，Join可能会直接要求Hashc1，从而减小了Hashc1和Hashc1这两个分支。

规避不合适的分布特性
数据倾斜（Skew）是指在分布中少许节点被分配了大部分数据，致使整个算法退化为单机操做。低并发（Under Partition）是指分布指定了过少的节点，是的分布式资源不能被有效利用。咱们但愿能避免这两种状况。
很显然，更好的统计信息会帮助咱们规避这两种状况。Skew和Under Partition的状况下，须要对代价估计作相应的惩罚，下降他们被选为最优计划的可能性。咱们定义”好”的分布是每一个节点处理的数据量在一个预设的范围，低于或高于这个范围都会被施加惩罚。估计这个数据量的依据包括：
一、输入数据记录数（row count）
二、重复度最高的数据（top values）
三、直方图（histogram）

总结在这篇文章中，咱们介绍了数据分布优化的问题和意义，并解释了MaxCompute在数据分布优化上的实践。这一优化效果已经体如今MaxCompute最新的发布中。从咱们的测试来看，这个优化有至关显著的效果。咱们对TPC-H进行了适当分区后，总体性能提高在20%的量级。即便没有对表数据分区，对用户彻底透明的运行时分区优化也有很好的效果。在咱们线上运行的环境中，14%的查询由于这个优化减小了至少一次数据重分布。