六、spark--spark調優

[TOC]

一、spark調優概論

1.1 什么是spark調優

? spark的計算本質是分布式計算，程序的性能受集群中的任何因素的影響，如：CPU、網絡帶寬、內存等。一般情況下，如果內存足夠大，那么其他因素影響性能。然后出現調優需求時，更多是因為資源不夠用的情況，所以才需要調節資源的使用情況，更加高效的使用資源。比如如果內存比較緊張，不足以存放所有數據（10億條），需要針對內存的使用，進行調優來減少內存的消耗

1.2 spark調優的主要方向

? Spark的性能優化，大部分的工作，是對于內存的使用，進行調優。通常情況下，Spark 處理的程序數據量較小，內存足夠使用，只要保證網絡通常，一般不會出現大的性能問題。但是，Spark應用程序的性能問題往往出現在針對大數據量進行計算時（數據突增）。這種情況往往是現環境是無法滿足的，所以可能導致集群崩潰。
? 除了內存調優之外，還有一些手段可以優化性能。比如spark使用過程中有和mysql交互的話，此時調優也要考慮到mysql的性能問題。

1.3 spark調優的主要技術手段

1、使用高性能序列化類庫。目的減少序列化時間以及序列化后數據的大小
2、優化數據結構。目的減少內存占用
3、對多次使用的RDD進行持久化（RDD cache）、checkpoint
4、使用序列化的持久化級別：MEMORY_ONLY不序列化，MEMORY_ONLY_SER序列化。
MEMORY_ONLY比MEMORY_ONLY_SER要占用更多內存空間。
但是要注意，序列化會增加cpu使用成本，所以要權衡好
5、Java虛擬機垃圾回收調優。
6、Shuffle調優，90%的問題都是shuffle導致（1.x版本時此問題嚴重，到2.x版本，官網基本已優化，所以到2.x版本，這個問題可忽略）

其他性能優化的方式：
提高計算并行度
廣播共享數據

下面會針對這6點調優手段進行分析

二、診斷spark內存使用情況

2.1 內存花費（對象內存花費）

1、每個 java/scala對象，由兩部分組成，一個是對象頭，占用16字節，主要包含對象的一些元信息，比如指向它的類的指針。另一個是對象本身。如果對象比較小，比如int，它的對象頭比自己對象本身都大。

2、String對象，會比他內部的原始數據，多出40個字節，用于保存string類型的元信息
String內部使用char數組來保存字符串序列，并且還要保存諸如數組長度之類的信息。String使用UTF-16編碼，所以每個字符會占用2個字節。
比如：包含10個字符的String，占用 2*10 + 40 個字節。

3、集合類型，比如HashMap和LinkedList，內部使用鏈表數據結構，對鏈表中的每個數據，使用Entry對象包裝。Entry對象，不光有對象頭，還有指向下一個Entry的指針，占用8個字節。所以一句話就是，這種內部還包含多個對象的類型，占用內存更多。因為對象多了，除了對象本身數據占用內存之外，更多對象也就會有更多對象頭，占用了不少內存空間。

4、基本數據類型的集合，比如int集合，內部會使用對象的包裝類 Integer來存儲元素。

2.2 獲取spark程序內存使用情況

到driver日志目錄下查看程序運行日志

less ${spark_home}/work/app-xxxxxx/0/stderr
觀察到類似如下信息：
INFO MemoryStore: Block broadcast_1 stored as values in memory (estimated size 320.9 KB, free 366.0 MB)
        19/07/05 05:57:47 INFO MemoryStore: Block rdd_3_1 stored as values in memory (estimated size 26.0 MB, free 339.9 MB)
        19/07/05 05:57:47 INFO Executor: Finished task 1.0 in stage 0.0 (TID 1). 2212 bytes result sent to driver
        19/07/05 05:57:48 INFO MemoryStore: Block rdd_3_0 stored as values in memory (estimated size 26.7 MB, free 313.2 MB)

estimated size 320.9 KB：當前使用的內存大概大小
free 366.0 MB：剩余空閑內存大小

這樣就可以知道任務使用內存的情況了

三、spark調優技術手段

2.1 使用高性能序列化類庫

2.1.1 spark序列化的使用情況

? spark作為一個分布式系統，和其他分布式系統一樣，都需要序列化。任何一個分布式系統中，序列化都是很重要的一環。如果使用的序列化技術，操作很慢，序列化后數據量大，會導致分布式系統應用程序性能下降很多。所以，Spark性能優化的第一步，就是進行序列化性能的優化。
? spark在一些地方是會使用序列化，比如shuffle的時候，但是spark對便捷性和性能進行了取舍，spark為了便捷性，默認使用了java的序列化機制，java的序列化機制之前也講過，性能不高，序列化速度慢，序列化后數據大。所以一般生產中，最好修改spark使用 的序列化機制

2.1.2 配置spark使用kryo來序列化

? spark支持使用kryo來實現序列化。kryo序列化速度比java快，占用空間小，大概小10倍。但是使用起來，相對沒有那么便捷。
配置spark使用kryo：

spark在讀取配置時，會讀取conf目錄下的配置文件，其中有一個 spark-defaults.conf 文件就是用來指定spark的一些工作參數的。

vim spark-defaults.conf
spark.serializer        org.apache.spark.serializer.KryoSerializer

這就配置了使用kryo，當然也可以在spark程序中使用 conf對象來來設置
conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")

2.1.3 kryo類庫的優化

（1）優化緩存大小
如果注冊的序列化的自定義類型，本身特別大，比如包含了100個以上字段，就會導致序列化的對象過大。此時需要對kyro本身進行優化。因為kyro本身內部緩存不夠存放這么大的對象。

設置：spark.kryoserializer.buffer.max  參數值調大，即可。

（2）提前注冊自定義類型
使用kryo時，為了更高的性能，最好提前注冊需要序列化的類，如：

在sparkConf 對象中注冊
conf.registerKryoClasses(Array(classof[Student],classof[Teacher]))

注意：這里基本都針對自定義的類，而且用scala編寫spark項目時，其實不會涉及到太多自定義類，不像java

2.2 優化數據結構

2.2.1 概述

優化數據結構，主要在于避免語法特性中所導致的額外內存開銷。
核心：優化算子函數內部使用到的局部數據或者算子外部的數據。
目的：減少對內存的消耗和占用。

2.2.2 具體手段

（1）優先使用數組以及字符串，而不是集合類。

即：優先使用array，而不是ArrayList，LinkedList，hashMap
使用int[] 比 List<Integer> 節省內存。

前面也說過，集合類包含更多的額外數據，以及復雜的類結構，所以占用內存多。此舉就是為了將結構簡單化，滿足使用的情況下，越簡單越好

（2）將對象轉換成字符串。

在企業中，將HashMap，List這種數據，統一使用String拼接成特殊格式的字符串。
舉例：
Map<Integer,Person> persons = new HashMap<>()
優化為：
id:name,address,idCardNum,family......|id:name,address,idCardNum,family......

（3）避免使用多層嵌套對象結構。

public class Teacher{private List<Student> students = new ArrayList<>()}
以上例子不好，因為Teacher類的內部又嵌套了大量的小的Student對象。
改進：
轉成json，處理字符串
{"teacherId":1,....,students[{"studentId":1,.....}]}

（4）對于能夠避免的場景，使用int代替String

雖然String性能比List高，但是int占用更少內存。
比如：數據庫主鍵，id，推薦使用自增主鍵，而不是uuid。

2.3 RDD緩存

這個就很簡單了，主要是將多次使用的RDD緩存在內存中，避免再次使用時重復計算。實現方法看前面spark core的文章

2.4 使用序列化來進行緩存

默認情況下，進行RDD緩存時，RDD對象是沒有序列化的，也就是持久化級別為 MEMORY_ONLY。建議使用 MEMORY_ONLY_SER進行持久化，因為這種方式同時會進行序列化，序列化后占用更少的內存的空間。實現方法看前面spark core的文章

2.5 jvm調優

2.5.1 背景

? 如果在持久化RDD的時候，持久化了大量的數據，那么Java虛擬機的垃圾回收就可能成為一個性能瓶頸。Java虛擬機會定期進行垃圾回收，此時就會追蹤所有Java對象，并且在垃圾回收時，追中找到那些已經不再使用的對象，清理舊對象，給新對象騰出空間。
? 垃圾回收的性能開銷，和內存中的對象數量成正比。而且要注意一點， 在做Java虛擬機調優前，必須要做好上面其他調優工作，這樣才有意義。因為上面的調優工作，是為了節省內存的開銷，更好、更高效的使用內存。上面的優化比起進行jvm調優獲得的好處要大得多。并且jvm調優好了，但是上層應用沒有好的內存使用方式，jvm優化了也白搭。

2.5.2 gc原理

這里提到這個，更多是讓讀者自己去理解這個原理，隨便百度都可以找到了，這里不重復。

2.5.3 檢測垃圾回收

我們可以對垃圾回收進行監測，包括多久進行一次垃圾回收，以及每次垃圾回收耗費的時間。
在 spark-submit腳本中，添加一個配置：

--conf "spark.executor.extraJavaOptions=-verbose:gc -XX:+PrintGCDetails -XX:+PrintGCTimesStamps"

注意：輸出到worker的日志中，而不是driver日志。

/usr/local/spark-2.1.0-bin-hadoop2.7/work/app-20190705055405-0000/0
這是driver日志

/usr/local/spark-2.1.0-bin-hadoop2.7/logs
這是worker日志

2.5.4 優化Executor內存比例

? 對于GC調優來說，最重要的調節，RDD緩存占用的內存空間 與 算子執行是創建對象所占用的內存空間 的比例。默認情況下，Spark使用每個Executor 60%的內存空間來緩存RDD，那么在task執行期間創建的對象，只有40%的內存空間來存放對象。
? 在這種情況下，很有可能因為內存不足，task創建的對象過大，導致40%的內存空間不夠用，觸發Java虛擬機垃圾回收操作。在極端的情況下，垃圾回收操作會被頻繁觸發。
? 根據實際情況，可以增大對象存儲空間，減少gc發生概率，方式：

conf.set("spark.storage.memoryFraction",0.5)
將RDD緩存占用空間比例降低到50%

2.6 shuffle

? 以往在spark1.x版本中，如果有shuffle時，那么每個map task就會根據result task(也可以叫reduce task)的個數，對map結果進行分區，分別給不同的result task處理，每個分區產生一個文件。當map數量很多時，就會產生大量文件，這會帶來性能問題。
? 在spark2.x中，將一個map task輸出的數據都放在一個文件中，然后加上一個索引文件，用于標識不同分區數據在文件中的位置，這樣就保證了一個task只產生一個文件。從而降低了IO壓力