????????今天跟大家分享的内容是maxcompute的优化内容，希望能够起到抛砖引玉的作用。我将从三个方面开展这次的主题分享，第一个部分是max和hive的对比；第二部分是调优方法论，主要是想总结一下我们从代码提交到执行结束，都有哪些位置可能会出现问题，一般情况下又是怎么解决的；最后是结合具体案例，对长尾问题的一些思考。

2. MaxCompute对比Hive

????????首先，先对其简单做个介绍，MaxCompute主要是应用在批量结构化数据的存储和计算，主要是数据仓库的和大数据的分析建模方向。这个是MaxCompute系统架构，可以看到MaxCompute支持SQL查询计算，MapReduce计算，支持机器学习，和面向迭代的图计算，对外提供三种Java API接口，非常简洁。MaxCompute和Hive一样可以使用SQL、UDF以及MapReduce开发。基于MaxCompute的无服务器的设计思路，用户只需关心业务和数据，不需要关心底层分布式架构及运维。

????????对于MaxCompute，可将飞天视为类似于Hadoop的框架，如Hive基于Hadoop，那么可以理解为MaxCompute基于飞天系统。大家都知道hive依赖hdfs进行数据存储，依赖yarn进行资源管理，正好也对应着盘古和伏羲。为什么阿里要去hadoop，做一个自己的系统，我的看法是hadoop开源更新不可控，无法满足快速迭代的业务需求，不如另起炉灶，但是底层架构有很多相似之处。

????????Maxcompute和hive都是有自己的客户端的，另外maxcompute对hive sql也是有兼容的，只需要一行设置即可。下面是他们的数据源和数据格式，阿里肯定支持自家的ots和oss，均支持hdfs。Maxcompute支持手动和自动的调优，hive仅仅支持手动调优。

????????找到了一个2016年的数据，100TB数据全排序，maxcompute比竞品要快要便宜。

3.调优方法论

?????????我们优化一个sql的前提是这个代码能够成功运行，需要知道提交一个任务都需要经历哪几个阶段。一个sql一般我们叫做一个odps作业，在我们提交一个odps作业之后，都会发生哪些动作呢？我们将整个sql的运行分为预处理、编译、执行和结束这四个阶段。

????????以上四个阶段只有预处理阶段没有logview产生，而且预处理阶段一般不会出现问题，这个阶段我们不做考虑，其他三个过程，又会因哪些原因发生阻塞呢，根据细分规则，我们接下来看如果我们的作业在这些位置被卡住，将会有哪些表现，并且我们分析是哪些原因导致的，以及可以采取哪些解决方案。

3.1 编译阶段

????????首先，编译阶段指的是什么时候呢？指的是已经了产生logview，但还没有执行计划。根据 logview 的子状态（SubStatusHistory）可以进一步细分为调度、优化、生成物理执行计划、数据跨集群复制等子阶段。编译阶段的问题主要表现为在某个子阶段卡住，即作业长时间停留在某一个子阶段。下面将介绍作业停留在每个子阶段的表现以及可能原因和解决方法。

3.1.1?调度阶段

【特征】子状态为“Waiting for cluster resource”，作业排队等待被编译。

【该阶段作业卡住的可能原因 1 】计算集群资源紧缺。

【解决方法】查看计算集群的状态，需要等待计算集群的资源。

【该阶段作业卡住的可能原因 2 】编译资源池资源不够：可能有人不小心用脚本一次提交太多作业，把编译资源池占满了。如果发现这种情况，要及时找值班同学咨询。

3.1.2 优化阶段

【特征】子状态为“SQLTask is optimizing query”，优化器正在优化执行计划。

【该阶段作业卡住的可能原因】如果执行计划复杂，需要等待较长时间做优化。

【解决方法】一般可接受的时间是10分钟以内，如果真的太长时间不退出，基本上可以认为是 odps 的 bug

3.1.3 生成物理执行计划阶段

【特征】子状态为“SQLTask is generating execution plan”。

【该阶段作业卡住的可能原因 1 】读取的分区太多。每个分区需要去根据分区信息来决定处理方式，决定 split，并且会写到生成的执行计划中。

【解决方法】需要好好设计 SQL，减少分区的数量，包括：分区裁剪、筛除不需要读的分区、把大作业拆成小作业。

【该阶段作业卡住的可能原因 2 】小文件太多。ODPS 会根据文件大小决定 split，小文件多了会导致计算 split 的过程耗时增加。

【解决方法】

避免小文件，可以执行一次?alter table merge smallfiles;?让 odps 把小文件 merge 起来

【注意】上面提到的“分区太多，小文件太多”不是指几十、几百个。基本都是要上万，上十万才会对生成物理执行计划的时间产生较大影响。

3.1.4 数据跨集群复制阶段

【特征】子状态列表里面出现多次“Task rerun”，result 里有错误信息“FAILED: ODPS-0110141:Data version exception”。作业看似失败了，实际还在执行，说明作业正在做数据的跨集群复制。

【该阶段作业卡住的可能原因 1 】project 刚做集群迁移，往往前一两天有大量需要跨集群复制的作业。

【解决方法】这种情况是预期中的跨集群复制，需要用户等待。

【该阶段作业卡住的可能原因 2 】可能是作业提交错集群，或者是中间 project 做过迁移，分区过滤没做好，读取了一些比较老的分区。

【解决方法】检查作业提交的集群是否正确。过滤掉不必要读取的老分区。

?3.2 执行阶段

????????执行阶段一般指的是：logview 的 detail 界面有执行计划（执行计划没有全都绿掉），且作业状态还是 Running。执行阶段卡住或执行时间比预期长的主要原因有等待资源，数据倾斜，UDF 执行低效，数据膨胀等等，下面将具体介绍每种情况的特征和解决思路。

3.2.1 等待资源

【特征】instance 处于 Ready 状态，或部分 instance 是 Running，部分是 Ready 状态。

【解决思路】

确定排队状态是否正常。可以通过 logview 的排队信息“Queue”看作业在队列的位置，作业有排队是正常的。作业的调度顺序不仅与作业提交时间、优先级有关，还和作业所需内存或CPU资源大小能否被满足有关，因此合理设置作业的参数很重要。

3.2.2 数据倾斜

【特征】task 中大多数 instance 都已经结束了，但是有某几个 instance 却迟迟不结束（长尾），这些?instance?运行的慢，可能是因为处理的数据多。

【解决方法】需要找到造成数据倾斜的具体位置，对症下药，利用 Logveiw2.0 查看任务执行图和 instance 运行情况来定位长尾实例。在确定造成数据倾斜的实例、数据来源等信息后，用户需要针对性的对代码甚至算法做一定的修改。

3.2.3 UDF执行低效

这里的 UDF 泛指各种用户自定义的扩展，包括UDF，UDAF，UDTF，UDJ，UDT等。

【特征】某个 task 执行效率低，且该 task 中有用户自定义的扩展。甚至是 UDF 的执行超时报错。

【排查方法】任务报错时，可以在?MaxCompute Studio?中快速通过 DAG 图判断报错的 task 中是否包含 UDF。此外，在 task 的 stdout 日志里，UDF 框架会打印 UDF 输入的记录数、输出记录数、以及处理时间，一般来讲，正常情况 Speed(records/s) 在百万或者十万级别，如果降到万级别，那么基本上就有性能问题了。

【解决思路】

检查 UDF 是否有 bug，有时候 bug 是由于某些特定的数据值引起的，比如出现某个值的时候会引起死循环。
检查 UDF 函数是否与内置函数同名。内置函数是有可能被同名 UDF 覆盖的。有相似功能的内置函数的情况下，尽可能不要使用 UDF。内置函数一般经过验证，实现比较高效，并且内置函数对优化器而言是白盒，能够做更多的优化。
UDF效率低的原因也可能是内存原因，某些UDF在内存计算、排序的数据量比较大时，会报内存溢出错误；内存不足也会导致gc频率过高。可以尝试调整内存参数解决内存不足的问题。

3.2.4 数据膨胀

【特征】task 的输出数据量比输入数据量大很多。比如 1G 的数据经过处理，变成了 1T，在一个 instance 下处理 1T 的数据，运行效率肯定会大大降低。输入输出数据量体现在 Task 的 I/O Record 和?I/O Bytes 这两项：

【解决思路】

检查代码是否有 bug；
JOIN 条件是不是写错了，变成笛卡尔积了；
检查 Aggregation聚合引起的数据膨胀。因为大多数 aggregator聚合是 recursive递归的，中间结果先做了一层 merge，中间结果不大，而且大多数 aggregator 的计算复杂度比较低，即使数据量不小，也能较快完成。所以通常情况下这些操作问题不大，但是select 中使用 aggregation 按照不同维度做 distinct，每一次 distinct 都会使数据做一次膨胀；使用?Grouping Sets (CUBE and ROLLUP)?，中间数据可能会扩展很多倍。有些操作如?collect_list、median?操作需要把全量中间数据都保留下来，可能会产生问题。
避免join引起的数据膨胀。比如：两个表 join，左表是人口数据，数据量很大，但是由于并行度足够，效率可观。右表是个维表，记录每种性别对应的一些信息（比如每种性别可能的坏毛病），虽然只有两种性别，但是每种都包含数百行。那么如果直接按照性别来 join，可能会让左表膨胀数百倍。要解决这个问题，可以考虑先将右表的行做聚合，变成两行数据，这样 join 的结果就不会膨胀了。

3.3 结束阶段

????????最后是结束阶段：大部分 SQL 作业在 Fuxi 作业结束后即停止。但有时 Fuxi 作业结束时，作业总体进度仍然处于运行状态。如下图中的 logview，右侧 Job Details 页面显示 Fuxi 作业所有阶段为 Terminated，但左侧代表作业整体进度的 Status 仍然显示 Running，造成这种现象的情况一般分为两种：一个 SQL 作业可能包含多个 Fuxi 作业，比如子查询多阶段执行，作业输出小文件过多导致的自动合并作业；Fuxi 作业结束后，SQL 在结束阶段运行于控制集群的逻辑占用时间较长，比如更新动态分区的元数据。

????????下面将举例介绍几种典型。

?3.3.1?子查询多阶段执行

????????大部分情况下，MaxCompute SQL 的子查询会被编译进同一个 Fuxi DAG，即所有子查询和主查询都通过一个 Fuxi 作业完成。但也有一些特殊子查询需要先将子查询单独执行。

SELECT product, sum(price) FROM sales WHERE ds in (SELECT DISTINCT ds FROM t_ds_set) GROUP BY product;

????????子查询?SELECT DISTINCT ds FROM t_ds_set?先执行，其结果需要被用来做分区裁剪，来优化主查询需要读取的分区数。这两次运行就会产生两个 fuxi 作业。

3.3.2 过多小文件

????????小文件主要带来存储和计算两方面问题。存储方面，小文件过多会给 Pangu 文件系统带来一定的压力，且影响空间的有效利用；计算方面，ODPS 处理单个大文件比处理多个小文件更有效率，小文件过多会影响整体的计算执行性能。因此，为了避免系统产生过多小文件，SQL 作业在结束时会针对一定条件自动触发合并小文件的操作。小文件的判定，是通过参数?odps.merge.smallfile.filesize.threshold?来设置的，该配置默认为 32MB，小于这个阈值的文件都会被判定为小文件。

????????自动合并小文件多出来的 merge task，虽然会增加当前作业整体执行时间，但是会让结果表在合并后产生的文件数和文件大小更合理，从而避免对文件系统产生过大压力，也使得表被后续的作业使用时，拥有更好的读取性能。如果结果实际不大，但是文件数过多，那么最好是先检查下小文件阈值配置是否正确。

3.3.3 动态分区元数据更新

????????Fuxi 作业执行完后，有可能还有一些元数据操作。比如要把结果数据挪到特定目录去，然后把表的元数据更新。有可能动态分区输出了太多分区，那么还是可能会消耗一定的时间的。

????????Fuxi 作业执行结束后，仍需要一段时间进行表的元数据更新。

4.计算长尾案例分享

????????最后，我们来讨论一些长尾的案例。长尾问题是分布式计算里最常见的问题之一，也是典型的疑难杂症。究其原因，是因为数据分布不均，导致各个节点的工作量不同，整个任务就需要等最慢的节点完成才能完成。我们从长尾现象和原因入手，针对实际业务处理过程中，遇到的长尾现象，分享给大家优化思路和解决方案。

????????正式进入这个主题之前，我们需要知道几个概念。我们提交到maxcompute上的任务都是由1个到n个fuxi job组成的。图中的任务就是由3个fuxi job来完成的，每一个fuxi job是由1个到n个fuxi task组成的。一个fuxi task是由1个到n个fuxi instance完成。复习了这些概念之后，我们进入下面的主题。

????????我们如何查看任务是否发生了长尾？长尾的现象是什么？logview是帮助我们排查这些问题重要的工具和手段，我们一般会逐一点击task，页面下端会显示这个task任务的所有instance的运行情况，有运行时间，起止时间，如果看到红框内有long tails说明发生了长尾。系统是怎么认定发生了长尾呢？是因为这个任务执行完成之后，系统会对每个instance时间做统计，求出一个平均值，一般来说如果某些instance是这个平均时间的两倍以上，就认定发生了长尾现象（logview1.0是三倍），logview2.0异常情况区分了长尾和数据倾斜，两个异常的关注点是不一样，数据倾斜关注的是数据的IO数据量大小，长尾关注的instance的执行时间。

????????我们看一下为什么会发生长尾呢？我们举一个例子，这个例子就是简单的wordcount实际场景的例子。这个场景下会用到两个task，一个map，一个reduce。我们看一下整个wordcount的流程。

????????整个wordcount的流程，输入的数据在起始端的时候会被分片，分片的数据进入map instance以后，map instance会去读每一条记录，然后把key放在第一个参数位置，value会记录上1，下一个阶段是shuffer阶段，shuffer阶段的前一个部分会做一个排序，排序之后还会做一个combiner的操作，combiner阶段会把相同的key的value累加起来，组成一个新的key-value值，在shuffer的后半段，相同key的记录会进入相同的reduce instance上去，reduce阶段会再做一个combiner操作，最后输出数据。我们看一下哪些阶段容易发生长尾问题，首先是map阶段，map阶段有可能发生长尾，但是在实际应用中，这种现象不经常发生，如果发生了，大家可以一块跟踪一下整个问题。下一个容易发生长尾的地方是shuffer dispatch的地方，就是shuffer后期分配reduce的阶段。相同的key会分配到相同的reduce上去处理，如果一个key的数据非常非常多，那这个reduce真的是非常幸运了，他完成的时间可能会比其他的instance要长，这个时候长尾现象就发生了。

4.1 group by长尾问题

????????我们接着看一下我们遇到的长尾的场景，以及怎么做优化？第一个场景就是group by ，比如我们app中某一个商品卖得很好，我们想要计算这个商品的pu，pv/uv，往往会做一个group by的操作。和我们刚才的wordcount例子一样，往往会是一个map+reduce，发生长尾的地方往往是shuffer dispatch的地方。所以如果发生了长尾，我们应该怎么解决呢？我们思考能不能把reduce这个hot key打散，不让他分发到同一个reduce instance。

????????系统提供了一个参数，这个参数会在shuffer dispatch的时候，算法里面会加入一个随机的因素，会更加均匀的将数据分发到reduce上。大家可以看到对比，在这个参数开启前，dag图是一个map+reduce，开启之后是一个map+两个reduce。这跟我们刚才的想法是一致的，中间的reduce的作用就是将hot key给打散。但是需要注意的是，这个参数仅仅对group by有效，如果长尾现象并不严重，用这种方法增加一次reduce，时间消耗反而更大。

4.2?Count Distinct长尾问题?

?????????再看另一个场景，count distinct长尾。当我们计算商品购买uv的时候，固定的特殊值比较多，也会发生长尾现象。我们可以不用distinct，用group by来改写。但是我们改写之后由26s变成了33秒。我们初衷是避免使用distinct，使用group by，并且开启了参数。这个实验充分说明如果长尾不是很严重，是否开启这个参数是值得商榷的。那还能怎么做呢？

????????我们可以使用where 先把这个特殊值（比如null）给过滤掉，我们count完了之后再加1就好了，所以这些优化是需要根据用户具体的数据结构、类型规模来决定。

?4.3?动态分区长尾问题

????????下一个场景是动态分区，在动态分区的场景里面，非常有可能发生长尾。上面是一个动态分区写入的一个语句，假如我们有n个map instance，我们的目标分区有m个，可能会产生n*m个小文件。大家都知道小文件过多会给我们后续查询带来很多的问题，所以系统针对这种情况是做了优化的，会尽可能避免产生过多的小文件。我们看一下执行计划，下面红框中有两个值，第一个是我们的分区，第二个值具体是0-10之间的一个数，系统为了避免小文件产生，做了一个reshuffer。reshuffer具体做了什么呢？会在shuffer dispatch的时候，相同目标分区的数据，会由最多10个reduce来处理，实际上减少了reduce的个数，也减少了小文件的产生。

????????但是这个任务花费了3分钟，并且发生了长尾现象。我们想既可以避免小文件产生，又不想发生长尾，可以怎么做呢？

?????????手动的把reshuffer开关给关掉（这个开关默认是打开的）。任务在10s左右就跑完了，效果还是很明显的。所以如果目标分区不多，建议先关闭reshuttle;任务执行完毕后，手动执行MergeTask，减少小文件。这个方法比较折中，大家可以尝试一下。

4.4 Join长尾问题?

????????join长尾，join长尾还是相当普遍的，每个join也会被解释器解释成一个mr任务，map端读取join两边表，reduce做join操作，没有办法避免shuffer dispatch的时候产生的hot key的问题。?

????????很多人会建议用mapjoin，mapjoin的原理是将join的操作由reduce阶段移到map端去做，会把小表放到每一个map instance上面，每一个map instance都拥有小表所有的内容，可以在map本地进行join。但是可想而知，如果一个map instance能存下小表所有内容，单机内存要求还是非常高的。所以一般会有这个限制，mapjoin的小表占用的内存总和不得超过512MB。???

但是我们实际使用场景中，表的大小超过512，还是非常常见的。很多人会说我两边数据都很大，没有办法使用mapjoin。那应该怎么办呢，我们可以尝试使用分而治之的方法解决。长尾的问题就是热点key太多，发送到一个reduce instance，导致运行时间过长，我们可以把热点key找出来，或者找到一个范围。将热点key和非热点key分开处理。