转载自：

https://blog.csdn.net/xwc35047/article/details/71039830

http://www.importnew.com/26541.html

https://www.cnblogs.com/dreamfly2016/p/5720526.html

==========================================================================================

1、Spark调优背景

目前Zeppelin已经上线一段时间，Spark作为底层SQL执行引擎，需要进行整体性能调优，来提高SQL查询效率。本文主要给出调优的结论，因为涉及参数很多，故没有很细粒度调优，但整体调优方向是可以得出的。

环境：服务器600+，spark 2.0.2，hadoop 2.6.0

2、调优结果

调优随机选取线上9条SQL，表横轴是调优测试项目，测试在集群空闲情况下进行，后一个的测试都是叠加前面测试参数。从数据可参数经过调优，理想环境下性能可提高50%到300%

3、 下面为调优分享PPT

1）一图概览

2） Spark集群优化——数据本地性

3）Spark集群优化——存储格式选择

4）Spark参数优化——计算资源

5） Spark参数优化——并行度

6）Spark参数优化——offheap内存

7）Spark参数优化——大小表join

8）Spark参数优化——其他

9） Spark参数优化——shuffle过程

10）Spark代码优化——RDD复用

11）Spark代码优化——选择合适算子

12） Spark代码优化——shuffle算子并行度调优

13）Spark代码优化——数据倾斜

14）Spark代码优化——优化数据结构

15）Spark代码优化——使用DateSet API

16）Spark代码优化——使用DateSet API

17） 目前Spark的瓶颈——内存

18） 目前Spark的瓶颈——内存

3、总结

调优参数虽名目多样，但最终目的是提高CPU利用率，降低带宽IO，提高缓存命中率，减少数据落盘。
不同数据量的最优参数都不相同，调优目的是让参数适应数据的量级以最大程度利用资源，经调优发现并不是所有参数有效，有的参数的效果也不明显，最后折中推荐如下调优参数以适应绝大多数SQL情况，个别SQL需要用户单独调参优化。（以下参数主要用于Spark Thriftserver，仅供参考）

参数	含义	默认值	调优值
spark.sql.shuffle.partitions	并发度	200	800
spark.executor.overhead.memory	executor堆外内存	512m	1.5g
spark.executor.memory	executor堆内存	1g	9g
spark.executor.cores	executor拥有的core数	1	3
spark.locality.wait.process	进程内等待时间	3	3
spark.locality.wait.node	节点内等待时间	3	8
spark.locality.wait.rack	机架内等待时间	3	5
spark.rpc.askTimeout	rpc超时时间	10	1000
spark.sql.autoBroadcastJoinThreshold	小表需要broadcast的大小阈值	10485760	33554432
spark.sql.hive.convertCTAS	创建表是否使用默认格式	false	true
spark.sql.sources.default	默认数据源格式	parquet	orc
spark.sql.files.openCostInBytes	小文件合并阈值	4194304	6291456
spark.sql.orc.filterPushdown	orc格式表是否谓词下推	false	true
spark.shuffle.sort.bypassMergeThreshold	shuffle read task阈值，小于该值则shuffle write过程不进行排序	200	600
spark.shuffle.io.retryWait	每次重试拉取数据的等待间隔	5	30
spark.shuffle.io.maxRetries	拉取数据重试次数	3	10

=========================================================================================

0、背景

上周四接到反馈，集群部分 spark 任务执行很慢，且经常出错，参数改来改去怎么都无法优化其性能和解决频繁随机报错的问题。

看了下任务的历史运行情况，平均时间 3h 左右，而且极其不稳定，偶尔还会报错：

1、优化思路

任务的运行时间跟什么有关？

（1）数据源大小差异

在有限的计算下，job的运行时长和数据量大小正相关，在本例中，数据量大小基本稳定，可以排除是日志量级波动导致的问题：

（2）代码本身逻辑缺陷

比如代码里重复创建、初始化变量、环境、RDD资源等，随意持久化数据等，大量使用 shuffle 算子等，比如reduceByKey、join等算子。

在这份100行的代码里，一共有 3 次 shuffle 操作，任务被 spark driver 切分成了 4 个 stage 串行执行，代码位置如下：

咱们需要做的就是从算法和业务角度尽可能减少 shuffle 和 stage，提升并行计算性能，这块是个大的话题，本次不展开详述。

（3）参数设置不合理

这块技巧相对通用，咱们来看看之前的核心参数设置：

num-executors=10 || 20 ，executor-cores=1 || 2， executor-memory= 10 || 20，driver-memory=20，spark.default.parallelism=64

假设咱们的 spark 队列资源情况如下：

memory=1T，cores=400

参数怎么设置在这里就有些技巧了，首先得明白 spark 资源的分配和使用原理：

在默认的非动态资源分配场景下， spark 是预申请资源，任务还没起跑就独占资源，一直到整个 job 所有 task 结束，比如你跳板机起了一个 spark-shell 一直没退出，也没执行任务，那也会一直占有所有申请的资源。（如果设置了 num-executors，动态资源分配会失效）

注意上面这句话，spark 的资源使用分配方式和 mapreduce/hive 是有很大差别的，如果不理解这个问题就会在参数设置上引发其它问题。

比如 executor-cores 设多少合适？少了任务并行度不行，多了会把整个队列资源独占耗光，其他同学的任务都无法执行，比如上面那个任务，在 num-executors=20 executor-cores=1 executor-memory= 10 的情况下，会独占20个cores，200G内存，一直持续3个小时。

那针对本case中的任务，结合咱们现有的资源，如何设置这 5 个核心参数呢？

1） executor_cores*num_executors 不宜太小或太大！一般不超过总队列 cores 的 25%，比如队列总 cores 400，最大不要超过100，最小不建议低于 40，除非日志量很小。

2） executor_cores 不宜为1！否则 work 进程中线程数过少，一般 2~4 为宜。

3） executor_memory 一般 6~10g 为宜，最大不超过 20G，否则会导致 GC 代价过高，或资源浪费严重。

4） spark_parallelism 一般为 executor_cores*num_executors 的 1~4 倍，系统默认值 64，不设置的话会导致 task 很多的时候被分批串行执行，或大量 cores 空闲，资源浪费严重。

5） driver-memory 早前有同学设置 20G，其实 driver 不做任何计算和存储，只是下发任务与yarn资源管理器和task交互，除非你是 spark-shell，否则一般 1-2g 就够了。

Spark Memory Manager：

6）spark.shuffle.memoryFraction(默认 0.2) ，也叫 ExecutionMemory。这片内存区域是为了解决 shuffles,joins, sorts and aggregations 过程中为了避免频繁IO需要的buffer。如果你的程序有大量这类操作可以适当调高。

7）spark.storage.memoryFraction(默认0.6)，也叫 StorageMemory。这片内存区域是为了解决 block cache(就是你显示调用dd.cache, rdd.persist等方法), 还有就是broadcasts,以及task results的存储。可以通过参数，如果你大量调用了持久化操作或广播变量，那可以适当调高它。

8）OtherMemory，给系统预留的，因为程序本身运行也是需要内存的， (默认为0.2)。Other memory在1.6也做了调整，保证至少有300m可用。你也可以手动设置 spark.testing.reservedMemory . 然后把实际可用内存减去这个reservedMemory得到 usableMemory。 ExecutionMemory 和 StorageMemory 会共享usableMemory * 0.75的内存。0.75可以通过 新参数 spark.memory.fraction 设置。目前spark.memory.storageFraction 默认值是0.5,所以ExecutionMemory，StorageMemory默认情况是均分上面提到的可用内存的。

例如，如果需要加载大的字典文件，可以增大executor中 StorageMemory 的大小，这样就可以避免全局字典换入换出，减少GC，在这种情况下，我们相当于用内存资源来换取了执行效率。

最终优化后的参数如下：

效果如下：

（4）通过执行日志分析性能瓶颈

最后的任务还需要一个小时，那这一个小时究竟耗在哪了？按我的经验和理解，一般单天的数据如果不是太大，不涉及复杂迭代计算，不应该超过半小时才对。

由于集群的 Spark History Server 还没安装调试好，没法通过 spark web UI 查看历史任务的可视化执行细节，所以我写了个小脚本分析了下前后具体的计算耗时信息，可以一目了然的看到是哪个 stage 的问题，有针对性的优化。

可以看到优化后的瓶颈主要在最后写 redis 的阶段，要把 60G 的数据，25亿条结果写入 redis，这对 redis 来说是个挑战，这个就只能从写入数据量和 kv 数据库选型两个角度来优化了。

（5）其它优化角度

当然，优化和高性能是个很泛、很有挑战的话题，除了前面提到的代码、参数层面，还有怎样防止或减少数据倾斜等，这都需要针对具体的场景和日志来分析，此处也不展开。

2、spark 初学者的一些误区

对于初学者来说 spark 貌似无所不能而且高性能，甚至在某些博客、技术人眼里 spark 取代 mapreduce、hive、storm 分分钟的事情，是大数据批处理、机器学习、实时处理等领域的银弹。但事实确实如此吗？

从上面这个 case 可以看到，会用 spark、会调 API 和能用好 spark，用的恰到好处是两码事，这要求咱们不仅了解其原理，还要了解业务场景，将合适的技术方案、工具和合适的业务场景结合——这世上本就不存在什么银弹。。。

说道 spark 的性能，想要它快，就得充分利用好系统资源，尤其是内存和CPU：核心思想就是能用内存 cache 就别 spill 落磁盘，CPU 能并行就别串行，数据能 local 就别 shuffle。

===========================================================================================

　　Spark从1.6.0版本开始，内存管理模块就发生了改变，旧版本的内存管理模块是实现了StaticMemoryManager类，现在被称为"legacy"。"Legacy"模式默认被置为不可用，这就意味着当你用Spark1.5.x和Spark1.6.x运行相同的代码会有不同的结果，应当多加注意。考虑的兼容性，可以通过设置spark.memory.useLegacyMode为可用，默认是false.

　　这篇文章介绍自spark1.6.0版本后的新的内存管理模型，它实现的是UnifiedMemoryManager类。

在这张图中你可以看到三个主要内存区域。

1.Reserved Memory.这部分内存是被系统预留的，它的大小也是被硬编码的。在Spark1.6.0版本，它的大小是300MB，这就意味着这部分内存不能计入Spark内存计算，除非重新编译源码或设置spark.testing.reservedMemory，它的大小是不可改变的，因为park.testing.reservedMemory只是一个测试参数所以在生产中不推荐使用。注意，这部分内存只是被称为“Reserved",实际上它不会被spark用来干任何事情 ，但是它限制了你在spark中可分配的内存大小。即使你想将全部JVM堆内存用于spark缓存数据，也不能使用这部分空闲内存（不是真的就浪费了，其实它存储了Spark的一些内部对象）。供参考，如果你不能为executor至少1.5 * Reserved Memory = 450MB的堆内存，任务将会失败并提示”please use larger heap size“的错误信息。

2.User Memory.这部分内存是分配Spark Memory内存之后的部分，而且这部分用来干什么完全取决于你。你可以用来存储RDD transformations过程使用的数据结构。例如，你可以通过mapPartitions transformation 重写Spark aggregation，mapPartitions transformations 保存hash表保证aggregation运行。这部分数据就保存在User Memory。再次强调，这是User Memory它完全由你决定存什么、如何使用，Spark完全不会管你拿这块区域用来做什么，怎么用，也不会考虑你的代码在这块区域是否会导致内存溢出。

3.Spark Memory.这部分内存就是由Spark管理了。这部分内存大小的计算：(“Java Heap” – “Reserved Memory”) *spark.memory.fraction，而且在spark1.6.0版本默认大小为：(“Java Heap” – 300MB) * 0.75。例如：如果堆内存大小有4G，将有2847MB的Spark Memory,Spark Memory=(4*1024MB-300)*0.75=2847MB。这部分内存会被分成两部分：Storage Memory和Execution Memory，而且这两部分的边界由spark.memory.storageFraction参数设定，默认是0.5即50%。新的内存管理模型中的优点是，这个边界不是固定的，在内存压力下这个边界是可以移动的。如一个区域内存不够用时可以从另一区域借用内存。下边来讨论如何移动及使用的：

　　1.Storage Memory.这部分内存即可以用来缓存spark数据也可以用来做unroll序列化数据的临时空间。广播变量以block的形式也存储在这里。你奇怪的是unroll,因为你可能会说，并不需要那么多空间去unroll block使其可用——在没有足够内存去unroll bolock的情况下，如果得到持久化级别的允许，将直接在这部分内存unroll block。至于广播变量，当它的持久化级别为MEMORY_AND_DISK时，就会缓存到此。

　　2.Execution Memory.这部分内存用于存储执行task过程中的一些对象。例如，它可以用来shuflle map端的中间缓存，也可以用来存储hash aggregation过程的hash table.在没有足够内存的时候，这部分内存支持溢室到磁盘，但是这部分内存的blocks不会被其它线程的task挤出去。

　　下边我们来说一下Storage Memory 和Execution Memory之间的边界移动。从Execution Memory的本质来看，你不能将这部分内存空间的数据挤出去，因为这部分内存的数据是用来计算的中间结果，如果计算过程找不到原来存到这的block数据任务就会失败。但是对于Storage Memory内存就不会这样，它只是用来缓存内存中数据，如果将里边的block数据驱逐出去，就会更新block 元数据映射信息使用到时告知该block被移除了，要想再拿到这些数据从HDD中读取即可（或者如果缓存级别没有溢写就重新计算）。

　　所以，我们只能Execution Memory可以向Storage Memory挤用空间，反之不可。那么当什么时候会发生Execution Memory 向Storage Memory挤用空间呢？有两种可能：

• 只要Storage Memory有可用空间，就可以增大Execution Memory 大小，减少Storage Memory 大小。
• Storage Memory的空间大小已经超出了初始设定的大小，并且将这部分空间全部占用，在这种情况下就可以强制将从Storage Memory中移出Blocks,减少它的空间到初始大小。

　　反过来，在只有当Execution Memory空间有空余时，Storage Memory才可以向Execution Memory借用空间，也就是说Execution Memory只要不够用了就可以向Storage Memory挤占空间不管Storage Memory有没有空余，而Storage Memory只能当Execution Memory有空余时才要以借用不能抢占。

　　初始Storage Memory 大小：“Spark Memory” * spark.memory.storageFraction =(“Java Heap” – “Reserved Memory”) *spark.memory.fraction * spark.memory.storageFraction。根据默认值，即(“Java Heap” – 300MB) * 0.75 * 0.5 = (“Java Heap” – 300MB) * 0.375.如果Java Heap=4G，那么就有1423.5MB大小的Storage Memory空间。

　　这就意味着当我们使用Spark cacheu并加载全部数据到executor中时，至少要将Storage Memory大小等于默认初始值大小。因为当Storage Memory区域还没满时，Execution Memory区域已经膨胀大于其初始设定大小时，我们不能强制将Execution Memory抢占的空间数据驱逐，所以最终Storage Memory会变小。

　　希望这篇文章可以帮你更好的理解spark新的内存管理机制，并以此来应用。