maxcompute MaxCompute 2.0 性能优化揭秘

回顾大数据技术领域的大事件，可以追溯到2006年Hadoop的正式推出。环顾四周，围绕数据库和数据处理引擎，业界充斥着各种大数据技术。这是技术专家的好时机。数据库领域只有300多个数据库，围绕Hadoop生态系统的大数据处理技术正在蓬勃发展。在2017年云起社区在线技术峰会的大数据技术峰会上，阿里巴巴云大数据计算平台架构师林伟做了题为“MaxCompute Brain:Cost-Based Optimizer”的分享，分享了阿里巴巴大数据计算服务的大脑——基于成本的优化器的设计和架构。

MaxCompute简介

MaxCompute是一个快速且全面管理的PB/EB数据仓库解决方案。MaxCompute是阿里巴巴内部的核心大数据平台，具有扩展1万台服务器和跨区域容灾的能力。它承担了集团内绝大部分的计算任务，支持数百万的日常运营规模。MaxCompute为用户提供了完整的数据导入方案和多种经典的分布式计算模型，可以更快地为用户解决海量数据计算问题，有效降低企业成本，保证数据安全。

MaxCompute架构

MaxCompute的基本架构如上图所示。底层是建立在物理机上提供统一存储的盘古分布式文件存储系统；盘古之上的一层是伏羲分布式调度系统，管理包括CPU、内存、网络、磁盘在内的所有计算资源；下一级是统一执行引擎，即MaxCompute执行引擎；在执行引擎之上，会创建各种计算模式，比如流计算、图计算、离线处理、内存计算、机器学习；在此之上，还会有一层相关的编程语言，就是MaxCompute语言；在语言方面，我们希望为各个应用端提供一个良好的平台，让数据工程师可以通过平台开发相关的应用，使应用能够在分布式场景下快速部署和运行。

MaxCompute的研发思路

MaxCompute的研发思路主要分为以下四个方面:

高性能、低成本和大规模。希望打造的MaxCompute平台能够提运算的高性能，尽可能降低用户的使用成本，并且在规模上面能够达到万台机器以及多集群的规模。稳定性，服务化。希望MaxCompute平台能够提供稳定性和服务化的方式，使得用户不用过多地考虑分布式应用的难度，而只需要注重于用户需要进行什么样的计算，让系统本身服务于用户，并能够提供稳定性，服务化的接口。易用性，服务于数据开发者。希望MaxCompute平台是易用的，并且能够很方便地服务于数据开发工程师，不需要数据工程师对于分布式的场景进行很深的理解，而只要关注于需要用这些数据进行什么样的运算就可以，接下来就是由MaxCompute平台帮助数据开发工程师高效并且低成本地执行自己的想法。多功能。希望MaxCompute能够具有更多的功能，不仅仅是支持流计算、图计算、批处理和机器学习等，而希望更多种类的计算能够在MaxCompute平台上得到更好的支持。

最大计算的大脑——优化器

基于以上研发思路，MaxCompute平台需要有更强大的大脑，需要更好的理解用户的数据、计算和用户本身。MaxCompute的大脑需要能够帮助用户更高效地优化操作，通过系统层面了解用户需要执行什么样的操作，从而达到上述各种目的。它使用户能够脱离分布式场景，而不用考虑如何高效地执行操作。相反，它将这项工作交给MaxCompute的大脑，为用户提供一个更智能的平台，这就是MaxCompute可以为用户带来的价值。

那么MaxCompute到底是什么大脑呢？实际上是一个优化器。优化器可以将所有信息串联在一起，通过了解系统中数据的相关性和用户的意图，可以充分分析各种环境，从而实现分布式场景下用户操作的最高效执行。在这次分享中，以离线计算为主要例子，介绍了MaxCompute-optimizer的大脑。

首先，简要介绍了离线计算的概念。MaxCompute离线计算的架构设计如上图所示。在计算层面，往往有类似于高级语言的脚本语言。MaxCompute提供了一种类似SQL的脚本语言，由FrontEnd提交，然后处理成逻辑执行计划。在优化器的指导下，逻辑执行计划被转化为更有效的物理执行计划。与运行时连接后，福喜分布式调度系统将物理执行计划分解为操作节点进行操作。

以上过程的核心在于如何充分了解用户的核心计划，通过优化得到高效的物理执行计划。这个过程称为优化器。目前开源社区的Hive和Spark中的一些优化器基本上都是基于规则的优化器。事实上，对于优化器来说，在单机系统上有这样一种分类，分为基于规则的优化器和基于成本的优化器。

在单机场景中，基于规则的优化器在Oracle 6-9i中使用，基于成本的优化器在Oracle 8中启动，而Oracle 10g完全取代了以前的基于规则的优化器。在大数据场景中，像Hive，一开始只有基于规则的优化器，新版的Hive也引入了基于成本的优化器，但Hive并不是真正的成本优化器。MaxCompute使用完全基于成本的优化器。那么这两个优化器有什么区别呢？事实上，基于规则的优化器理论上是根据对逻辑模式的识别来转换规则的，即识别一个模式可能会触发一个规则，将执行计划从A更改为B，但这种方法对数据不敏感，优化是局部贪婪的，就像一个登山者只看着自己前方10米范围内的地方在向上，而没有考虑到自己应该往下走才能到达更高的山顶一样。因此，基于规则的优化器很容易陷入局部优秀但全局较差的场景，并且很容易通过应用规则的顺序来产生。基于成本的优化器通过火山模型尝试各种可能的等价执行计划，然后根据数据的统计信息计算这些等价执行计划的“成本”，最后选择成本最低的执行计划，以达到全局最优。

这里我们分享一个具体的例子来帮助您理解为什么基于规则的优化器不能实现全局优化。上图中的脚本是指先完成A、B、C上的联接，联接的结果在一列上分组，计算平均值。上面的查询过程可以绘制成树状的逻辑执行计划，在数据库字段中往往是自下而上的，也就是说对于逻辑计划树来说，叶节点是输入，最终的目标输出是根节点，所以最终的数据流是自下而上的。可以看出，在这个逻辑计划中，假设大小为(B)，对三个表A、B和C执行连接

基于成本的优化器采用不同的方案。首先，通过火山模型将查询扩展成几个等价的可执行计划。例如，您可以在加入c之前加入A和b，或者在加入A之前加入b和c。在这两个计划中，因为下面的计划中有一个交换，所以在基于成本的优化器的后面会有一个成本成本模型。通过计算，发现第一个方案在成本上更优越，因此将选择最佳方案来执行。在基于成本的优化器中，在分布式场景下建立了许多独特的成本模型，并考虑了非SQL。由于很多场景都是与互联网相关的应用，用户需要大量来自非SQL的支持，所以用户自定义函数可以帮助用户实现一些非SQL与关系数据相结合的查询优化。最后，还有一些针对各种分布式场景的优化，这也是基于成本的优化器所做的一些工作，它不同于独立的优化器。

接下来我们来分享一下火山火山模型的相关性。其实火山模型是成本模型的一个引擎，这个引擎是在单机系统上提出来的。火山模型中也有一些规则，但是与基于规则的优化器中的规则不同，它中的规则更像是一些转换函数。火山模型将首先对逻辑执行计划进行分组，然后为了在组上完成一个任务，它将首先探索组中的局部表达式，然后根据一些规则应用一些转换。这些变换在原理上是代数等价的，实际上并不是每次等价关系改变时都替换原来的逻辑执行计划树，而是在原来的基础上拆分一个新的树。所以最后会有很多等价的执行计划树，可以通过基于成本的优化器来选择最佳的执行计划。火山模型的原理是首先希望每个规则都是局部的，即局部和正交的规则越好，空的探索越全面。例如，如果在平面上定义了四个方向，那么可以通过这四个方向搜索整个2D平面的任意点。同样的优化问题是在空之间选择最优方案，所以我们希望探索规则每次变化都能正交，这样就可以用更少的规则探索整个空，那么如何探索/[/。

前面的分享比较抽象，这里给出进一步的例子，希望加深大家对优化过程的理解。假设有一个非常复杂的逻辑执行计划树，这是用户的真实任务。现在，提取了一小部分。在优化方案时，我们会先分析现有的规则是否能匹配模式。假设图中的两个节点可以匹配模式，一个是过滤器，另一个是项目。理论上，filter想推送到叶子节点，也就是说越早进行过滤越好。现在有一种模式:如果项目上出现过滤，就是需要在项目之前做过滤，这样就可以转换成另一个计划，这两个节点可以变成新的节点，也就是可以改变过滤和项目的顺序，这就是套用规则的过程。同样还有另一个节点，比如聚合操作可以和其他模式匹配，然后找到对应的规则转化为等价的节点操作，这样就可以通过重用一个树节点的模式来维护多个树。在这个例子中，我们可以看到使用了两个规则，看起来节点只是一个存储，但实际上描述了四个等价的树。之后会计算这四个等价树的成本，最后选择成本最低的树作为执行计划。这是整个基于成本的优化过程，但是可以看出，在逻辑计划树很大，规则变化很多的情况下，整个探索空还是很大的，所以需要在很多因素上考虑优化过程。

接下来，我们将介绍优化引擎的一般算法。下图是优化引擎的简化算法，但实际实现中还有很多因素需要考虑，下图没有显示。

首先注册一个逻辑执行计划中的所有逻辑节点，同时将规则与已有的逻辑模式进行匹配，然后将匹配成功的规则推入规则队列，然后循环弹出规则队列中的规则，真正应用规则。当然，应用规则有两个条件。一种是应用后可以生成等价树，即可以在树的局部拆分另一个树状态，拆分后的树可以匹配其他模式。如果本地所有的规则都匹配了，就可以计算成本了。计算成本得到最佳方案后，可以放弃这部分的继续优化。如果你认为目前的方案还是不最优，可以记录成本，继续优化树的其他部分，直到最终找到最优方案。

分布式查询中优化问题的一个例子

在这里，我们将给出一些分布式查询中的优化问题的例子，这些问题不同于独立系统中的优化问题。

例1其实很简单，就是连接两个表，T1已经按照A和B分区了；在一种方法中，因为T1是根据A和B划分的，并且连接条件在A之上，所以T1需要在与T2连接之前根据A进行划分。但是，如果T1表非常大，远远大于T2表，那么我们不想根据重新分区T1。相反，我们可以采用另一种方案，即把T2作为一个整体，把T2的所有数据广播到T1的每个数据上。因为联接条件是对A做内部联接，所以我们可以做这样的选择，这样可以避免刷新大数据。在这种场景下，如何感知加入条件是关键。上例中两个方案没有绝对最优，但需要根据数据大小、T2数据量和T1数据片段的分布来决定哪个方案是最优解。SIFMOD12上有很多论文讨论这个问题，这里就不详细描述了。

再来分享一个分布式优化问题的例子。如图所示，t1和T2仍然在a上连接，并且会有条件限制T1。a >: 20。完成后，项目将被执行，完成的结果将被视为一个新的b列，最终所有结果将由b排序..T1和T2都是范围分区，但这不是散列分区，并且因为已经进行了全局排序，所以在这里连接时可以使用两个表之间的范围分区边界，而无需再次刷新数据。比如目前已知的join会出现哪些大于20的分区，可以根据选择的边界读取相应的数据后进行，可以尽量避免数据混排。加入后会有一个用户自定义的方法，这个方法得到的结果由b按照顺序规则排序，假设这个foo()方法是一个单调递增的函数。这样就可以利用上面的条件，就是我们已经按照范围进行了划分，在join和foo()之后仍然可以保存B的顺序，这样就可以不用引入交换，直接通过B来进行排序。这是一个分布中的查询优化，也就是说，如果你能了解数据中的碎片，了解数据的分布，了解用户定义的函数方法，以及这些方法是如何与优化器交互的，你就可以优化分布式查询。实际上，通过用户的Annotation，我们可以知道用户的方法具有哪些特性，可以维护哪些数据属性。

用户定义函数UDF

在分布式系统中，尤其是非SQL，需要扩展大量的用户自定义函数，因为很多查询过程并不像join和aggregate那么简单，需要对很多唯一的函数进行建模，所以需要实现用户自定义函数。一旦有了用户定义的函数，优化器通常很难理解UDF，因此优化的范围将受到极大的限制。比如上图中间的黄色节点包含了用户自定义函数，但是系统可能不知道这个函数是做什么的，所以在优化的时候可能会分成三个更小的可优化段，进一步的优化会分三个小段进行。如果优化器能理解用户定义的函数在做什么，它就能穿透UDF实现更广泛的优化。那么，UDF的哪些特性可以帮助优化器穿透它呢？实际上，我们可以分析UDF是否是逐行的，考虑它是否是逐行处理的，是否没有跨行，考虑UDF是否是一个单调的函数，在处理过程中某些列是否保持不变，即它们是否可以被穿透，它是否可以保持数据碎片化或有序化，以及关于成本的一些信息，它的选择性是高还是低。以及输出的数据分布是多还是少等等。，可以更好地被优化器优化，从而为优化器打开更大的优化范围空，实现更灵活的优化，帮助Cost模型选择更好的方案。这就是阿里巴巴在MaxCompute优化器上所做的。

优化规则

MaxCompute的基于成本的优化器做了很多优化，实现了如下图所示的各种优化，这里就不描述了。从下图可以看出，查询中有许多优化要做。所有这些优化都是整个系统引擎上的操作符，这些操作符也在改变图形，产生了很多等价的树。优化引擎通过成本模型选择最佳方案。

成本模型

什么是成本模型？其实Cost模型最需要关注的是自己的成本模型。每一个Cost模型都需要关注局部性，比如输入什么样的Cost，加入后会得到什么样的Cost，而不是关注全局情况。全局方案的成本是引擎累计的总成本。一个好的成本模型努力反映客观的物理实现，成本模型不需要与真实的完全相同。Cost模型的最终目的是区分方案的优劣，只需要能够选择更好的方案，不需要Cost绝对值的任何特性。目前，传统数据库的成本模型仍然是很久以前的。即使硬件结构发生了变化，只要是冯诺依曼架构且架构没有发生变化，就可以用Cost模型来选择最佳方案。

事实上，优化器中还可以考虑许多其他因素。比如规则方面，需要根据规则进行等价变换，最后根据Cost模型选择最佳方案。随着逻辑方案规模的扩大，要列举出所有可能的方案，需要花费大量的时间。特别是在MaxCompute上，希望逻辑执行计划越大越好，因为这样会给优化引擎更大的空范围，但是这样带来的是在枚举所有计划时，有些枚举的计划实际上是不必要的，可能已经处于未优化的情况。因此，如何有效地剪枝，如何避免对空的不必要的探索，也是实现一个好的优化器的考虑因素。此外，在探索空之间进行选择时，您可以将时间花在最有可能是最优化的空计划上。这可能是一个更好的选择，因为你不能希望通过NP-hard time选择最好的方案，而应该希望在有限的时间内选择更好的执行方案。所以在优化领域，不一定要找到最好的方案，而是要避免最坏的方案。

为什么基于成本的优化器对MaxCompute平台越来越重要？

这是因为阿里巴巴希望走出Hive的查询语句，提供更复杂的存储过程。如上图所示，更复杂的查询过程和存储过程可以通过if-else的变量赋值和预处理来编写，但是基于规则的优化器会因为贪婪算法而越来越偏向，很有可能最终得不到全局最优方案。逻辑规划的复杂性使得优化范围空更大，但也使得对优化器的要求更高，因此需要更好的基于成本的优化器来帮助选择更好的优化器。在分布式和非SQL场景中，基于成本的优化器不同于传统的单机优化器，因此需要对数据、操作和用户有更深入的了解，使基于成本的优化器更加智能。

了解数据

那么，从扩展的角度来看，理解数据是什么意思呢？在数据格式上，理解数据需要理解更多的数据索引和异构数据类型，理解结构化数据、非结构化数据和半结构化数据。在大数据场景中，数据是一个带有一些幂律属性和百万稀疏列的表，在这样的场景中需要进行优化；为了理解数据，我们还需要理解丰富的数据碎片方法，这些方法只在分布式场景中可用。数据碎片可以是范围/哈希/直接哈希，存储可以是列存储/列分组，需要层次分区。还需要了解完善的数据统计和运行时数据，需要了解直方图、Distinct值、数据量等等。

理解操作

从理解操作方面，需要更好的理解用户自定义函数，与优化器进行交互，让用户通过Annotation显示操作中数据的属性，从而进行全局优化。运行时会进行更多的优化，比如中间运行到某个阶段需要判断数据的大小，根据数据的大小选择并行化，根据数据的位置选择网络拓扑上的优化策略。它还可以平衡实时性、规模、性能、成本和可靠性，使用网络洗牌进行内存计算和流量计算。

了解用户

从理解用户的角度出发，需要了解用户场景对优化器的影响，了解多租户场景下用户在规模、性能、延迟、成本等方面的不同需求，让优化器在这样的场景下选择最佳解决方案；在生态上，优化器是核心优化引擎，希望在语言上更开放，希望与更多的语言和生态接口，希望提供强大的ide，为开发者提供完整的开发体验；最后希望在统一的平台上提供多种操作模式，让优化器真正成为操作的大脑。