本文分為四個(gè)部分，基本涵蓋了所有Spark優(yōu)化的點(diǎn)，面試和實(shí)際工作中必備。

Spark性能優(yōu)化：開發(fā)調(diào)優(yōu)篇

需要對(duì)名為“hello.txt”的HDFS文件進(jìn)行一次map操作，再進(jìn)行一次reduce操作//也就是說(shuō)，需要對(duì)一份數(shù)據(jù)執(zhí)行兩次算子操作。//錯(cuò)誤的做法：對(duì)于同一份數(shù)據(jù)執(zhí)行多次算子操作時(shí)，創(chuàng)建多個(gè)RDD。//這里執(zhí)行了兩次textFile方法，針對(duì)同一個(gè)HDFS文件，創(chuàng)建了兩個(gè)RDD出來(lái)，//然后分別對(duì)每個(gè)RDD都執(zhí)行了一個(gè)算子操作。//這種情況下，Spark需要從HDFS上兩次加載hello.txt文件的內(nèi)容，并創(chuàng)建兩個(gè)單獨(dú)的RDD；//第二次加載HDFS文件以及創(chuàng)建RDD的性能開銷，很明顯是白白浪費(fèi)掉的。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(...)//正確的用法：對(duì)于一份數(shù)據(jù)執(zhí)行多次算子操作時(shí)，只使用一個(gè)RDD。//這種寫法很明顯比上一種寫法要好多了，因?yàn)槲覀儗?duì)于同一份數(shù)據(jù)只創(chuàng)建了一個(gè)RDD，//然后對(duì)這一個(gè)RDD執(zhí)行了多次算子操作。//但是要注意到這里為止優(yōu)化還沒(méi)有結(jié)束，由于rdd1被執(zhí)行了兩次算子操作，第二次執(zhí)行reduce操作的時(shí)候，//還會(huì)再次從源頭處重新計(jì)算一次rdd1的數(shù)據(jù)，因此還是會(huì)有重復(fù)計(jì)算的性能開銷。//要徹底解決這個(gè)問(wèn)題，必須結(jié)合“原則三：對(duì)多次使用的RDD進(jìn)行持久化”，//才能保證一個(gè)RDD被多次使用時(shí)只被計(jì)算一次。val rdd1 = sc.textFile("[hdfs://192.168.0.1:9000/hello.txt](hdfs://192.168.0.1:9000/hello.txt)")rdd1.map(...)rdd1.reduce(...)

// 錯(cuò)誤的做法。// 有一個(gè)格式的RDD，即rdd1。// 接著由于業(yè)務(wù)需要，對(duì)rdd1執(zhí)行了一個(gè)map操作，創(chuàng)建了一個(gè)rdd2，//而rdd2中的數(shù)據(jù)僅僅是rdd1中的value值而已，也就是說(shuō)，rdd2是rdd1的子集。JavaPairRDD rdd1 = ...JavaRDD rdd2 = rdd1.map(...)// 分別對(duì)rdd1和rdd2執(zhí)行了不同的算子操作。rdd1.reduceByKey(...)rdd2.map(...)// 正確的做法。// 上面這個(gè)case中，其實(shí)rdd1和rdd2的區(qū)別無(wú)非就是數(shù)據(jù)格式不同而已，//rdd2的數(shù)據(jù)完全就是rdd1的子集而已，卻創(chuàng)建了兩個(gè)rdd，并對(duì)兩個(gè)rdd都執(zhí)行了一次算子操作。// 此時(shí)會(huì)因?yàn)閷?duì)rdd1執(zhí)行map算子來(lái)創(chuàng)建rdd2，而多執(zhí)行一次算子操作，進(jìn)而增加性能開銷。// 其實(shí)在這種情況下完全可以復(fù)用同一個(gè)RDD。// 我們可以使用rdd1，既做reduceByKey操作，也做map操作。// 在進(jìn)行第二個(gè)map操作時(shí)，只使用每個(gè)數(shù)據(jù)的tuple._2，也就是rdd1中的value值，即可。JavaPairRDD rdd1 = ...rdd1.reduceByKey(...)rdd1.map(tuple._2...)// 第二種方式相較于第一種方式而言，很明顯減少了一次rdd2的計(jì)算開銷。// 但是到這里為止，優(yōu)化還沒(méi)有結(jié)束，對(duì)rdd1我們還是執(zhí)行了兩次算子操作，rdd1實(shí)際上還是會(huì)被計(jì)算兩次。// 因此還需要配合“原則三：對(duì)多次使用的RDD進(jìn)行持久化”進(jìn)行使用，//才能保證一個(gè)RDD被多次使用時(shí)只被計(jì)算一次。

/ 如果要對(duì)一個(gè)RDD進(jìn)行持久化，只要對(duì)這個(gè)RDD調(diào)用cache()和persist()即可。// 正確的做法。// cache()方法表示：使用非序列化的方式將RDD中的數(shù)據(jù)全部嘗試持久化到內(nèi)存中。// 此時(shí)再對(duì)rdd1執(zhí)行兩次算子操作時(shí)，只有在第一次執(zhí)行map算子時(shí)，才會(huì)將這個(gè)rdd1從源頭處計(jì)算一次。// 第二次執(zhí)行reduce算子時(shí)，就會(huì)直接從內(nèi)存中提取數(shù)據(jù)進(jìn)行計(jì)算，不會(huì)重復(fù)計(jì)算一個(gè)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()方法表示：手動(dòng)選擇持久化級(jí)別，并使用指定的方式進(jìn)行持久化。// 比如說(shuō)，StorageLevel.MEMORY_AND_DISK_SER表示，內(nèi)存充足時(shí)優(yōu)先持久化到內(nèi)存中，//內(nèi)存不充足時(shí)持久化到磁盤文件中。// 而且其中的_SER后綴表示，使用序列化的方式來(lái)保存RDD數(shù)據(jù)，此時(shí)RDD中的每個(gè)partition//都會(huì)序列化成一個(gè)大的字節(jié)數(shù)組，然后再持久化到內(nèi)存或磁盤中。// 序列化的方式可以減少持久化的數(shù)據(jù)對(duì)內(nèi)存/磁盤的占用量，進(jìn)而避免內(nèi)存被持久化數(shù)據(jù)占用過(guò)多，//從而發(fā)生頻繁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(...)

// 傳統(tǒng)的join操作會(huì)導(dǎo)致shuffle操作。// 因?yàn)閮蓚€(gè)RDD中，相同的key都需要通過(guò)網(wǎng)絡(luò)拉取到一個(gè)節(jié)點(diǎn)上，由一個(gè)task進(jìn)行join操作。val rdd3 = rdd1.join(rdd2)// Broadcast+map的join操作，不會(huì)導(dǎo)致shuffle操作。// 使用Broadcast將一個(gè)數(shù)據(jù)量較小的RDD作為廣播變量。val rdd2Data = rdd2.collect()val rdd2DataBroadcast = sc.broadcast(rdd2Data)// 在rdd1.map算子中，可以從rdd2DataBroadcast中，獲取rdd2的所有數(shù)據(jù)。// 然后進(jìn)行遍歷，如果發(fā)現(xiàn)rdd2中某條數(shù)據(jù)的key與rdd1的當(dāng)前數(shù)據(jù)的key是相同的，//那么就判定可以進(jìn)行join。// 此時(shí)就可以根據(jù)自己需要的方式，將rdd1當(dāng)前數(shù)據(jù)與rdd2中可以連接的數(shù)據(jù)，//拼接在一起（String或Tuple）。val rdd3 = rdd1.map(rdd2DataBroadcast...)// 注意，以上操作，建議僅僅在rdd2的數(shù)據(jù)量比較少（比如幾百M(fèi)，或者一兩G）的情況下使用。// 因?yàn)槊總€(gè)Executor的內(nèi)存中，都會(huì)駐留一份rdd2的全量數(shù)據(jù)。

// 以下代碼在算子函數(shù)中，使用了外部的變量。// 此時(shí)沒(méi)有做任何特殊操作，每個(gè)task都會(huì)有一份list1的副本。val list1 = ...rdd1.map(list1...)// 以下代碼將list1封裝成了Broadcast類型的廣播變量。// 在算子函數(shù)中，使用廣播變量時(shí)，首先會(huì)判斷當(dāng)前task所在Executor內(nèi)存中，是否有變量副本。// 如果有則直接使用；如果沒(méi)有則從Driver或者其他Executor節(jié)點(diǎn)上遠(yuǎn)程拉取一份放到本地Executor內(nèi)存中。// 每個(gè)Executor內(nèi)存中，就只會(huì)駐留一份廣播變量副本。val list1 = ...val list1Broadcast = sc.broadcast(list1)rdd1.map(list1Broadcast...)

// 創(chuàng)建SparkConf對(duì)象。val conf = new SparkConf().setMaster(...).setAppName(...)// 設(shè)置序列化器為KryoSerializer。conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")// 注冊(cè)要序列化的自定義類型。conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))

Spark性能優(yōu)化：資源調(diào)優(yōu)篇

./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性能優(yōu)化：數(shù)據(jù)傾斜調(diào)優(yōu)

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中的每個(gè)key都打上一個(gè)隨機(jī)前綴。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);}});// 第二步，對(duì)打上隨機(jī)前綴的key進(jìn)行局部聚合。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中每個(gè)key的隨機(jī)前綴。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);}});// 第四步，對(duì)去除了隨機(jī)前綴的RDD進(jìn)行全局聚合。JavaPairRDD globalAggrRdd = removedRandomPrefixRdd.reduceByKey(new Function2() {private static final long serialVersionUID = 1L;@Overridepublic Long call(Long v1, Long v2) throws Exception {  return v1 + v2;}});

// 首先將數(shù)據(jù)量比較小的RDD的數(shù)據(jù)，collect到Driver中來(lái)。List> rdd1Data = rdd1.collect()// 然后使用Spark的廣播功能，將小RDD的數(shù)據(jù)轉(zhuǎn)換成廣播變量，這樣每個(gè)Executor就只有一份RDD的數(shù)據(jù)。// 可以盡可能節(jié)省內(nèi)存空間，并且減少網(wǎng)絡(luò)傳輸性能開銷。final Broadcast>> rdd1DataBroadcast = sc.broadcast(rdd1Data);// 對(duì)另外一個(gè)RDD執(zhí)行map類操作，而不再是join類操作。JavaPairRDD> joinedRdd = rdd2.mapToPair(new PairFunction, String, Tuple2>() {  private static final long serialVersionUID = 1L;  @Override   public Tuple2> call(Tuple2 tuple)  throws Exception {  // 在算子函數(shù)中，通過(guò)廣播變量，獲取到本地Executor中的rdd1數(shù)據(jù)。  List> rdd1Data = rdd1DataBroadcast.value();  // 可以將rdd1的數(shù)據(jù)轉(zhuǎn)換為一個(gè)Map，便于后面進(jìn)行join操作。  Map rdd1DataMap = new HashMap();  for(Tuple2 data : rdd1Data) {  rdd1DataMap.put(data._1, data._2);}// 獲取當(dāng)前RDD數(shù)據(jù)的key以及value。String key = tuple._1;String value = tuple._2;// 從rdd1數(shù)據(jù)Map中，根據(jù)key獲取到可以join到的數(shù)據(jù)。Row rdd1Value = rdd1DataMap.get(key);return new Tuple2(key, new Tuple2(value, rdd1Value));}});// 這里得提示一下。// 上面的做法，僅僅適用于rdd1中的key沒(méi)有重復(fù)，全部是唯一的場(chǎng)景。// 如果rdd1中有多個(gè)相同的key，那么就得用flatMap類的操作，在進(jìn)行join的時(shí)候不能用map，而是得遍歷rdd1所有數(shù)據(jù)進(jìn)行join。// rdd2中每條數(shù)據(jù)都可能會(huì)返回多條join后的數(shù)據(jù)。

// 首先從包含了少數(shù)幾個(gè)導(dǎo)致數(shù)據(jù)傾斜key的rdd1中，采樣10%的樣本數(shù)據(jù)。JavaPairRDD sampledRDD = rdd1.sample(false, 0.1);// 對(duì)樣本數(shù)據(jù)RDD統(tǒng)計(jì)出每個(gè)key的出現(xiàn)次數(shù)，并按出現(xiàn)次數(shù)降序排序。// 對(duì)降序排序后的數(shù)據(jù)，取出top 1或者top 100的數(shù)據(jù)，也就是key最多的前n個(gè)數(shù)據(jù)。// 具體取出多少個(gè)數(shù)據(jù)量最多的key，由大家自己決定，我們這里就取1個(gè)作為示范。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中分拆出導(dǎo)致數(shù)據(jù)傾斜的key，形成獨(dú)立的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中分拆出不導(dǎo)致數(shù)據(jù)傾斜的普通key，形成獨(dú)立的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，就是那個(gè)所有key的分布相對(duì)較為均勻的rdd。// 這里將rdd2中，前面獲取到的key對(duì)應(yīng)的數(shù)據(jù)，過(guò)濾出來(lái)，分拆成單獨(dú)的rdd，并對(duì)rdd中的數(shù)據(jù)使用flatMap算子都擴(kuò)容100倍。// 對(duì)擴(kuò)容的每條數(shù)據(jù)，都打上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中分拆出來(lái)的導(dǎo)致傾斜的key的獨(dú)立rdd，每條數(shù)據(jù)都打上100以內(nèi)的隨機(jī)前綴。// 然后將這個(gè)rdd1中分拆出來(lái)的獨(dú)立rdd，與上面rdd2中分拆出來(lái)的獨(dú)立rdd，進(jìn)行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中分拆出來(lái)的包含普通key的獨(dú)立rdd，直接與rdd2進(jìn)行join。JavaPairRDD> joinedRDD2 = commonRDD.join(rdd2);// 將傾斜key join后的結(jié)果與普通key join后的結(jié)果，uinon起來(lái)。// 就是最終的join結(jié)果。JavaPairRDD> joinedRDD = joinedRDD1.union(joinedRDD2);

解決方案七：使用隨機(jī)前綴和擴(kuò)容RDD進(jìn)行join

// 首先將其中一個(gè)key分布相對(duì)較為均勻的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;}});// 其次，將另一個(gè)有數(shù)據(jù)傾斜key的RDD，每條數(shù)據(jù)都打上100以內(nèi)的隨機(jī)前綴。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);}});// 將兩個(gè)處理后的RDD進(jìn)行join即可。JavaPairRDD> joinedRDD = mappedRDD.join(expandedRDD);

解決方案八：多種方案組合使用

Spark性能優(yōu)化：Shuffle調(diào)優(yōu)篇

一、調(diào)優(yōu)概述

二、ShuffleManager發(fā)展概述

2.1、HashShuffleManager運(yùn)行原理

2.1.1、未經(jīng)優(yōu)化的HashShuffleManager

2.1.2、優(yōu)化后的HashShuffleManager

2.2、SortShuffleManager運(yùn)行原理

2.2.1、普通運(yùn)行機(jī)制

2.2.2、bypass運(yùn)行機(jī)制

三、shuffle相關(guān)參數(shù)調(diào)優(yōu)

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--

掃描下方二維碼
添加好友，備注【交流】
可私聊交流，也可進(jìn)資源豐富學(xué)習(xí)群