Spark调优总结

「爱情、让人受尽委屈。」 2021-09-12 02:10 451阅读 0赞

下面主要是笔者根据使用Spark过程中的一些调优做一些汇总。

1、调整分配excutor memory

-num-executors 100 --executor-cores 4 --driver-memory 6g --executor-memory 6g

首先先将参数调大一点，然后程序跑通过后。再通过Spark任务的监控页面看资源使用的情况。再来调整各项目参数。一般情况下，调整这几个参数都会很有用。

2、调整JVM垃圾回收算法

可以通过 --conf 来批明使用的垃圾回收算法，对于多核处理器来说，一般是使用CMS+PartNew。自己也可以设置一些其它参数，如年轻代、年老代的各大小，垃圾回收日志的校园等。如下：

--conf "spark.driver.extraJavaOptions=-XX:PermSize=512m -XX:MaxPermSize=512m  -XX:+CMSClassUnloadingEnabled -XX:MaxTenuringThreshold=31 -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+UseCMSCompactAtFullCollection -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=10 -XX:+UseCompressedOops -XX:+PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime -XX:+PrintHeapAtGC -XX:+PrintGCApplicationConcurrentTime -verbose:gc -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/app/hadoop/bdp/bdp-core/logs/ -Xloggc:/app/hadoop/bdp/bdp-core/logs/gc1.log" \
    --conf "spark.executor.extraJavaOptions=-XX:NewSize=512m -XX:MaxNewSize=512m -XX:PermSize=512m -XX:MaxPermSize=512m  -XX:+CMSClassUnloadingEnabled -XX:MaxTenuringThreshold=31 -XX:+UseConcMarkSweepGC -XX:+CMSParallelRemarkEnabled -XX:+UseCMSCompactAtFullCollection -XX:+UseCMSInitiatingOccupancyOnly -XX:CMSInitiatingOccupancyFraction=10 -XX:+UseCompressedOops -XX:+PrintGC -XX:+PrintGCDetails -XX:+PrintGCTimeStamps -XX:+PrintGCDateStamps -XX:+PrintGCApplicationStoppedTime -XX:+PrintHeapAtGC -XX:+PrintGCApplicationConcurrentTime -verbose:gc -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/app/hadoop/bdp/bdp-core/logs/ -Xloggc:/app/hadoop/bdp/bdp-core/logs/gc.log" \

一般情况下，内存溢出了有可能是各个年代的分配不对，可以通过查看垃圾回收的日志，来看具估是在哪一个环节频繁发生GC。这里注意，可以设置driver，也可以设置execuotr的。

3、设置存储和执行内存的分配比例

设置存储内存比例spark.storage.memoryFraction，默认0.6

设置执行内存比例spark.executor.memoryFraction，默认0.2

如果程序中很少用或用得少persist(）的场景，可以通过提高执行内存比例来加快运行速度。但是这里要注意的是，这两个参数只在1.6版本之前有较好的效果，因为1.6之后存储内存和执行内存不是严格限制死的，可以相互借用。所以1.6版本之后默认这两个参数设置是不起作用，除非你指明使用1.6版本之前的静态内存管理方案。

4、缓存多次用到的中间RDD,复用相同的RDD

适当使用perist()方法，有时会带来点效果。但是如果缓存数据量大的话，建议开户硬盘缓存，默认的是内存缓存。复用相同的RDD意思是不同的RDD不要依赖的上一个RDD的计算逻辑是一样的，这是上一个RDD又是不同的实例RDD。

5、广播大变量

大常量可以这么做，注意要在Driver端做广播

6、合理设置分区数量

比如通过Spakr存放到Hive里,可以使用parallelize来指明分区数量，避免小文件过多。然后Spark再去读取时又生成多个task.一般情况下，一个分区会生成一个task.

7、多表关联，小表放前面

这是写SQL的常识，请谨记！

8、使用forecahPartiton、mapPartition代替foreach、map

因为forecahPartiton、mapPartition的入参是一个带有iterator的函数，意味着所有的分区RDD可以并行的执行同一个function。这在数量量大时会带来很显著的效率提升。

**Foreach与foreachPartition都是在每个partition中对iterator进行操作,不同的是,foreach是直接在每个partition中直接对iterator执行foreach操作,而传入的function只是在foreach内部使用,而foreachPartition是在每个partition中把iterator给传入的function,让function自己对iterator进行处理.**

9、慎用collection

collection会将数据拉取集中到Driver机器上，如果数据量一大。很容易Driver机器就OOM。

10、coalesce、repartition使用

coalesce、repartition这个函数能用来重新指定分区个数,一般在多次中间操作后使用。可以通过反复调节这个参数，有时会带来意想不到的效果

11、尽量将filter提前

原理就是为了RDD在进行下步是能尽量少。

12、reduceByKey或者aggregateByKey算子来替代掉groupByKey算子

在可能的情况下，建议使用reduceByKey或者aggregateByKey算子来替代掉groupByKey算子。因为reduceByKey和aggregateByKey算子都会使用用户自定义的函数对每个节点本地的相同key进行预聚合。而groupByKey算子是不会进行预聚合的，全量的数据会在集群的各个节点之间分发和传输，性能相对来说比较差。

13、操作Hive时，查询条件一定要带上分区字段，限制非空，尽量可能多的条件

查询带分区字段，避免全表扫描。

最后，再分享一个方案，可以通过查看Spark的任务执行页面，根据执行任务,对各个任务做不同的调整

先来看看一个 stage 里所有 task 运行的一些性能指标，其中的一些说明：  
Scheduler Delay: spark 分配 task 所花费的时间  
Executor Computing Time: executor 执行 task 所花费的时间  
Getting Result Time: 获取 task 执行结果所花费的时间  
Result Serialization Time: task 执行结果序列化时间  
Task Deserialization Time: task 反序列化时间  
Shuffle Write Time: shuffle 写数据时间  
Shuffle Read Time: shuffle 读数据所花费时间