除了要避免在開發過程中對一份完全相同的數據創建多個RDD之外，在對不同的數據執行算子操作時還要儘可能地複用一個RDD。

比如說，有一個RDD的數據格式是key-value類型的，另一個是單value類型的，這兩個RDD的value數據是完全一樣的，那麼此時我們可以只使用key-value類型的那個RDD，因為其中已經包含了另一個的數據。

對於類似這種多個RDD的數據有重疊或者包含的情況，我們應該儘量複用一個RDD，這樣可以儘可能地減少RDD的數量，從而儘可能減少算子執行的次數。

原則三：對多次使用的RDD進行持久化

當你在Spark代碼中多次對一個RDD做了算子操作後，恭喜，你已經實現Spark作業第一步的優化了，也就是儘可能複用RDD。此時就該在這個基礎之上，進行第二步優化了，也就是要保證對一個RDD執行多次算子操作時，這個RDD本身僅僅被計算一次。

Spark中對於一個RDD執行多次算子的默認原理是這樣的：每次你對一個RDD執行一個算子操作時，都會重新從源頭處計算一遍，計算出那個RDD來，然後再對這個RDD執行你的算子操作。

這種方式的性能是很差的。

因此對於這種情況，我們的建議是：對多次使用的RDD進行持久化。此時Spark就會根據你的持久化策略，將RDD中的數據保存到內存或者磁盤中。以後每次對這個RDD進行算子操作時，都會直接從內存或磁盤中提取持久化的RDD數據，然後執行算子，而不會從源頭處重新計算一遍這個RDD，再執行算子操作。

如何選擇一種最合適的持久化策略：

• 默認情況下，性能最高的當然是MEMORY_ONLY，但前提是你的內存必須足夠足夠大，可以綽綽有餘地存放下整個RDD的所有數據。因為不進行序列化與反序列化操作，就避免了這部分的性能開銷；對這個RDD的後續算子操作，都是基於純內存中的數據的操作，不需要從磁盤文件中讀取數據，性能也很高；而且不需要複製一份數據副本，並遠程傳送到其它節點上。但是這裡必須要注意的是，在實際的生產環境中，恐怕能夠直接用這種策略的場景還是有限的，如果RDD中數據比較多時（比如幾十億），直接用這種持久化級別，會導致JVM的OOM內存溢出異常。
• 如果使用MEMORY_ONLY級別時發生了內存溢出，那麼建議嘗試使用MEMORY_ONLY_SER級別。該級別會將RDD數據序列化後再保存在內存中，此時每個partition僅僅是一個字節數組而已，大大減少了對象數量，並降低了內存佔用。這種級別比MEMORY_ONLY多出來的性能開銷，主要就是序列化與反序列化的開銷。但是後續算子可以基於純內存進行操作，因此性能總體還是比較高的。此外，可能發生的問題同上，如果RDD中的數據量過多的話，還是可能會導致OOM內存溢出的異常。
• 如果純內存的級別都無法使用，那麼建議使用MEMORY_AND_DISK_SER策略，而不是MEMORY_AND_DISK策略。因為既然到了這一步，就說明RDD的數據量很大，內存無法完全放下。序列化後的數據比較少，可以節省內存和磁盤的空間開銷。同時該策略會優先儘量嘗試將數據緩存在內存中，內存緩存不下才會寫入磁盤。
• 通常不建議使用DISK_ONLY和後綴為_2的級別。因為完全基於磁盤文件進行數據的讀寫，會導致性能急劇降低，有時還不如重新計算一次所有RDD。後綴為_2的級別，必須將所有數據都複製一份副本，併發送到其它節點上，數據複製以及網絡傳輸會導致較大的性能開銷，除非是要求作業的高可用性，否則不建議使用。

原則四：儘量避免使用shuffle類算子

如果有可能的話，要儘量避免使用shuffle類算子。因為Spark作業運行過程中，最消耗性能的地方就是shuffle過程。shuffle過程，簡單來說，就是將分佈在集群中多個節點上的同一個key拉取到同一個節點上，進行聚合或join等操作。比如reduceByKey、join等算子，都會觸發shuffle操作。

shuffle過程中，各個節點上的相同key都會先寫入本地磁盤文件中，然後其它節點需要通過網絡傳輸拉取各個節點上的磁盤文件中的相同key。而且相同key都拉取到同一個節點進行聚合操作時，還有可能會因為一個節點上處理的key過多，導致內存不夠存放，進而溢寫到磁盤文件中。因此在shuffle過程中，可能會發生大量的磁盤文件讀寫的IO操作，以及數據的網絡傳輸操作。磁盤IO和網絡數據傳輸也是shuffle性能較差的主要原因。

所以在我們的開發過程中，能避免則儘可能避免使用reduceByKey、join、distinct、repartition等會進行shuffle的算子，儘量使用map類的非shuffle算子。這樣的話，沒有shuffle操作或者僅有較少shuffle操作的Spark作業，可以大大減少性能開銷。

原則五：使用map-side預聚合的shuffle操作

如果因為業務需要，一定要使用shuffle操作，無法用map類的算子來替代，那麼儘量使用可以map-side預聚合的算子。

所謂的map-side預聚合，說的是在每個節點本地對相同的key進行一次聚合操作，類似於MapReduce中的本地combiner。

map-side預聚合之後，每個節點本地就只會有一條相同的key，因為多條相同的key都被聚合起來了。其它節點在拉取所有節點上的相同key時，就會大大減少需要拉取的數據數量，從而也就減少了磁盤IO以及網絡傳輸開銷。

通常來說，在可能的情況下，建議使用reduceByKey或者aggregateByKey算子來替代掉groupByKey算子。因為reduceByKey和aggregateByKey算子都會使用用戶自定義的函數對每個節點本地的相同key進行預聚合。而groupByKey算子是不會進行預聚合的，全量的數據會在集群的各個節點之間分發和傳輸，性能相對來說比較差。

比如以下兩幅圖就是典型的例子，分別基於reduceByKey和groupByKey進行單詞計數。其中第一張圖是groupByKey的原理圖，可以看到，沒有進行任何本地聚合時，所有數據都會在集群節點之間傳輸；第二張圖是reduceByKey的原理圖，可以看到，每個節點本地的相同key數據，都進行了預聚合，然後才傳輸到其他節點上進行全局聚合。

原則六：使用高性能的算子

除了shuffle相關的算子有優化原則之外，其它的算子也都有著相應的優化原則：

• 使用reduceByKey/aggregateByKey替代groupByKey。
• 詳情見“原則五：使用map-side預聚合的shuffle操作”。
• 使用mapPartitions替代普通map。
• mapPartitions類的算子，一次函數調用會處理一個partition所有的數據，而不是一次函數調用處理一條，性能相對來說會高一些。但是有的時候，使用mapPartitions會出現OOM（內存溢出）的問題。因為單次函數調用就要處理掉一個partition所有的數據，如果內存不夠，垃圾回收時是無法回收掉太多對象的，很可能出現OOM異常。所以使用這類操作時要慎重！
• 使用foreachPartitions替代foreach。
• 原理類似於“使用mapPartitions替代map”，也是一次函數調用處理一個partition的所有數據，而不是一次函數調用處理一條數據。在實踐中發現，foreachPartitions類的算子，對性能的提升還是很有幫助的。比如在foreach函數中，將RDD中所有數據寫MySQL，那麼如果是普通的foreach算子，就會一條數據一條數據地寫，每次函數調用可能就會創建一個數據庫連接，此時就勢必會頻繁地創建和銷燬數據庫連接，性能是非常低下；但是如果用foreachPartitions算子一次性處理一個partition的數據，那麼對於每個partition，只要創建一個數據庫連接即可，然後執行批量插入操作，此時性能是比較高的。實踐中發現，對於1萬條左右的數據量寫MySQL，性能可以提升30%以上。
• 使用filter之後進行coalesce操作。
• 通常對一個RDD執行filter算子過濾掉RDD中較多數據後（比如30%以上的數據），建議使用coalesce算子，手動減少RDD的partition數量，將RDD中的數據壓縮到更少的partition中去。因為filter之後，RDD的每個partition中都會有很多數據被過濾掉，此時如果照常進行後續的計算，其實每個task處理的partition中的數據量並不是很多，有一點資源浪費，而且此時處理的task越多，可能速度反而越慢。因此用coalesce減少partition數量，將RDD中的數據壓縮到更少的partition之後，只要使用更少的task即可處理完所有的partition。在某些場景下，對於性能的提升會有一定的幫助。
• 使用repartitionAndSortWithinPartitions替代repartition與sort類操作。
• repartitionAndSortWithinPartitions是Spark官網推薦的一個算子。官方建議，如果是需要在repartition重分區之後還要進行排序，就可以直接使用repartitionAndSortWithinPartitions算子。因為該算子可以一邊進行重分區的shuffle操作，一邊進行排序。shuffle與sort兩個操作同時進行，比先shuffle再sort來說，性能可能是要高的。

原則七：廣播大變量

有時在開發過程中，會遇到需要在算子函數中使用外部變量的場景（尤其是大變量，比如100M以上的大集合），那麼此時就應該使用Spark的廣播（Broadcast）功能來提升性能。

在算子函數中使用到外部變量時，默認情況下，Spark會將該變量複製多個副本，通過網絡傳輸到task中，此時每個task都有一個變量副本。如果變量本身比較大的話（比如100M，甚至1G），那麼大量的變量副本在網絡中傳輸的性能開銷，以及在各個節點的Executor中佔用過多內存導致的頻繁GC，都會極大地影響性能。

因此對於上述情況，如果使用的外部變量比較大，建議使用Spark的廣播功能，對該變量進行廣播。

廣播後的變量，會保證每個Executor的內存中，只駐留一份變量副本，而Executor中的task執行時共享該Executor中的那份變量副本。這樣的話，可以大大減少變量副本的數量，從而減少網絡傳輸的性能開銷，並減少對Executor內存的佔用開銷，降低GC的頻率。

原則八：使用Kryo優化序列化性能

在Spark中，主要有三個地方涉及到了序列化：

• 在算子函數中使用到外部變量時，該變量會被序列化後進行網絡傳輸（見“原則七：廣播大變量”中的講解）。
• 將自定義的類型作為RDD的泛型類型時（比如JavaRDD、Student是自定義類型），所有自定義類型對象，都會進行序列化。因此這種情況下，也要求自定義的類必須實現Serializable接口。
• 使用可序列化的持久化策略時（比如MEMORY_ONLY_SER），Spark會將RDD中的每個partition都序列化成一個大的字節數組。

對於這三種出現序列化的地方，我們都可以通過使用Kryo序列化類庫，來優化序列化和反序列化的性能。

Spark默認使用的是Java的序列化機制，也就是ObjectOutputStream/ObjectInputStream API來進行序列化和反序列化。但是Spark同時支持使用Kryo序列化庫，Kryo序列化類庫的性能比Java序列化類庫的性能要高很多。

官方介紹，Kryo序列化機制比Java序列化機制，性能高10倍左右。Spark之所以默認沒有使用Kryo作為序列化類庫，是因為Kryo要求最好要註冊所有需要進行序列化的自定義類型，因此對於開發者來說，這種方式比較麻煩。

以下是使用Kryo的代碼示例，我們只要設置序列化類，再註冊要序列化的自定義類型即可（比如算子函數中使用到的外部變量類型、作為RDD泛型類型的自定義類型等）：

原則九：優化數據結構

Java中，有三種類型比較耗費內存：

• 對象，每個Java對象都有對象頭、引用等額外的信息，因此比較佔用內存空間。
• 字符串，每個字符串內部都有一個字符數組以及長度等額外信息。
• 集合類型，比如HashMap、LinkedList等，因為集合類型內部通常會使用一些內部類來封裝集合元素，比如Map.Entry。

因此Spark官方建議，在Spark編碼實現中，特別是對於算子函數中的代碼，儘量不要使用上述三種數據結構，儘量使用字符串替代對象，使用原始類型（比如Int、Long）替代字符串，使用數組替代集合類型，這樣儘可能地減少內存佔用，從而降低GC頻率，提升性能。

但在實踐中，我們往往會發現，要做到該原則其實並不容易。因為我們同時要考慮到代碼的可維護性，如果一個代碼中，完全沒有任何對象抽象，全部是字符串拼接的方式，那麼對於後續的代碼維護和修改，無疑是一場巨大的災難。

同理，如果所有操作都基於數組實現，而不使用HashMap、LinkedList等集合類型，那麼對於我們的編碼難度以及代碼可維護性，也是一個極大的挑戰。

因此筆者建議，可以在可能以及合適的情況下，使用佔用內存較少的數據結構，但是前提是要保證代碼的可維護性。

原文: https://tech.meituan.com/spark_tuning_basic.html

原文太長，所以拆開了

---

分享到:

---

閱讀更多 陳序猿9527 的文章

關鍵字: 離線 Spark 計算

---