头号作家丨SparkSQL优化案例学习

OPPO TECH 2021-12-06

1533

栏目介绍

“头号作家”是由OPPO工程师们解读时下热门技术的专题栏目。在这里，你不仅能看到最新最火的动态趋势，更能与OPPO的优秀工程师们共同学习技术干货。

头号作家

■ luckyfish

■ 专注大数据领域业务问题的解决和性能优化，提升大数据服务使用体验。6年大数据经验，做业务同学最信任的技术朋友！

最近因为海外业务核心报表准点率专项，我们在约1个半月的时间里做了一次集中的SparkSQL任务优化，加上平时的业务支持工作，又积攒了一些经验分享给大家。

简单来说，2021年Q1到Q2的海外核心报表准点率专项的优化成绩，涉及的35个核心任务做到了：

1）费用上，整体每年节省数百万。

2）运行时间上，平均单个任务节省运行时间是150.02分钟/天。

计算任务优化做为专项的核心措施之一，有力支撑了最终目标的达成。从下图可以看出，任务准点率（早上9点）也从优化前大概60%提升到现在较为稳定的95%，效果非常明显。

我们先简单了解一下有哪些关键优化措施：

减少大表扫描

● 加上必要的分区条件。

● 使用缓存或者临时表技术避免多次扫描大表。

增加并行度-聚合操作拆分

● 如count(distinct xxx)和sum操作分开运行后再union all，加大并行度。

避免数据倾斜

避免大量shuffle操作

● 尽可能使用广播关联，减少排序和网络读写操作。

以下是近8个月来一些新的优化点和对老经验的补充。

业务逻辑优化

性能优化的最高阶原则即是：干掉业务。很多时候技术人员常常陷入技术细节，希望从技术上让某个计算任务提升性能，但却忘了问一个问题：这个任务是否必要？这个时候需要跟业务方深入沟通，判断一个消耗资源巨大的任务的必要性，是否可能尽可能的进行精简甚至完全干掉。此次专项中，我们发现有一个表的生成由于计算量非常大，每天要计算上个月的全部数据，花费2个小时，而且各种参数都接近最优，因此跟业务同学进行了沟通，看下业务逻辑是否有优化的空间。

业务同学非常配合，很快发现已经有现成的数据表可以复用，从而一下轻松的节省了2个小时的巨量且重复计算。

在快速发展的业务和人员变动较为频繁的情况下，业务代码常常出现这种重复的情况，因此做优化的时候，我们最应该和上下游业务的同学紧密配合，判断存在性能瓶颈的业务逻辑存在的必要性，有概率彻底且轻松的解决问题。

读数据的优化

数据计算的第一步是读取数据。目前主要有2个方面需要注意。

充分利用所有分区条件

在大数据框架下，由于没有传统数据库的索引机制，分区条件成了核心的减少扫描量的办法。此次优化中，除了一般来说最核心的日期分区条件（如dayno=yyyymmdd）外，对于一些基础性的超大表，我们还需要充分利用更细化的分区条件，如hour, log_tag，act_code等等。

在优化dwd_browser_url_load_inc_d 任务的时候，我们发现其源表 obrowser.f_evt_obrowser_sdk_log_101 的分区条件有3个：

但代码里只有dayno和另外一个普通字段act_code的筛选条件。经过对历史数据的验证，发现所有符合 act_code in ('20083306', '20083185') 这一过滤条件的数据都落在了分区条件 category_code in ('10004','1000409') 里。经过与业务同学的确认，发现这个分区条件是可以安全添加的，添加后数据扫描范围大大下降。

同样的思路在总共5个任务中进行了应用，效果非常明显（当然也配合了其他的措施）：

数据分块的优化设置

Spark引擎读取数据表的时候本质是读取数据文件，而且可以通过规则将数据文件进行逻辑上的分块，以便每个子任务处理的数据量合适。在某些场合，读取数据后需要做的事情非常多的情况下，我们需要减小文件分块，让每个task（Spark任务的最小执行单元）处理的数据量减少，从而提升速度。

在优化 rpt_otheme_act_retention 任务的时候，由于从基础表里读出数据后，后续跟了一个lateral view explode(split(xxxxx)) 的操作，数据量膨胀了几倍甚至几十倍，导致处理速度较慢。一个task处理多少数据量，在spark中是通过 spark.sql.files.maxPartitionBytes参数确认，目前默认是256m 。通过将此参数设置为64m（以及关闭另外一个参数，后续会提到）我们成功的将运行时间减少了9成。

需要指出的是，这个参数不宜设置过小，因为太多task会引发资源调度的压力，最后抵消数据量减少带来的收益。尽量按当前设置的一倍进行减少实验，直到业务接受的水平即可。

另外一个需要注意的点是有部分任务使用的源文件是不可切割的格式，是否切割受文件压缩方式决定，常见的如GZIP,SNAPPY等是不可切割。当前Spark引擎默认是用zlib压缩orc格式的文件，因此按规范建表和写表的话，都是可以切割的。

算的优化

计算是任务的核心步骤。要优化，主要的原则是：1、少算；2、争取更多资源来计算。下面细讲。

少算 - 确保数据去重后再做关联

在优化 alg_meta_app_info 任务的时候，我们发现其最终的计算结果为1亿7千万行，然后需要出库到mysql，平均耗时超过了3个多小时。经过排查发现其中的一个作为关联的从表（右表）维表数据大量重复，数据一直异常。和业务同学确认后发现是源头异常，经过修正后的数据再关联只有6百万行，几分钟就完成了出库。

这个案例显示出关联数据重复带来的可怕效应，因此大家在做关联操作的时候，第一反应是确保至少维表数据要保证不重复，以免数倍甚至数十倍的数据膨胀效应。

少算 - 去除无意义的代码

在优化 tb_search_input_down 这个任务的时候，我们发现代码在出库到mysql的时候，在结尾加了一个 order by xxxx, xxx 。由于是写数据，这个order by并没有任何的意义。

去掉后配合hive切换到sparksql，耗时减少了9成。

类似的情况在日常支持工作中也偶有发生。order by这个操作，只有在查询展示的时候才有意义和开窗函数等极少数场景下有意义，而且极其耗费计算资源。在关系型数据库里，数据存放不同的物理或者逻辑位置，并不减少数据携带的信息，因此写入时对其排序是完全不必要的。在离线计算脚本里，除了row_number函数或者按顺序取样的情况下，都不要加order by或者类似的如sort by语句，没有任何意义。

争取更多资源用于计算 - 增加并行度

一个常见的反馈是任务从切换HiveSQL到SparkSQL后速度反而变慢了。这个问题的本质其实是SparkSQL任务默认的并发executor数量是400，然而HiveSQL的MR任务没有这个限制。在某些大计算量的情况下，资源占多少成了速度的关键因素。因此在某些特定的任务下，如巨量运算，我们需要通过一些参数的设置来加大对资源的获取以换取运行速度的提升。主要的方法有以下：

1）提升并发executor的数量，提升计算功率（即增加对内存和CPU的使用）。

主要参数如下：

spark.default.parallelism（默认400）：确认RDD（spark基础数据结构）的最大并行度。

spark.sql.shuffle.partitions（默认400）：确认SQL操作里的join或者聚合等操作的最大并行度。

spark.dynamicAllocation.maxExecutors（默认400）：确认最大分配的执行单元executor的数量。

为了任务使用资源的公平，数据平台团队设定了以上的默认值，以避免出现某些任务占用了业务队列的全部资源，造成阻塞和浪费。但某些场景下，对于一些关键任务，大家可以通过配套增加以上三个参数，以实现对计算资源上限的提升，总的原则是谨慎调大。

2）增大文件分块数量以便增加分配task数量，减少每个task的运行时间。

少算 - 一次就成功，减少任务重试

SparkSQL任务是运行在分布式系统上，系统节点的健康程度和负载情况差别很大，加上处理的数据情况也很难预料，因此重试的发生非常普遍。从job到stage到task，每个层面都存在重试。重试带来的是时间和资源的双重浪费。如果观察任务日志的时候发现重试情况非常多，虽然最终任务是成功了，但仍然需要调优，减少重试。目前遇到最频繁的重试即shuffle失败引发重试，下面详解一下。

一个stage意味着一次N个计算节点上的数据进行重新分发汇总到M个计算节点（一般来说N>M），即shuffle动作发生了。这就意味着大量的数据读写和网络传输，失败的概率通常较高，尤其在任务高峰时期，机器承担的任务非常重，常常出现无法在超时时间内响应，造成shuffle的失败。目前我们集群遇到的主要失败都集中在shuffle fetch阶段。即需要shuffle的数据生成后，下一阶段的机器要取这些数据，取的过程发生各种异常，当然主要是机器无法响应。在SparkUI上我们能看到类似以下的截图就知道是大概率是这种问题了。

以下参数可以缓解这类失败：

写的优化

同时解决小文件和写任务倾斜

小文件是大数据任务里的梦魇，它是分布式系统固有的病根。目前国内和印度我们都开发了对应的小文件自动合并功能，不用太过操心。但新加坡由于要使用EMR集群提供低成本的计算资源，不得不使用其附带版本的Spark引擎，该版本暂时无法提供自动合并功能，依赖于业务同学进行管控。我们也在催促其适配，但可能进展会比较慢。在此之前需要我们了解对应的处理方法，这里分两种情况。

1）固定分区

这时候我们可以暂时通过手动指定hint来解决，如:

这个用法会在任务最后阶段生成一个新的stage来交换数据以实现更集中地处理数据。这里N决定了最后生成的文件数量。这里要注意N太大，会导致文件太多，N太小会导致单个任务写任务的数据量太大，速度慢，因此要适中。目前的经验值是，一个任务写大概1千万行数据比较适中。

2）动态分区

在优化 oppo_dwd_browser_url_load_inc_d 任务的时候，由于是动态分区写入，每个动态分区最后生成的文件数量是大体一致的，如果我们按上面的办法使用repartition（N），每个最终生成的分区都会生成N个文件，这对于大的分区是可以接受的，但对于其他小数据量的分区，同样制造了大量的小文件。这种情况下，如果我们把N调整很小，对于大数据量的分区，会造成上面提到单个任务写的数据量太大，速度慢，从而陷入两难的境地。

针对这个情况最好的做法是使用distribute by + case when语句。具体用法如下图示例（distribute部分写在代码最后面）：

稍微解释一下ceil(rand()*N)的意义。这里rand()会产生一个0到1之间的小数，通过乘以N再取ceil，能实现从1到N的随机整数的生成。这样，每一行数据都会分发到1....N这里的某一组，相当于加了一个虚拟的分组使用的列。

distribute by + case when同时解决了某些巨型动态分区里数据量大需要多个任务来写数据的需求，又照顾到了整体的大多瘦分区的文件数量问题，是比较完善的方案。

文件格式

在优化 f_evt_os_launch_new 这个任务的时候，在写的阶段，无论如何优化都没有质的提升。后面查看了文件格式，因为历史原因还是最基础的text格式。我之前做过对比，同一个数据表，text（行式存储）和orc（列式存储）的实际容量占用平均是20：1。这不仅意味着text格式的存储费用是20倍差异，更导致在生成数据的时候写得更慢，且下游读数据的效率也大大降低，基本上来说是百害而无一利。这通常是历史原因导致的。在和业务同学沟通后，新建了orc格式的表替换老格式表，优化目的轻松达到了。

总结：除极端情况（text格式作为接口文件需要被外界直接读取）外，所有大数据表都建议存储为orc格式，这也是当前OPPO数据平台的官方推荐格式。

资源申请优化

由于当前我们的SparkSQL任务主要使用Livy组件提交任务，而Livy提交Spark任务到获取执行资源（driver和executor）都是需要时间的，大概20秒到2分钟，视当时集群的繁忙程度。执行资源的生成，分配及销毁都需要花时间，所以即使这些时间占比较小，但遇到资源紧张的空档，也可能造成不必要的等待。尤其是一些逻辑复杂的脚本，步骤可能有十多个，光是申请资源就花到20分钟，完全不必要。最佳实践是，将逻辑关系非常紧密的SQL语句都写到一个SQL里，以分号分隔。如果前面的某段SQL失败了，整个语句也会失败，因此不用再做更多的异常处理。

在凌晨任务高峰，资源紧张的环境下，这类优化效果非常明显。毕竟对于一些早期的关键任务，节省5分钟，意味着后续几百个任务都提前运行5分钟，对于提升报表的准点率效果明显。

这方面的优化本身有两种场景。

代码块本身分散，可直接拼接一起

这个比较常见，也比较好处理，举例说明即可。如下分块的老代码：

建议优化为：

这样代码本身更简洁，最重要的是节省了可能十到二十分钟的资源申请时间。

代码需要进行适当改造才能拼接

在优化 f_app_version_res_v2 任务的时候，我们面临的情况是，首先需要将mysql表导出到hive表的某个分区，然后将这个分区的数据再依次复制到其他11个分区里，一共要做12次离线计算。光是资源申请调度的开销就有约20多分钟左右。经过仔细分析，我们使用笛卡尔积的方式巧妙的将另外11次的复制操作转换为1次，大大节省了申请资源的时间开销。

解释一下上述代码的特别点。首先我们将12个分区的值通过explode+split存到一个全局临时视图里，然后通过笛卡尔关联的方式实现一份数据按分区值复制12份，最后依赖动态分区写入机制，实现一个SQL一次性写12个分区，避免了12个分区的数据串行写入，而且只需要一次申请资源。

此任务的另外一个小优化，也是使用了动态分区，将7次循环的SQL转换成了1次的SQL：

经过优化，加上对某个大维表的全量抽取变成增量抽取，一共完成了约7成时间的优化。由于核心维表几乎是所有任务的上游，因此这部分的时间节省含金量相当高。

这部分原则可以简化为：勿以善小而不为。

规范优化

Hive表与mysql表交互

目前一类常见的场景是将大数据计算出来的数据出库到mysql中，或者将mysql的数据入库为hive表。很多同事会发现这类与mysql交互的任务常常持续数小时，非常苦恼。本质上这里的慢并不是在于大数据部分，瓶颈基本上都是mysql端。目前的经验是，单次与mysql的读写规模需要控制在100万条以内，过大的数据量会导致任务失败或者任务时间超长。这类任务大家要牢记：将计算和IO都尽量转移到hive表上。具体操作上，需要把握以下两点：

1、大mysql入库hive表，必须设计增量入库方案。一般来讲mysql的表设计相对更严格和规范，总有办法找到增量同步的解决方案，千万不能图省事一股脑的全量覆盖。随着业务量增长，任务总有一天会慢得不可接受或者失败（mysql连接有时长限制）。如果有计算逻辑，尽量放到hive表里操作，而不要在mysql库里跑。在优化f_app_version_res_v2任务里某个维表大小约500万，光出库时长就接近1个小时，优化后的增量同步仅需10分钟。

2、hive出库到mysql表供报表查询的情况下，超过数百万的情况就极大可能不合理。因为这样的数据量放到mysql表里要去查询本身就不算很合适。数百万对于mysql来说是一个比较舒服的规模（不考虑分库分表等操作），再大的话尽量尝试使用大数据方案。一个比较好的选择是presto引擎直接查询hive库表，完全不用出库。

代码格式

虽然代码怎么写不会影响最后的执行细节，但对于排错和优化特别有用。希望大家都能成为对代码有追求的码农，写出有美感的代码。在这里推荐一些基础性原则：

● 一个字段独立一行

● 一个条件独立一行

● 一个case when then（或者类似独立性强的场景）独立一行

● 核心关键字（select, join, where,group by等）保持严格对齐

● 同级子表或者子查询保持对齐

● 对代码在适当的位置添加必要的注释

● 如不是极端情况禁止写select * ，将字段都列出来

以下是代码示例。基本上一眼就能看核心信息，有类似语法错误的提示也几乎瞬间就能定位到问题。

参数滥用问题

大数据任务由于环境复杂，业务逻辑复杂，因此相对其他系统来说有相当多的参数需要进行优化。但经常发生的一个情况，无论是因为对参数理解不准确或者是复制模板代码导致的参数滥用问题比较严重。主要的问题有：

● 参数使用错误导致性能下降

● 参数设置不当导致资源浪费

在优化oppo_dws_browser_duration_user_sum_inc_d 任务的时候，我发现相比在自己的日常环境下手动跑完全一样的代码，性能能好转近一倍。仔细排查后发现脚本里有一个参数：spark.sql.hive.convertMetastoreOrc=false 。此参数意为是否使用Spark引擎内建的orc格式处理方法来读写由hiveSQL语法创建的orc表，默认是true，即会使用，这会使读写orc表效率提升很多，但少数情况下存在兼容性问题。可能由于此代码复制了某个有兼容性问题的模板才引入了此参数，但却大大的降低了性能。

在另外的一些场景下，我们发现参数被设置得极为浪费。比如我们的Spark Executor的内存为4G1核（外加1G的overhead内存），这是经过慎重考虑的默认参数。但可能部分同学遇到任务失败等问题或者希望加快运行速度，于是在网上搜索到一些不够严谨的资料后就将这些参数使劲调大，最后即使没有让情况更坏，也会造成以下后果：

● 使用资源的浪费。由于我们的Executor默认最大的并发是400个，所以当你调大1G的内存的时候，意味着最大可能并发占用400G内存，这是相当大的资源支出。

● 引发OOM。各参数之间有一定配套关系，比如memory和memoryOverhead两个参数，如果memory很大，Overhead很小，比例相差过大，会引发Overhead内存不够而任务失败。

● 集群资源的浪费。由于集群内的机器的CPU和内存有一定比例，且申请executor进程需要成套，比如 48个核+256G内存，这个时候如果你设置一个executor为10个核40G内存10G Overhead内存，此类机型最多生成4个executor，其他8个核就生生的浪费了。

参数设置的2大原则请大家牢记：

1）如无必要，勿增参数。

2）如增参数，务必理解。

优化思路图

为方便大家记忆，我整理了一个思维导图。