對多次使用的RDD進行持久化

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

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

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

因此對於這種情況，我們的建議是：對多次使用的RDD進行持久化。此時Spark就會根據你的持久化策略，將RDD中的資料儲存到記憶體或者磁碟中。以後每次對這個RDD進行運算元操作時，都會直接從記憶體或磁碟中提取持久化的RDD資料，然後執行運算元，而不會從源頭處重新計算一遍這個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").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").persist(StorageLevel.MEMORY_AND_DISK_SER)

rdd1.map(...)

rdd1.reduce(...)

對於persist()方法而言，我們可以根據不同的業務場景選擇不同的持久化級別。

Spark的持久化級別

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

• 預設情況下，效能最高的當然是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的級別，必須將所有資料都複製一份副本，併發送到其它節點上，資料複製以及網路傳輸會導致較大的效能開銷，除非是要求作業的高可用性，否則不建議使用。