Spark几种调优方式
adayan0 2019-10-27
1、避免创建重复的RDD和不必要的内存空间浪费
错误代码：
val rdd1 = sc.textFile("D:\\abc\\wordcount\\input\\hello.txt")
rdd1.map(...)
val rdd2 = sc.textFile("D:\\abc\\wordcount\\input\\hello.txt")
rdd2.reduce(...)错误解析：
这种情况下，Spark需要从文件中加载两次hello.txt文件的内容，并创建两个单独的RDD；第二次加载HDFS文件以及创建RDD的性能开销，很明显是白白浪费掉的
正确代码：
val rdd1 = sc.textFile("D:\\abc\\wordcount\\input\\hello.txt")
rdd1.map(...)
rdd1.reduce(...)
这种写法很明显比上一种写法要好多了，因为我们对于同一份数据只创建了一个RDD，然后对这一个RDD执行了多次算子操作2、尽最大可能复用同一个RDD
错误代码：
val rdd1 = sc.textFile("D:\\abc\\wordcount\\input\\hello.txt")
Val rdd2 = rdd1.map(...)
//分别对rdd1和rdd2进行了不同的算子操作
rdd1.reduceByKey(...)
rdd2.map(...)错误解析：
上面这个案例中，其实rdd1和rdd2的区别无非就是数据格式不同而已，rdd2的数据完全就是rdd1的子集而已，却创建了两个rdd，并对两个rdd都执行了一次算子操作。 此时会因为对rdd1执行map算子来创建rdd2，而多执行一次算子操作，进而增加性能开销。
正确代码：
val rdd1 = sc.textFile("D:\\abc\\wordcount\\input\\hello.txt")
rdd1.reduceByKey(...)
rdd1.map(tuple._2...)
在进行第二个map操作时，只使用每个数据的tuple._2，也就是rdd1中的value值，即可。第二种方式相较于第一种方式而言，很明显减少了一次rdd2的计算开销3、对多次使用的RDD进行持久化操作
错误代码：
val rdd1 = sc.textFile("D:\\abc\\wordcount\\input\\hello.txt")
rdd1.map(...)
.reduce(...)
.reduceByKey(...)错误解析：
Spark中对于一个RDD执行多次算子的默认原理是这样的：每次你对一个RDD执行一个算子操作时，都会重新从源头处计算一遍，计算出那个RDD来，然后再对这个RDD执行你的算子操作。这种方式的性能是很差的
正确代码：
val rdd1 = sc.textFile("D:\\abc\\wordcount\\input\\hello.txt").cache()
rdd1.map(...)
rdd1.reduce(...)
或者
Val rdd1=sc.textFile("D:\\abc\\wordcount\\input\\hello.txt").persist
(StorageLevel.MEMORY_AND_DISK_SER)
rdd1.map(...)
rdd1.reduce(...)
如果要对一个RDD进行持久化，只要对这个RDD调用cache()和persist()即可。
cache()方法表示：使用非序列化的方式将RDD中的数据全部尝试持久化到内存中。此时再对rdd1执行两次算子操作时，只有在第一次执行map算子时，才会将这个rdd1从源头处计算一次。第二次执行reduce算子时，就会直接从内存中提取数据进行计算，不会重复计算一个rdd。
persist()方法表示：手动选择持久化级别，并使用指定的方式进行持久化。 比如说，StorageLevel.MEMORY_AND_DISK_SER表示，内存充足时优先持久化到内存中，内存不充足时持久化到磁盘文件中。而且其中的_SER后缀表示，使用序列化的方式来保存RDD数据，此时RDD中的每个partition都会序列化成一个大的字节数组，然后再持久化到内存或磁盘中。序列化的方式可以减少持久化的数据对内存/磁盘的占用量，进而避免内存被持久化数据占用过多，从而发生频繁GC。（总共有12中方式源码）4、尽量避免使用shuffle算子
错误代码：
val rdd3 = rdd1.join(rdd2)
错误分析：
Spark作业运行过程中，最消耗性能的地方就是shuffle过程。所以能避免则尽可能避免使用reduceByKey、join、distinct、repartition等会进行shuffle的算子，尽量使用map类的非shuffle算子。这样的话，没有shuffle操作或者仅有较少shuffle操作的Spark作业，可以大大减少性能开销
正确代码：
val rdd2 = rdd2.collect() val rdd2Broadcast = sc.broadcast(rdd2) Broadcast+map的join操作，不会导致shuffle操作。使用Broadcast将一个数据量较小的RDD作为广播变量。以上操作，建议仅仅在rdd2的数据量比较少（比如几百M，或者一两G）的情况下使用 因为每个Executor的内存中，都会驻留一份rdd2的全量数据
5、使用性能较高的算子
groupByKey代码：
val Data1 = lines.flatMap(_.split(" "))
.map((_,1))
.groupByKey().mapValues(_.sum)reduceByKey代码：
val Data2 = lines.flatMap(_.split(" "))
.map((_, 1))
.reduceByKey(_ + _)使用reduceByKey替代groupByKey，因为reduceByKey会先进行一次局部聚合
再比如使用mapPartitions替代普通map
mapPartitions类的算子，一次函数调用会处理一个partition所有的数据，而不是一次函数调用处理一条，性能相对来说会高一些，使用mapPartitions会出现OOM（内存溢出）的问题，所以使用这类操作时要慎重
6、使用Kryo优化序列化性能
代码：
object YouHua {
def main(args: Array[String]): Unit = {
val conf = new SparkConf()
.setAppName("haha")
.setMaster("local[*]")
.set("spark.testing.memory", "2147480000")
.registerKryoClasses(Array(classOf[Student]))
val sc = new SparkContext(conf)
}
}
// spark默认Java的序列化
class Student{}Kryo序列化机制比Java序列化机制，性能高10倍左右，但是spark默认的是java的序列化，因为这个不需要自己手动注册。而Kryo序列化机制我们只要设置序列化类，再注册要序列化的自定义类型即可，注意有一个类需要注册一次所以说比较费劲
7、解决数据倾斜
如何定位导致数据倾斜的代码：
数据倾斜只会发生在shuffle过程中。这里给大家罗列一些常用的并且可能会触发shuffle操作的算子：distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition等。出现数据倾斜时，可能就是你的代码中使用了这些算子中的某一个所导致的
解决办法
（1）加盐
val lines = sc.makeRDD(List("ni ni aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa aa"))
lines.flatMap(_.split(" "))
//首先我们拼一个时间也可以拼随机数
.map(x => (System.currentTimeMillis().toString + "-" + x,1))
//按照key聚合一下，这样就会打散，均匀分发各个分区
.reduceByKey(_+_)
//去掉我们拼的前缀
.map(x => (x._1.substring(14),x._2))
//最后在聚合得到我们想要的结果
.reduceByKey(_+_)
.mapPartitionsWithIndex(func1)（2）自定义分区
val func1 = (index:Int,iter:Iterator[(String,Int)]) => {
iter.map(index +":" + _)
}
lines.flatMap(_.split(" "))
.map(x => (x,1))
//按照自定义的分区随机分发数据
.reduceByKey(new MyPartitioner(),_+_)
//在按照HashPartitioner聚合数据
.reduceByKey(new HashPartitioner(4),_+_)
.mapPartitionsWithIndex(func1)
.foreach(println)
class MyPartitioner extends Partitioner{
override def numPartitions: Int = 4
override def getPartition(key: Any): Int = {
//这里我们给返回随机数
Random.nextInt(numPartitions)
}
}（1）采样倾斜key并分拆join操作
如果出现数据倾斜，是因为其中某一个RDD/Hive表中的少数几个key的数据量过大，而另一个RDD/Hive表中的所有key都分布比较均匀，那么采用这个解决方案是比较合适的
实现思路
举例：一号RDD中多个Key都有数据倾斜、二号RDD相对均衡
- 先从一号RDD中利用sample进行抽样 Val sampledRDD = rdd1.sample(false, 0.1)
- 对样本数据RDD统计出每个key的出现次数，并按出现次数降序排序。对降序排序后的数据，取出top 10或者top 100的数据，也就是key最多的前n个数据。具体取出多少个数据量最多的key，由大家自己决定
- 从一号RDD中分离出来导致数据倾斜的key，形成新的RDD
- 从一号RDD中拆出不导致数据倾斜的普通key，形成独立的RDD
- 先给一号RDD中导致数据倾斜的key加盐（随机数（或者自己想要的前缀））再从二号RDD中，前面获取到的key对应的数据，过滤出来，分拆成单独的rdd，并对rdd中的数据也都加盐，加上前面一号RDD的所有盐。然后将这个一号RDD中分拆出来的独立rdd，与上面二号RDD分拆出来的独立rdd，进行join
举例说明：
6.将一号RDD中分拆出来的包含不倾斜key的独立rdd，直接与rdd2进行join
7.将倾斜key join后的结果与普通key join后的结果，uinon起来就是最终的结果
