本文分为四个部分，基本涵盖了所有Spark优化的点，面试和实际工作中必备。

Spark性能优化：开发调优篇

需要对名为“hello.txt”的HDFS文件进行一次map操作，再进行一次reduce操作//也就是说，需要对一份数据执行两次算子操作。//错误的做法：对于同一份数据执行多次算子操作时，创建多个RDD。//这里执行了两次textFile方法，针对同一个HDFS文件，创建了两个RDD出来，//然后分别对每个RDD都执行了一个算子操作。//这种情况下，Spark需要从HDFS上两次加载hello.txt文件的内容，并创建两个单独的RDD；//第二次加载HDFS文件以及创建RDD的性能开销，很明显是白白浪费掉的。val rdd1 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)")rdd1.map(...)val rdd2 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)")rdd2.reduce(...)//正确的用法：对于一份数据执行多次算子操作时，只使用一个RDD。//这种写法很明显比上一种写法要好多了，因为我们对于同一份数据只创建了一个RDD，//然后对这一个RDD执行了多次算子操作。//但是要注意到这里为止优化还没有结束，由于rdd1被执行了两次算子操作，第二次执行reduce操作的时候，//还会再次从源头处重新计算一次rdd1的数据，因此还是会有重复计算的性能开销。//要彻底解决这个问题，必须结合“原则三：对多次使用的RDD进行持久化”，//才能保证一个RDD被多次使用时只被计算一次。val rdd1 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)")rdd1.map(...)rdd1.reduce(...)

// 错误的做法。// 有一个格式的RDD，即rdd1。// 接着由于业务需要，对rdd1执行了一个map操作，创建了一个rdd2，//而rdd2中的数据仅仅是rdd1中的value值而已，也就是说，rdd2是rdd1的子集。JavaPairRDD rdd1 = ...JavaRDD rdd2 = rdd1.map(...)// 分别对rdd1和rdd2执行了不同的算子操作。rdd1.reduceByKey(...)rdd2.map(...)// 正确的做法。// 上面这个case中，其实rdd1和rdd2的区别无非就是数据格式不同而已，//rdd2的数据完全就是rdd1的子集而已，却创建了两个rdd，并对两个rdd都执行了一次算子操作。// 此时会因为对rdd1执行map算子来创建rdd2，而多执行一次算子操作，进而增加性能开销。// 其实在这种情况下完全可以复用同一个RDD。// 我们可以使用rdd1，既做reduceByKey操作，也做map操作。// 在进行第二个map操作时，只使用每个数据的tuple._2，也就是rdd1中的value值，即可。JavaPairRDD rdd1 = ...rdd1.reduceByKey(...)rdd1.map(tuple._2...)// 第二种方式相较于第一种方式而言，很明显减少了一次rdd2的计算开销。// 但是到这里为止，优化还没有结束，对rdd1我们还是执行了两次算子操作，rdd1实际上还是会被计算两次。// 因此还需要配合“原则三：对多次使用的RDD进行持久化”进行使用，//才能保证一个RDD被多次使用时只被计算一次。

/ 如果要对一个RDD进行持久化，只要对这个RDD调用cache()和persist()即可。// 正确的做法。// cache()方法表示：使用非序列化的方式将RDD中的数据全部尝试持久化到内存中。// 此时再对rdd1执行两次算子操作时，只有在第一次执行map算子时，才会将这个rdd1从源头处计算一次。// 第二次执行reduce算子时，就会直接从内存中提取数据进行计算，不会重复计算一个rdd。val rdd1 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)").cache()rdd1.map(...)rdd1.reduce(...)// persist()方法表示：手动选择持久化级别，并使用指定的方式进行持久化。// 比如说，StorageLevel.MEMORY_AND_DISK_SER表示，内存充足时优先持久化到内存中，//内存不充足时持久化到磁盘文件中。// 而且其中的_SER后缀表示，使用序列化的方式来保存RDD数据，此时RDD中的每个partition//都会序列化成一个大的字节数组，然后再持久化到内存或磁盘中。// 序列化的方式可以减少持久化的数据对内存/磁盘的占用量，进而避免内存被持久化数据占用过多，//从而发生频繁GC。val rdd1 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)").persist(StorageLevel.MEMORY_AND_DISK_SER)rdd1.map(...)rdd1.reduce(...)

// 传统的join操作会导致shuffle操作。// 因为两个RDD中，相同的key都需要通过网络拉取到一个节点上，由一个task进行join操作。val rdd3 = rdd1.join(rdd2)// Broadcast+map的join操作，不会导致shuffle操作。// 使用Broadcast将一个数据量较小的RDD作为广播变量。val rdd2Data = rdd2.collect()val rdd2DataBroadcast = sc.broadcast(rdd2Data)// 在rdd1.map算子中，可以从rdd2DataBroadcast中，获取rdd2的所有数据。// 然后进行遍历，如果发现rdd2中某条数据的key与rdd1的当前数据的key是相同的，//那么就判定可以进行join。// 此时就可以根据自己需要的方式，将rdd1当前数据与rdd2中可以连接的数据，//拼接在一起（String或Tuple）。val rdd3 = rdd1.map(rdd2DataBroadcast...)// 注意，以上操作，建议仅仅在rdd2的数据量比较少（比如几百M，或者一两G）的情况下使用。// 因为每个Executor的内存中，都会驻留一份rdd2的全量数据。

// 以下代码在算子函数中，使用了外部的变量。// 此时没有做任何特殊操作，每个task都会有一份list1的副本。val list1 = ...rdd1.map(list1...)// 以下代码将list1封装成了Broadcast类型的广播变量。// 在算子函数中，使用广播变量时，首先会判断当前task所在Executor内存中，是否有变量副本。// 如果有则直接使用；如果没有则从Driver或者其他Executor节点上远程拉取一份放到本地Executor内存中。// 每个Executor内存中，就只会驻留一份广播变量副本。val list1 = ...val list1Broadcast = sc.broadcast(list1)rdd1.map(list1Broadcast...)

// 创建SparkConf对象。val conf = new SparkConf().setMaster(...).setAppName(...)// 设置序列化器为KryoSerializer。conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")// 注册要序列化的自定义类型。conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))

Spark性能优化：资源调优篇

./bin/spark-submit \--master yarn-cluster \--num-executors 100 \--executor-memory 6G \--executor-cores 4 \--driver-memory 1G \--conf spark.default.parallelism=1000 \--conf spark.storage.memoryFraction=0.5 \--conf spark.shuffle.memoryFraction=0.3 \

Spark性能优化：数据倾斜调优

val conf = new SparkConf()val sc = new SparkContext(conf)val lines = sc.textFile("[hdfs://...](hdfs://...)")val words = lines.flatMap(_.split(" "))val pairs = words.map((_, 1))val wordCounts = pairs.reduceByKey(_ + _)wordCounts.collect().foreach(println(_))

val sampledPairs = pairs.sample(false, 0.1)val sampledWordCounts = sampledPairs.countByKey()sampledWordCounts.foreach(println(_))

// 第一步，给RDD中的每个key都打上一个随机前缀。JavaPairRDD randomPrefixRdd = rdd.mapToPair(new PairFunction, String, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2 call(Tuple2 tuple)throws Exception {  Random random = new Random();  int prefix = random.nextInt(10);  return new Tuple2(prefix + "_" + tuple._1, tuple._2);}});// 第二步，对打上随机前缀的key进行局部聚合。JavaPairRDD localAggrRdd = randomPrefixRdd.reduceByKey(new Function2() {private static final long serialVersionUID = 1L;@Overridepublic Long call(Long v1, Long v2) throws Exception {  return v1 + v2;}});// 第三步，去除RDD中每个key的随机前缀。JavaPairRDD removedRandomPrefixRdd = localAggrRdd.mapToPair(new PairFunction, Long, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2 call(Tuple2 tuple)throws Exception {  long originalKey = Long.valueOf(tuple._1.split("_")[1]);  return new Tuple2(originalKey, tuple._2);}});// 第四步，对去除了随机前缀的RDD进行全局聚合。JavaPairRDD globalAggrRdd = removedRandomPrefixRdd.reduceByKey(new Function2() {private static final long serialVersionUID = 1L;@Overridepublic Long call(Long v1, Long v2) throws Exception {  return v1 + v2;}});

// 首先将数据量比较小的RDD的数据，collect到Driver中来。List> rdd1Data = rdd1.collect()// 然后使用Spark的广播功能，将小RDD的数据转换成广播变量，这样每个Executor就只有一份RDD的数据。// 可以尽可能节省内存空间，并且减少网络传输性能开销。final Broadcast>> rdd1DataBroadcast = sc.broadcast(rdd1Data);// 对另外一个RDD执行map类操作，而不再是join类操作。JavaPairRDD> joinedRdd = rdd2.mapToPair(new PairFunction, String, Tuple2>() {  private static final long serialVersionUID = 1L;  @Override   public Tuple2> call(Tuple2 tuple)  throws Exception {  // 在算子函数中，通过广播变量，获取到本地Executor中的rdd1数据。  List> rdd1Data = rdd1DataBroadcast.value();  // 可以将rdd1的数据转换为一个Map，便于后面进行join操作。  Map rdd1DataMap = new HashMap();  for(Tuple2 data : rdd1Data) {  rdd1DataMap.put(data._1, data._2);}// 获取当前RDD数据的key以及value。String key = tuple._1;String value = tuple._2;// 从rdd1数据Map中，根据key获取到可以join到的数据。Row rdd1Value = rdd1DataMap.get(key);return new Tuple2(key, new Tuple2(value, rdd1Value));}});// 这里得提示一下。// 上面的做法，仅仅适用于rdd1中的key没有重复，全部是唯一的场景。// 如果rdd1中有多个相同的key，那么就得用flatMap类的操作，在进行join的时候不能用map，而是得遍历rdd1所有数据进行join。// rdd2中每条数据都可能会返回多条join后的数据。

// 首先从包含了少数几个导致数据倾斜key的rdd1中，采样10%的样本数据。JavaPairRDD sampledRDD = rdd1.sample(false, 0.1);// 对样本数据RDD统计出每个key的出现次数，并按出现次数降序排序。// 对降序排序后的数据，取出top 1或者top 100的数据，也就是key最多的前n个数据。// 具体取出多少个数据量最多的key，由大家自己决定，我们这里就取1个作为示范。JavaPairRDD mappedSampledRDD = sampledRDD.mapToPair(new PairFunction, Long, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2 call(Tuple2 tuple)throws Exception {return new Tuple2(tuple._1, 1L);}    });JavaPairRDD countedSampledRDD = mappedSampledRDD.reduceByKey(new Function2() {private static final long serialVersionUID = 1L;@Overridepublic Long call(Long v1, Long v2) throws Exception {return v1 + v2;}});JavaPairRDD reversedSampledRDD = countedSampledRDD.mapToPair(new PairFunction, Long, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2 call(Tuple2 tuple)throws Exception {return new Tuple2(tuple._2, tuple._1);}});final Long skewedUserid = reversedSampledRDD.sortByKey(false).take(1).get(0)._2;// 从rdd1中分拆出导致数据倾斜的key，形成独立的RDD。JavaPairRDD skewedRDD = rdd1.filter(new Function, Boolean>() {private static final long serialVersionUID = 1L;@Overridepublic Boolean call(Tuple2 tuple) throws Exception {return tuple._1.equals(skewedUserid);}});// 从rdd1中分拆出不导致数据倾斜的普通key，形成独立的RDD。JavaPairRDD commonRDD = rdd1.filter(new Function, Boolean>() {private static final long serialVersionUID = 1L;@Overridepublic Boolean call(Tuple2 tuple) throws Exception {return !tuple._1.equals(skewedUserid);}});// rdd2，就是那个所有key的分布相对较为均匀的rdd。// 这里将rdd2中，前面获取到的key对应的数据，过滤出来，分拆成单独的rdd，并对rdd中的数据使用flatMap算子都扩容100倍。// 对扩容的每条数据，都打上0～100的前缀。JavaPairRDD skewedRdd2 = rdd2.filter(new Function, Boolean>() {private static final long serialVersionUID = 1L;@Overridepublic Boolean call(Tuple2 tuple) throws Exception {return tuple._1.equals(skewedUserid);}}).flatMapToPair(new PairFlatMapFunction, String, Row>() {private static final long serialVersionUID = 1L;@Overridepublic Iterable> call(Tuple2 tuple) throws Exception {Random random = new Random();List> list = new ArrayList>();for(int i = 0; i < 100; i++) {list.add(new Tuple2(i + "_" + tuple._1, tuple._2));}return list;}});// 将rdd1中分拆出来的导致倾斜的key的独立rdd，每条数据都打上100以内的随机前缀。// 然后将这个rdd1中分拆出来的独立rdd，与上面rdd2中分拆出来的独立rdd，进行join。JavaPairRDD> joinedRDD1 = skewedRDD.mapToPair(new PairFunction, String, String>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2 call(Tuple2 tuple)throws Exception {Random random = new Random();int prefix = random.nextInt(100);return new Tuple2(prefix + "_" + tuple._1, tuple._2);}}).join(skewedUserid2infoRDD).mapToPair(new PairFunction>, Long, Tuple2>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2> call(Tuple2> tuple)throws Exception {long key = Long.valueOf(tuple._1.split("_")[1]);return new Tuple2>(key, tuple._2);}});// 将rdd1中分拆出来的包含普通key的独立rdd，直接与rdd2进行join。JavaPairRDD> joinedRDD2 = commonRDD.join(rdd2);// 将倾斜key join后的结果与普通key join后的结果，uinon起来。// 就是最终的join结果。JavaPairRDD> joinedRDD = joinedRDD1.union(joinedRDD2);

解决方案七：使用随机前缀和扩容RDD进行join

// 首先将其中一个key分布相对较为均匀的RDD膨胀100倍。JavaPairRDD expandedRDD = rdd1.flatMapToPair(new PairFlatMapFunction, String, Row>() {private static final long serialVersionUID = 1L;@Overridepublic Iterable> call(Tuple2 tuple)throws Exception {List> list = new ArrayList>();for(int i = 0; i < 100; i++) {list.add(new Tuple2(0 + "_" + tuple._1, tuple._2));}return list;}});// 其次，将另一个有数据倾斜key的RDD，每条数据都打上100以内的随机前缀。JavaPairRDD mappedRDD = rdd2.mapToPair(new PairFunction, String, String>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2 call(Tuple2 tuple)throws Exception {Random random = new Random();int prefix = random.nextInt(100);return new Tuple2(prefix + "_" + tuple._1, tuple._2);}});// 将两个处理后的RDD进行join即可。JavaPairRDD> joinedRDD = mappedRDD.join(expandedRDD);

解决方案八：多种方案组合使用

Spark性能优化：Shuffle调优篇

一、调优概述

二、ShuffleManager发展概述

2.1、HashShuffleManager运行原理

2.1.1、未经优化的HashShuffleManager

2.1.2、优化后的HashShuffleManager

2.2、SortShuffleManager运行原理

2.2.1、普通运行机制

2.2.2、bypass运行机制

三、shuffle相关参数调优

spark.shuffle.file.buffer

spark.reducer.maxSizeInFlight

spark.shuffle.io.maxRetries

spark.shuffle.io.retryWait

spark.shuffle.memoryFraction

spark.shuffle.manager

spark.shuffle.sort.bypassMergeThreshold

spark.shuffle.consolidateFiles

写在最后的话

--end--

扫描下方二维码
添加好友，备注【交流】
可私聊交流，也可进资源丰富学习群