處理鍵－值對

儘管Spark的大部操作支援包含所有物件型別的RDDs,但是還有一些操作只支援鍵－值對的的RDDs.最常見的是類似"洗牌"的操作，比如以鍵值來分組或聚合所有的元素。

在Scala裡，這些操作對包含2元組的RDD是自動可用的。（Scala語言內建的元組，通過(a,b)這樣的形式建立）。對鍵－值對可用的操作在PairRDDFunctions類裡，將自動包含在含有2元組的RDD裡。

比如，下面的程式碼使用在鍵－值對上使用reduceByKey操作來計算檔案中同一行出現的次數。

val lines = sc.textFile("data.txt")
val pairs = lines
 
.map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)

注：當使用自定義的物件作為鍵－值對的鍵值時，你必須保證自定義物件的equals()方法伴隨有一個匹配的hashcode()方法。

轉換：

下表列出了一些Spark中常用的轉換操作。

動作

下表是Spark中常用的動作操作:

shuffle 操作

Spark的一些特定操作會觸發shuffle事件。shuffle是Spark的一種機制，用來將資料重新分配,使它的不同分組跨分割槽。這通常涉及複製資料到不同的執行體和機器，所以shuffle操作被認為是一種複雜且昂貴的操作。

背景

為了瞭解shuffle過程中發生了什麼，我們用reduceByKey操作來說明。reduceByKey操作產生了一個新的RDD，原RDD中所有對應同一個鍵的值組合成一個元組－－－鍵和在該鍵對應的所有元素上執行一個reduce函式後的結果。該操作的挑戰是同一個鍵所對應的所有元素不一定在同一分割槽，甚至不在同一個機器上，但是它們必須被統一定位被計算結果。

在Spark裡,資料通常並不是跨分割槽分佈，而是分佈在便於進行特定操作的地方。在計算的過程中，單一任務將會在單一的分割槽上進行操作。這樣，為了為reduceByKey這個reduce任務的執行組織所有的資料，Spark需要執行一個所有對所有的操作，它必須從所有的分割槽中讀取所有鍵值對應的所有值，然後跨分割槽彙集值來計算每個鍵對應的最終結果－－－－－這就是shuffle.如下圖所示，

儘管shuffle操作後每個分割槽上的元素集合是確定的，但是分割槽的順序以及分割槽中元素的順序是不確定的。如果你希望預測shuffle操作後資料的順序，可以使用如下操作：

1.mapPartions用來對使用的每個分割槽排序，如.sorted

2.repartitionAndSortWithinPartitions高效地對分割槽排序並重新分割槽

3.sortBy來建立一個全域性有序的RDD

會引起shuffle的操作包括“重分割槽”操作，如repartition和coalesce."ByKey"操作（除了counting）比如groupByKey和reduceByKey,和"join"操作比如cogroup和join.

效能影響

Shuffle操作屬於昂貴的操作，因為它涉及磁碟I/O,資料序列化，和網路I/O。為是組織Shuffle的資料，Spark生成一個map任務集合來組織資料，和一個reduce任務集合來聚合資料，這種概念來自於MapReduce,和Spark自身的map,reduce操作沒有直接關係。

在實現上，每個單獨的map任務將結果儲存在記憶體中，直到無法分配記憶體。然後，這些結果以目標分割槽排序並存儲在單一的檔案中。在reduce中，reduce任務讀取已排序的相關塊.

某些shuffle操作會消耗大量的堆記憶體，因為它們使用基於記憶體的資料結構來組織記錄，在傳遞它們之前或之後。特別地， reduceByKey和aggregateByKey在map中建立這些結構然後'ByKey'操作在reduce中生成這些結構. 當沒有足夠的記憶體時， Spark將這些資料寫到磁碟上, 引起額外的磁碟I/O和垃圾回收處理。

shuffle也會產生大量的中間檔案。在Spark1.3.1中，這些檔案在Spark停止之前不會清除，這意味著長時間執行的Spark會消耗可用的磁碟空間。
這樣做是為了使shuffle操作不用重新計算如果需要使用上次的結果。臨時檔案目錄在配置sparkContext時spark.local.dir配置指定。

shuffle操作可以被一些配置引數調整，這些將在spark的配置指導中詳細說明。

RDD持久化

將資料通過操作持久化（或快取）在記憶體中是Spark的重要能力之一。當你快取了一個RDD，每個節點都快取了RDD的所有分割槽。這樣就可以在記憶體中進行計算，或從其它的動作中還原RDD(或派生自其的RDD)。這樣可以使以後在RDD上的動作更快（通常可以提高10倍）。快取對於遍歷演算法和更快的互動體驗至關重要。

你可以對希望快取的RDD通過使用persist或cache方法進行標記。它通過動作操作第一次在RDD上進行計算後，它就會被快取在節點上的記憶體中。Spark的快取具有容錯性，如果RDD的某一分割槽丟失，它會自動使用最初建立RDD時的轉換操作進行重新計算。

另外，RDD可以被持久化成不同的級別。比如，可以允許你儲存在磁碟，記憶體，甚至是序列化的JAVA物件（節省空間），備份在不同的節點上，或者儲存在基於記憶體的檔案系統Tachyon上。通過向persist()方法傳遞StorageLevel物件來設定。cache方法是使用預設級別StorageLevel.MEMORY_ONLY的方法。所有的級別如下：

Spark也自動地持久化一些中間資料在shuffle操作中，儘管使用者沒有呼叫persist.這樣做是為了避免重新計算整個輸入當一個節點失敗時。我們建議使用者對結果RDD呼叫persist，如果計劃重複使用它們。

選擇何種儲存級別？

Spark的儲存級別是為了提供記憶體使用率和CPU效率的均衡。我們建議您通過以下方式來選擇：

1.如果你的RDD在預設的儲存級別下工作的很好，就不要用其它的級別。這是最有CPU效率的選項，允許RDD上的操作儘快的完成。

2.如果不行，試下MEMORY_ONLY_SER並使用一個能快速序列化的庫，這樣更節省空間，同時訪問速度也比較快。

3.不要儲存資料到磁碟，除非在資料集上的計算操作是昂貴的，或者過濾了大量的資料。否則重新計算可能比從磁碟中讀取更快

4.使用備份級別，如果你需要更快的恢復。(比如，使用Spark為網路應用程式提供服務）。所有的儲存級別都通過重新計算提供了全面的容錯性，但是備份級別允許你繼續在RDD上執行任務而無需重新計算丟失的分割槽。

5.在擁有大量記憶體和多應用程式的環境中，實驗下OFF_HEAP方式有以下優勢：

 允許多個執行體共享同一個記憶體池。

 顯著地減少了垃圾回收的消……

 如果某個執行體崩潰，快取資料不會丟失。

刪除資料

Spark自動地監控每個節點上的快取使用率。並通過LRU演算法刪除過時的資料。如果你想手動刪除一個RDD而不是等待它從快取中過時，使用RDD.unpersist方法。