正文
Spark性能调优之代码方面的优化
提示：扫一扫查出行【扫一扫了解最新限行尾号】
复制提示
Spark性能调优之代码方面的优化1.避免创建重复的RDD 对性能没有问题，但会造成代码混乱 2.尽可能复用同一个RDD，减少产生RDD的个数 3.对多次使用的RDD进行持久化（cache,persist,checkpoint）如何选择一种最合适的持久化策略？ 默认MEMORY_ONLY, 性能很高， 而且不需要复制一份数据的副本，远程传送到其他节点上（BlockManager中的BlockTransferService），但是这里必须要注意的是，在实际的生产环境中，恐怕能够直接用这种策略的场景还是有限的，如果RDD中数据比较多时（比如几十亿），直接用这种持久化级别，会导致JVM的OOM内存溢出异常。 如果使用MEMORY_ONLY级别时发生了内存溢出，建议尝试使用MEMORY_ONLY_SER级别，该级别会将RDD数据序列化后再保存在内存中，此时每个partition仅仅是一个字节数组而已，大大减少了对象数量，并降低了内存占用。这种级别比MEMORY_ONLY多出来的性能开销，主要就是序列化与反序列化的开销。 如果纯内存的级别都无法使用，那么建议使用MEMORY_AND_DISK_SER策略，而不是
MEMORY_AND_DISK策略。因为既然到了这一步，就说明RDD的数据量很大，内存无法完全放下。序列化后的数据比较少，可以节省内存和磁盘的空间开销。同时该策略会优先尽量尝试将数据缓存在内存中，内存缓存不下才会写入磁盘。 通常不建议使用DISK_ONLY和后缀为_2的级别：因为完全基于磁盘文件进行数据的读写，会导致性能急剧降低，有时还不如重新计算一次所有RDD。后缀为_2的级别，必须将所有数据都复制一份副本，并发送到其他节点上，数据复制以及网络传输会导致较大的性能开销，除非是要求作业的高可用性，否则不建议使用。 checkpoint 可以使数据安全，切断依赖关系（如果某一个rdd丢失了，重新计算的链太长？） 4.尽量避免使用shuffle类的算子 广播变量模拟join(一个RDD比较小，另一个RDD比较大) 5.使用map-side预聚合shuffle操作 reduceByKey aggregateByKey
6.使用高性能的算子 有哪些高性能的算子？ reduceByKey/aggregateByKey 替代 groupByKey mapPartitions 替代普通map Transformation算子 foreachPartitions 替代 foreach Action算子 repartitionAndSortWithinPartitions 替代repartition与sort类操作 rdd.partitionBy() //其实自定义一个分区器 repartition coalesce（numPartitions，true） 增多分区使用这个
coalesce（numPartitions，false） 减少分区 没有shuffle只是合并partition 7.广播变量 开发过程中，会遇到需要在算子函数中使用外部变量的场景（尤其是大变量，比如100M以上的大集合），那么此时就应该使用Spark的广播（Broadcast）功能来提升性能,如果使用的外部变量比较大，建议使用Spark的广播功能，对该变量进行广播。广播后的变量，会保证每个Executor的内存中，只驻留一份变量副本，而Executor中的task执行时共享该Executor中的那份变量副本。这样的话，可以大大减少变量副本的数量，从而减少网络传输的性能开销，并减少对Executor内存的占用开销，降低GC的频率 Executor 2G, 2G*0.48 8.使用kryo序列化性能？ • Spark支持使用Kryo序列化机制。Kryo序列化机制，比默认的Java序列化机制，速度要快，序列化后的数据要更小，大概是Java序列化机制的1/10。所以Kryo序列化优化以后，可以让网络传输的数据变少；在集群中耗费的内存资源大大减少。 • 对于这三种出现序列化的地方，我们都可以通过使用Kryo序列化类库，来优化序列化和反序列化的性能。Spark默认使用的是Java的序列化机制，也就是ObjectOutputStream/ObjectInputStream API来进行序列化和反序列化。但是Spark同时支持使用Kryo序列化库，Kryo序列化类库的性能比Java序列化类库的性能要高很多。官方介绍，Kryo序列化机制比Java序列化机制，性能高10倍左右。Spark之所以默认没有使用Kryo作为序列化类库，是因为Kryo要求最好要注册所有需要进行序列化的自定义类型，因此对于开发者来说，这种方式比较麻烦 自定义的类有哪些，都要注册
9.优化数据结构 尽量使用字符串代替对象，使用原始类型（Int,long）替代字符串，使用数组替代集合类型，这样尽可能地减少内存占用，从而降低GC频率，提升性能。 10.使用高性能的fastutil • fastutil是扩展了Java标准集合框架（Map、List、Set；HashMap、ArrayList、HashSet）的类库，提供了特殊类型的map、set、list和queue； • fastutil能够提供更小的内存占用，更快的存取速度；我们使用fastutil提供的集合类，来替代自己平时使用的JDK的原生的Map、List、Set，好处在于，fastutil集合类，可以减小内存的占用，并且在进行集合的遍历、根据索引（或者key）获取元素的值和设置元素的值的时候，提供更快的存取速度； • fastutil最新版本要求Java 7以及以上版本； • fastutil的每一种集合类型，都实现了对应的Java中的标准接口（比如fastutil的map，实现了Java的Map接口），因此可以直接放入已有系统的任何代码中。RandomExtractCars 提供了一些集合，性能更高
必须是java7以上的版本
1.避免创建重复的RDD
对性能没有问题，但会造成代码混乱
2.尽可能复用同一个RDD，减少产生RDD的个数
3.对多次使用的RDD进行持久化（cache,persist,checkpoint）
如何选择一种最合适的持久化策略？
默认MEMORY_ONLY, 性能很高， 而且不需要复制一份数据的副本，远程传送到其他节点上（BlockManager中的BlockTransferService），但是这里必须要注意的是，在实际的生产环境中，恐怕能够直接用这种
策略的场景还是有限的，如果RDD中数据比较多时（比如几十亿），直接用这种持久化
级别，会导致JVM的OOM内存溢出异常。
如果使用MEMORY_ONLY级别时发生了内存溢出，建议尝试使用MEMORY_ONLY_SER级别，该级别会将RDD数据序列化后再保存在内存中，此时每个partition仅仅是一个字节数组而已，大大减少了对象数量，并降低了内存占用。这种级别比MEMORY_ONLY多出来的性能开销，主要就是序列化与反序列化的开销。
如果纯内存的级别都无法使用，那么建议使用MEMORY_AND_DISK_SER策略，而不是
MEMORY_AND_DISK策略。因为既然到了这一步，就说明RDD的数据量很大，内存无法完全放下。序列化后的数据比较少，可以节省内存和磁盘的空间开销。同时该策略会优先尽量尝试将数据缓存在内存中，内存缓存不下才会写入磁盘。
通常不建议使用DISK_ONLY和后缀为_2的级别：因为完全基于磁盘文件进行数据的读写，会导致性能急剧降低，有时还不如重新计算一次所有RDD。后缀为_2的级别，必须将所有数据都复制一份副本，并发送到其他节点上，数据复制以及网络传输会导致较大的性能开销，除非是要求作业的高可用性，否则不建议使用。
checkpoint 可以使数据安全，切断依赖关系（如果某一个rdd丢失了，重新计算的链太长？）
4.尽量避免使用shuffle类的算子
广播变量模拟join(一个RDD比较小，另一个RDD比较大)
5.使用map-side预聚合shuffle操作
reduceByKey aggregateByKey
6.使用高性能的算子
有哪些高性能的算子？
reduceByKey/aggregateByKey 替代 groupByKey
mapPartitions 替代普通map Transformation算子
foreachPartitions 替代 foreach Action算子
repartitionAndSortWithinPartitions 替代repartition与sort类操作
rdd.partitionBy() //其实自定义一个分区器
repartition coalesce（numPartitions，true） 增多分区使用这个
coalesce（numPartitions，false） 减少分区 没有shuffle只是合并partition
7.广播变量
开发过程中，会遇到需要在算子函数中使用外部变量的场景（尤其是大变量，比如100M以上的大集合），那么此时就应该使用Spark的广播（Broadcast）功能来提升性能,如果使用的外部变量比较大，建议使用Spark的广播功能，对该变量进行广播。广播后的变量，会保证每个Executor的内存中，只驻留一份变量副本，而Executor中的
task执行时共享该Executor中的那份变量副本。这样的话，可以大大减少变量副本的数量，从而减少网络传输的性能开销，并减少对Executor内存的占用开销，降低GC的频率
Executor 2G, 2G*0.48
8.使用kryo序列化性能？
• Spark支持使用Kryo序列化机制。Kryo序列化机制，比默认的Java序列化机制，速度要快
，序列化后的数据要更小，大概是Java序列化机制的1/10。所以Kryo序列化优化以后，可
以让网络传输的数据变少；在集群中耗费的内存资源大大减少。
• 对于这三种出现序列化的地方，我们都可以通过使用Kryo序列化类库，来优化序列化和反序列化的性能。Spark默认使用的是Java的序列化机制，也就是ObjectOutputStream/ObjectInputStream API来进行序列化和反序列化。但是Spark同时支持使用Kryo序列化库，Kryo序列化类库的性能比Java序列化类库的性能要高很多。官方介绍，Kryo序列化机制比Java序列化机制，性能高10倍左右。Spark之所以默认没有
使用Kryo作为序列化类库，是因为Kryo要求最好要注册所有需要进行序列化的自定义类型，因此对于开发者来说，这种方式比较麻烦
自定义的类有哪些，都要注册
9.优化数据结构
尽量使用字符串代替对象，使用原始类型（Int,long）替代字符串，使用数组替代集合类型，这样尽可能地减少内存占用，从而降低GC频率，提升性能。
10.使用高性能的fastutil
• fastutil是扩展了Java标准集合框架（Map、List、Set；HashMap、ArrayList、HashSet）的类库，提供了特殊类型的map、set、list和queue；
• fastutil能够提供更小的内存占用，更快的存取速度；我们使用fastutil提供的集合类，来替代自己平时使用的JDK的原生的Map、List、Set，好处在于，fastutil集合类，可以减小内存的占用，并且在进行集合的遍历、根据索引（或者key）获取元素的值和设置元素的值的时候，提供更快的存取速度；
• fastutil最新版本要求Java 7以及以上版本；
• fastutil的每一种集合类型，都实现了对应的Java中的标准接口（比如fastutil的map，实现了Java的Map接口），因此可以直接放入已有系统的任何代码中。
RandomExtractCars
提供了一些集合，性能更高
必须是java7以上的版本
