文章正文

RDD全稱叫做彈性分散式資料集(Resilient Distributed Datasets)，它是一種分散式的記憶體抽象，表示一個只讀的記錄分割槽的集合，它只能通過其他RDD轉換而建立，為此，RDD支援豐富的轉換操作(如map, join, filter, groupBy等)，通過這種轉換操作，新的RDD則包含了如何從其他RDDs衍生所必需的資訊，所以說RDDs之間是有依賴關係的。基於RDDs之間的依賴，RDDs會形成一個有向無環圖DAG，該DAG描述了整個流式計算的流程，實際執行的時候，RDD是通過血緣關係(Lineage)一氣呵成的，即使出現資料分割槽丟失，也可以通過血緣關係重建分割槽，總結起來，基於RDD的流式計算任務

1、RDD特點

RDD表示只讀的分割槽的資料集，對RDD進行改動，只能通過RDD的轉換操作，由一個RDD得到一個新的RDD，新的RDD包含了從其他RDD衍生所必需的資訊。RDDs之間存在依賴，RDD的執行是按照血緣關係延時

1.1 分割槽

如下圖所示，RDD邏輯上是分割槽的，每個分割槽的資料是抽象存在的，計算的時候會通過一個compute函式得到每個分割槽的資料。如果RDD是通過已有的檔案系統構建，則compute函式是讀取指定檔案系統中的資料，如果RDD是通過其他RDD轉換而來，則compute函式是執行轉換邏輯將其他RDD的資料進行轉換。

1.2 只讀

如下圖所示，RDD是隻讀的，要想改變RDD中的資料，只能在現有的RDD基礎上建立新的RDD。

由一個RDD轉換到另一個RDD，可以通過豐富的操作運算元實現，不再像MapReduce那樣只能寫map和reduce了，如下圖所示。

RDD的操作運算元包括兩類，一類叫做transformations，它是用來將RDD進行轉化，構建RDD的血緣關係；另一類叫做actions，它是用來觸發RDD的計算，得到RDD的相關計算結果或者將RDD儲存的檔案系統中。下圖是RDD所支援的操作運算元列表。

1.3 依賴

RDDs通過操作運算元進行轉換，轉換得到的新RDD包含了從其他RDDs衍生所必需的資訊，RDDs之間維護著這種血緣關係，也稱之為依賴。如下圖所示，依賴包括兩種，一種是窄依賴，RDDs之間分割槽是一一對應的，另一種是寬依賴，下游RDD的每個分割槽與上游RDD(也稱之為父RDD)的每個分割槽都有關，是多對多的關係。

通過RDDs之間的這種依賴關係，一個任務流可以描述為DAG(有向無環圖)，如下圖所示，在實際執行過程中寬依賴對應於Shuffle(圖中的reduceByKey和join)，窄依賴中的所有轉換操作可以通過類似於管道的方式一氣呵成執行(圖中map和union可以一起執行)。

1.4 快取

如果在應用程式中多次使用同一個RDD，可以將該RDD快取起來，該RDD只有在第一次計算的時候會根據血緣關係得到分割槽的資料，在後續其他地方用到該RDD的時候，會直接從快取處取而不用再根據血緣關係計算，這樣就加速後期的重用。如下圖所示，RDD-1經過一系列的轉換後得到RDD-n並儲存到hdfs，RDD-1在這一過程中會有個中間結果，如果將其快取到記憶體，那麼在隨後的RDD-1轉換到RDD-m這一過程中，就不會計算其之前的RDD-0了。

1.5 checkpoint

雖然RDD的血緣關係天然地可以實現容錯，當RDD的某個分割槽資料失敗或丟失，可以通過血緣關係重建。但是對於長時間迭代型應用來說，隨著迭代的進行，RDDs之間的血緣關係會越來越長，一旦在後續迭代過程中出錯，則需要通過非常長的血緣關係去重建，勢必影響效能。為此，RDD支援checkpoint將資料儲存到持久化的儲存中，這樣就可以切斷之前的血緣關係，因為checkpoint後的RDD不需要知道它的父RDDs了，它可以從checkpoint處拿到資料。

1.6 小結

總結起來，給定一個RDD我們至少可以知道如下幾點資訊：

• 分割槽數以及分割槽方式；
• 由父RDDs衍生而來的相關依賴資訊；
• 計算每個分割槽的資料，計算步驟為：
  • 如果被快取，則從快取中取的分割槽的資料；
  • 如果被checkpoint，則從checkpoint處恢復資料；
  • 根據血緣關係計算分割槽的資料。

2、程式設計模型

在Spark中，RDD被表示為物件，通過物件上的方法呼叫來對RDD進行轉換。經過一系列的transformations定義RDD之後，就可以呼叫actions觸發RDD的計算，action可以是嚮應用程式返回結果(count, collect等)，或者是向儲存系統儲存資料(saveAsTextFile等)。在Spark中，只有遇到action，才會執行RDD的計算(即延遲計算)，這樣在執行時可以通過管道的方式傳輸多個轉換。

要使用Spark，開發者需要編寫一個Driver程式，它被提交到叢集以排程執行Worker，如下圖所示。Driver中定義了一個或多個RDD，並呼叫RDD上的action，Worker則執行RDD分割槽計算任務。

3、應用舉例

下面介紹一個簡單的spark應用程式例項WordCount，統計一個數據集中每個單詞出現的次數，首先將從hdfs中載入資料得到原始RDD-0，其中每條記錄為資料中的一行句子，經過一個flatMap操作，將一行句子切分為多個獨立的詞，得到RDD-1，再通過map操作將每個詞對映為key-value形式，其中key為詞本身，value為初始計數值1，得到RDD-2，將RDD-2中的所有記錄歸併，統計每個詞的計數，得到RDD-3，最後將其儲存到hdfs。

import org.apache.spark._
import SparkContext._

object WordCount {
  def main(args: Array[String]) {
    if (args.length < 2) {
      System.err.println("Usage: WordCount <inputfile> <outputfile>");
      System.exit(1);
    }
    val conf = new SparkConf().setAppName("WordCount")
    val sc = new SparkContext(conf)
    val result = sc.textFile(args(0))
                   .flatMap(line => line.split(" "))
                   .map(word => (word, 1))
                   .reduceByKey(_ + _)
    result.saveAsTextFile(args(1))
  }
}

4、小結

基於RDD實現的Spark相比於傳統的Hadoop MapReduce有什麼優勢呢？總結起來應該至少有三點：

• RDD提供了豐富的操作運算元，不再是隻有map和reduce兩個操作了，對於描述應用程式來說更加方便；
• 通過RDDs之間的轉換構建DAG，中間結果不用落地；
• RDD支援快取，可以在記憶體中快速完成計算。

文章來源

http://sharkdtu.com/posts/spark-rdd.html