性能優(yōu)化概覽
why
Spark是基于內(nèi)存的計算，所以集群的CPU、網(wǎng)絡(luò)帶寬、內(nèi)存等都可能成為性能的瓶頸。
when
Spark應(yīng)用開發(fā)成熟時，滿足業(yè)務(wù)要求后，就可以開展性能優(yōu)化了。
what
一般來說，Spark應(yīng)用程序80%的優(yōu)化集中在內(nèi)存、磁盤IO、網(wǎng)絡(luò)IO，即Driver、Executor的內(nèi)存、shuffle的設(shè)置、文件系統(tǒng)的配置，集群的搭建，集群和文件系統(tǒng)的搭建（文件系統(tǒng)的集群在同一個局域網(wǎng)內(nèi)）。
how
web UI+log是Spark性能優(yōu)化的倚天劍和屠龍刀。
driver的log信息大致如“INFO BlockManagerMasterActor: Added rdd_0_1 in memory on mbk.local:50311 (size: 717.5 KB, free: 332.3 MB)”的日志信息。這就顯示了每個partition占用了多少內(nèi)存。
內(nèi)存都去哪了
Java對象頭
每個Java對象，都有一個對象頭，會占用16個字節(jié)，主要是包括了一些對象的元信息，比如指向它的類的指針。如果一個對象本身很小，比如就包括了一個int類型的field，那么它的對象頭實際上比對象自己還要大。
String對象
Java的String對象會比它內(nèi)部的原始數(shù)據(jù)多出40個字節(jié)。因為它內(nèi)部使用char數(shù)組來保存內(nèi)部的字符序列的，并且還得保存諸如數(shù)組長度之類的信息；而且String使用的是UTF-16編碼，每個字符會占用2個字節(jié)。比如，包含10個字符的String，會占用60個字節(jié)。
集合類型
Java中的集合類型，比如HashMap和LinkedList，內(nèi)部使用的是鏈表數(shù)據(jù)結(jié)構(gòu)，所以對鏈表中的每一個數(shù)據(jù)，都使用了Entry對象來包裝。Entry對象不光有對象頭，還有指向下一個Entry的指針，通常占用8個字節(jié)。
其他
元素類型為原始數(shù)據(jù)類型（比如int）的集合，內(nèi)部通常會使用原始數(shù)據(jù)類型的包裝類型，比如Integer，來存儲元素。
List<Integer> list = new ArrayList<Integer>()
性能優(yōu)化方法
數(shù)據(jù)序列化
Spark默認(rèn)序列化機制
Spark自身對于序列化的便捷性和性能進行了一個取舍和權(quán)衡。默認(rèn)，Spark傾向于序列化的便捷性，使用了Java自身提供的序列化機制——基于ObjectInputStream和ObjectOutputStream的序列化機制。
Java序列化機制的缺陷
Java序列化機制的性能并不高，序列化的速度相對較慢；而且序列化以后的數(shù)據(jù)，還是相對來說比較大，還是比較占用內(nèi)存空間。
Kryo序列化機制
Spark也支持使用Kryo類庫來進行序列化。Kryo序列化機制比Java序列化機制更快，而且序列化后的數(shù)據(jù)占用的空間更小，通常比Java序列化的數(shù)據(jù)占用的空間要小10倍。SparkConf().set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
Kryo使用場景
算子函數(shù)使用到了外部的大數(shù)據(jù)的情況。
比如自定義了一個MyConfiguration對象，里面包含了100m的數(shù)據(jù)。然后，在算子函數(shù)里面，使用到了這個外部的大對象。
conf.registerKryoClasses(XXX.class)
優(yōu)化Kryo緩存大小
如果注冊的要序列化的自定義的類型，本身特別大，就需要調(diào)整Kryo緩存的大小，默認(rèn)值是2M。SparkConf.set(“spark.kryoserializer.buffer.mb”,nM)。
數(shù)據(jù)結(jié)構(gòu)優(yōu)化
場景
算子中用到的內(nèi)部和外部的數(shù)據(jù)，優(yōu)化之后，會減少內(nèi)存的消耗和占用。
優(yōu)先使用數(shù)組以及字符串而不是集合類
比如將
List<Integer> list = new ArrayList<Integer>()
替換為
int[] arr = new int[]
這樣array既比List少了額外信息的存儲開銷，還能使用原始數(shù)據(jù)類型（int）來存儲數(shù)據(jù)，要節(jié)省內(nèi)存的多。
將
Map<Integer, Person> persons = new HashMap<Integer, Person>()
優(yōu)化為特殊的字符串格式
id:name,address|id:name,address...。
避免使用多層嵌套的對象結(jié)構(gòu)
如
public class Teacher {
private List<Student> students = new ArrayList<Student>()
}
就是非常不好的例子。因為Teacher類的內(nèi)部又嵌套了大量的小Student對象。優(yōu)化為json字符串來存儲數(shù)據(jù)
{
"teacherId": 1,
"teacherName": "leo",
students:[
{"studentId": 1, "studentName":"tom"},
{"studentId":2, "studentName":"marry"}
]
}
盡量使用int替代String
如用int行ID替代UUID等。
RDD持久化
持久化的場景
對RDD反復(fù)使用和重要的、關(guān)鍵的、耗時長的RDD。
持久化方法
使用cache（）|persist（）方法進行持久化，使用unpersist（）方法取消持久化。
持久化策略
Spark提供的多種持久化級別，主要是為了在CPU和內(nèi)存消耗之間進行取舍。優(yōu)先使用MEMORY_ONLY,內(nèi)存不足時使用MEMORY_ONLY_SER。
注意事項
JavaRDD<String> targetwords = words.filter(new Function<String, Boolean>() {}).cache();
不應(yīng)該是
JavaRDD<String> targetwords = words.filter(new Function<String, Boolean>() {})；
targetwords.cache();
Spark自己也會在shuffle操作時進行數(shù)的持久化，主要是為了在節(jié)點失敗時避免重算整個過程。
提高并行度
Spark集群的資源并不一定會被充分利用到，所以要盡量設(shè)置合理的并行度，來充分地利用集群的資源，以充分提高Spark應(yīng)用程序的性能。
Spark會自動設(shè)置以文件作為輸入源的RDD的并行度，依據(jù)其大小，比如HDFS，就會給每一個block創(chuàng)建一個partition，也依據(jù)這個設(shè)置并行度。對于reduceByKey等會發(fā)生shuffle的操作，就使用并行度最大的父RDD的并行度即可。
手動使用textFile()、parallelize()等方法的第二個參數(shù)來設(shè)置并行度；
使用spark.default.parallelism參數(shù)來設(shè)置統(tǒng)一的并行度
Spark官方的推薦是，給集群中的每個cpu core設(shè)置2~3個task。
比如說，spark-submit設(shè)置了executor數(shù)量是10個，每個executor要求分配2個core，那么application總共會有20個core。此時可以設(shè)置new SparkConf().set("spark.default.parallelism", "60")
來設(shè)置合理的并行度，從而充分利用資源。
廣播共享數(shù)據(jù)
優(yōu)化前
默認(rèn)情況下，算子函數(shù)使用到的外部數(shù)據(jù)，會被拷貝到每個task中，如果使用到的外部數(shù)據(jù)很大，那么就會占用大量的內(nèi)存空間和網(wǎng)絡(luò)傳輸。
優(yōu)化后
外部數(shù)據(jù)在每個節(jié)點上只保留一份副本，大大節(jié)省了內(nèi)存和網(wǎng)絡(luò)傳輸。
廣播共享數(shù)據(jù)的用戶
創(chuàng)建廣播變量
...
Broadcast<T> broadcast = sc.broadcast(T);
...
使用廣播變量
...
broadcast.value();
...
數(shù)據(jù)本地化
數(shù)據(jù)本地化對性能的影響
數(shù)據(jù)本地化對于Spark Job性能有著巨大的影響，如果數(shù)據(jù)與要計算它的代碼是在一起的，那么性能當(dāng)然會非常高。Spark傾向于使用最好的本地化級別來調(diào)度task，如果沒有任何未處理的數(shù)據(jù)在空閑的executor上，那么Spark就會放低本地化級別。這時有兩個選擇：等待直到executor上的cpu釋放出來，那么就分配task過去或者立即在任意一個executor上啟動一個task。
數(shù)據(jù)本地化級別
PROCESS_LOCAL：數(shù)據(jù)和計算它的代碼在同一個JVM進程中。
NODE_LOCAL：數(shù)據(jù)和計算它的代碼在一個節(jié)點上，但是不在一個進程中；
NO_PREF：數(shù)據(jù)從哪里過來，性能都是一樣的。
RACK_LOCAL：數(shù)據(jù)和計算它的代碼在一個機架上。
ANY：數(shù)據(jù)可能在任意地方，比如其他網(wǎng)絡(luò)環(huán)境內(nèi)，或者其他機架上。
優(yōu)化參數(shù)
spark.locality.wait（3000毫秒）
spark.locality.wait.node
spark.locality.wait.process
spark.locality.wait.rack
reduceByKey和groupByKey優(yōu)化
如果能用reduceByKey，那就用reduceByKey，因為它會在map端，先進行本地combine，可以大大減少要傳輸?shù)絩educe端的數(shù)據(jù)量，減小網(wǎng)絡(luò)傳輸?shù)拈_銷。
只有在reduceByKey處理不了時，才用groupByKey().map()來替代。
JVN垃圾回收調(diào)優(yōu)
GC對性能的影響
默認(rèn)情況下，Executor的內(nèi)存空間60%用于RDD的緩存，40%分配給Task用于運行。Task很可能很快就耗光了內(nèi)存而觸發(fā)GC。GC發(fā)生時將停止一切工作線程，GC本身需要花費時間，如果再頻繁發(fā)生GC，將嚴(yán)重影響Spark應(yīng)用程序的性能。
GC 優(yōu)化
可通過調(diào)整比例達(dá)到優(yōu)化GC的目的。
SparkConf().set(“spark.storage.memoryFraction”, “0.5”)
比值在0.6~0.1之間調(diào)整。
若配合使用序列化持久化級別如MEMORY_ONLY_SER何kryo等手段，將會有更好的性能優(yōu)化。
shuffle優(yōu)化
spark.shuffle.consolidateFiles：是否開啟shuffle block file的合并，默認(rèn)為false
spark.reducer.maxSizeInFlight：reduce task的拉取緩存，默認(rèn)48m
spark.shuffle.file.buffer：map task的寫磁盤緩存，默認(rèn)32k
spark.shuffle.io.maxRetries：拉取失敗的最大重試次數(shù)，默認(rèn)3次
spark.shuffle.io.retryWait：拉取失敗的重試間隔，默認(rèn)5s
spark.shuffle.memoryFraction：用于reduce端聚合的內(nèi)存比例，默認(rèn)0.2，超過比例就會溢出到磁盤上
