数据分布的问题在大数据处理领域由来已久。很不幸，如今流行的大数据处理系统仍然没有很好地解决这个问题。在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，避免一次重新分布。考虑以下查询：

SELECT uid, count(*) FROM (  SELECT uid FROM user JOIN line ON user.uid = line.uid) GROUP BY uid

此时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服从Hash[id](10)，b服从Hash[id](20)，对于Sorted Merge Join，它可以选择要求Hash[id](10)或Hash[id](20)，甚至任意Hash(id)。

这些复杂度加大了最优计划的搜索空间。事实上，最优计划是相对于关系代数数量的一个NPC问题。为了缩小搜索空间，我们引入了启发式的分支选择算法。在编译一个关系代数时，不仅要满足后继操作的要求，还要考虑前序操作提供满足分布的可能性，后者被称作Pulled Up Property的模块。

Pulled Up Property猜测并筛选可能的前序delivered property，用于在编译时减少搜索宽度。考虑上图的查询，在Join编译时，因为Sink的需求下推，它被要求提供一个Hash[c1](30)。Pulled Up Property则从前序操作猜测可能会提供Hash[c1](10)和Hash[c1](15)，综合考虑，Join可能会直接要求Hash[c1](30)，从而减少了Hash[c1](10)和Hash[c1](15)这两个分支。

规避不合适的分布特性

数据倾斜（Skew）是指在分布中少量节点被分配了大部分数据，导致整个算法退化为单机操作。低并发（Under Partition）是指分布指定了过少的节点，是的分布式资源不能被有效利用。我们希望能避免这两种情况。

很显然，更好的统计信息会帮助我们规避这两种情况。Skew和Under Partition的情况下，需要对代价估计做相应的惩罚，降低他们被选为最优计划的可能性。我们定义”好”的分布是每个节点处理的数据量在一个预设的范围，低于或高于这个范围都会被施加惩罚。估计这个数据量的依据包括：

• 输入数据记录数（row count）；

• 重复度最高的数据（top values）；

• 直方图（histogram）。

总结

在这篇文章中，我们介绍了数据分布优化的问题和意义，并解释了MaxCompute在数据分布优化上的实践。这一优化效果已经体现在MaxCompute最新的发布中。

从我们的测试来看，这个优化有相当显著的效果。我们对TPC-H进行了适当分区后，整体性能提升在20%的量级。即使没有对表数据分区，对用户完全透明的运行时分区优化也有很好的效果。在我们线上运行的环境中，14%的查询因为这个优化减少了至少一次数据重分布。

作者：徐冬，阿里巴巴计算平台事业部高级技术专家，负责实现MaxCompute的基于代价的优化器。拥有多年分布式大数据处理领域经验，参与过多个分布式系统的搭建、开发与维护，是Apache Hive、Apache Calcite的贡献者。

声明：本文为作者投稿，版权归其所有。