spark性能优化框架
--基础设置
1、资源参数设置：
--num-executors 配置Executor的数量 默认为2
--driver-memory 配置Driver内存 默认为1G
--executor-memory 配置每个Executor的内存 默认为1G
--executor-cores 配置每个Executor 的CPU core数量 默认是1核
task的并行度=executors 个数 × 每个executor包含的cpu core数
executor 的内存提升可以为cache、huffle操作、task执行操作提供更多内存
2、并行度设置：
task的数量推荐是总CPU core数的2-3倍
3、使用缓存和checkpoint。
checkpoint可以视为cache的保障机制
4、使用广播变量
广播变量将变量发送到每个executors中，就不需要每个task保存一个变量的副本，减少了内存消耗。比如变量20m，某个任务配置20个executor，共500个task。如果不启动广播变量，则共消耗500*20m，共10g。开启广播变量则只需消耗400m。
val 广播变量名=sc.broadcast(会被各个task用到的变量，即需要广播的变量)
广播变量名.value //获取广播变量值
5、序列化问题？
默认使用java的序列化机制，因为Kyro序列化方式不支持所有对象的序列化。Kryo在2.x版本在简单类型、字符串类型已经默认使用Kryo序列化方式了。
val conf=new SparkConf().setMaster().setAppName()
//设置使用Kryo序列化库
conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
//在Kryo序列化库中注册自定义的类集合
conf.set("spark.kyro.registrator","bigdata.com.MyKryoRegistrator");
6、本地化等待时长设置
首先spark在分配task执行的算法，是希望将task在本地节点执行，task和数据在同一个executor中，性能最好。但节点会在执行任务时没有足够的资源去再分配task执行。这时候会涉及到本地化的等待时长问题。
本地化类型和解析
根据程序的运行情况选择合适的进程本地化等待时长，比如使用client模式对程序进行测试，在日志中可以看到spark数据本地化的级别。当大部分为process_local时，则无需调节等待时长。
val conf =new SparkConf().set("spark.locality.wait","6")
--算子优化
1、RDD的复用
能复用的RDD尽量复用
2、能尽早过滤数据就尽早过滤数据。
filter
3、foreachpartition
rdd.foreache（）表示对每个元素进行操作
rdd.foreachepartition() 表示对每个分区进行操作
=>如果涉及数据库的相关操作，那么foreachpartition()每个分区的数据只需要创建一次连接
4、filter+coalesce减少分区
当过滤数据后，可能会出现某些分区的数据较多，某些分区的数据较少。这样如果仍然按照之前的partition相等的task数量去处理当前数据，可能会出现资源浪费和数据倾斜。这时候考虑将一些分区的数据进行合并，减少分区数。
5、reducebykey 本地预聚合
reducebykey对比普通的shuffle，在map端会先对本地的数据进行conbine操作。减少了网络磁盘IO的资源消耗。
--shuffle优化
1、map端和reduce端的缓冲区大小
map端缓冲的默认配置是32k，如果每个task处理640k的数据，发生溢写的次数是20次。可以通过调节map端缓冲大小，减少溢写次数。
val conf = new Sparkconf().set("spark.shuffle.file.buffler","64")
reduce端的buffer缓冲区大小决定reduce task每次能够以拉取的数据量，默认是48m。在内存资源较为充足的情况下，可以增加该缓冲区大小，减少网络IO的资源消耗。
val conf=new SparkConf().set("spark.reducer.maxSizeInFlight","96m")
2、reduce端的重试次数和等待间隔
reduce端在拉取数据时，由于网络、jvm进行full gc等原因造成拉取失败，默认会进行3次的重试。在实际工作中，对于大数据量的情况下，调节该参数可以大幅度提升稳定性。
val conf=new SparkConf().set("spark.shuffle.io.maxRetries","6")
reduce拉取数据失败，会默认等待5s，再进行重试。可以通过调节该参数提升shuffle的成功率
val conf=new SparkConf().set("spark.shuflle.io.retryWait","60s")
3、bypass机制
sortshuffle分为普通机制和bypass机制。在普通机制的情况下，map数据会先写入到内存区域中，默认为5m，此时会对数据根据key的hash值进行排序。在bypass的机制下，则不会进行排序操作，可以减少内存开销。
bypass机制的开启条件：
Ⅰ参数值大于reduce task的数量
set("spark.shuffle.sort.bypassMergeThreshold","400")
Ⅱ不是聚合类的shuffle算子（如reducebykey）
4、延申：shuffle算子有哪些？哪些会进行聚合，哪些不会？
org.apache.spark.rdd.OrderedRDDFunctions#sortByKey
org.apache.spark.rdd.OrderedRDDFunctions#repartitionAndSortWithinPartitions
org.apache.spark.rdd.PairRDDFunctions#partitionBy
org.apache.spark.rdd.RDD#coalesce
org.apache.spark.rdd.PairRDDFunctions#combineByKey 聚合
org.apache.spark.rdd.PairRDDFunctions#aggregateByKey 聚合
org.apache.spark.rdd.PairRDDFunctions#foldByKey
org.apache.spark.rdd.PairRDDFunctions#reduceByKey 聚合
org.apache.spark.rdd.PairRDDFunctions#countApproxDistinctByKey
org.apache.spark.rdd.PairRDDFunctions#groupByKey 聚合
org.apache.spark.rdd.PairRDDFunctions#cogroup
org.apache.spark.rdd.PairRDDFunctions#subtractByKey
网友评论