Spark其核心概念RDD是什麼？

我們知道在spark引擎裏其核心概念就是RDD。那麼RDD你又瞭解過多少呢？下面小編就為大家分享下RDD是什麼吧。

RDD全稱叫做彈性分佈式數據集(Resilient Distributed Datasets)，它是一種分佈式的內存抽象，表示一個只讀的記錄分區的集合，它只能通過其他RDD轉換而創建，為此，RDD支持豐富的轉換操作 ( 如: map, join, filter, groupBy 等)，通過這種轉換操作，新的RDD則包含了如何從其他RDDs衍生所必需的信息，所以説RDDs之間是有依賴關係的。

基於RDDs之間的依賴，RDDs會形成一個有向無環圖DAG，該DAG描述了整個流式計算的流程，實際執行的時候，RDD是通過血緣關係(Lineage)一氣呵成的，即使出現數據分區丟失，也可以通過血緣關係重建分區。

總結起來，基於RDD的流式計算任務可描述為：從穩定的物理存儲(如分佈式文件系統)中加載記錄，記錄被傳入由一組確定性操作構成的DAG，然後寫回穩定存儲。另外RDD還可以將數據集緩存到內存中，使得在多個操作之間可以重用數據集，基於這個特點可以很方便地構建迭代型應用(圖計算、機器學習等)或者交互式數據分析應用。

可以説Spark最初也就是實現RDD的一個分佈式系統，後面通過不斷髮展壯大成為現在較為完善的大數據生態系統，簡單來講，Spark-RDD的關係類似於Hadoop-MapReduce關係。

RDD特點

RDD表示只讀的分區的數據集，對RDD進行改動，只能通過RDD的轉換操作，由一個RDD得到一個新的RDD，新的RDD包含了從其他RDD衍生所必需的信息。

RDDs之間存在依賴，RDD的執行是按照血緣關係延時計算的。如果血緣關係較長，可以通過持久化RDD來切斷血緣關係。

分區

如下圖所示，RDD邏輯上是分區的，每個分區的數據是抽象存在的，計算的時候會通過一個compute函數得到每個分區的數據。

如果RDD是通過已有的文件系統構建，則compute函數是讀取指定文件系統中的數據，如果RDD是通過其他RDD轉換而來，則compute函數是執行轉換邏輯將其他RDD的數據進行轉換。

只讀

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

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

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

依賴

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

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

緩存

如果在應用程序中多次使用同一個RDD，可以將該RDD緩存起來，該RDD只有在第一次計算的時候會根據血緣關係得到分區的數據，在後續其他地方用到該RDD的時候，會直接從緩存處取而不用再根據血緣關係計算，這樣就加速後期的重用。

如下圖所示，RDD-1經過一系列的轉換後得到RDD-n並保存到hdfs，RDD-1在這一過程中會有個中間結果，如果將其緩存到內存，那麼在隨後的RDD-1轉換到RDD-m這一過程中，就不會計算其之前的RDD-0了。

Checkpoint

雖然RDD的血緣關係天然地可以實現容錯，當RDD的某個分區數據失敗或丟失，可以通過血緣關係重建。但是對於長時間迭代型應用來説，隨着迭代的進行，RDDs之間的血緣關係會越來越長，一旦在後續迭代過程中出錯，則需要通過非常長的血緣關係去重建，勢必影響性能。

為此，RDD支持checkpoint將數據保存到持久化的存儲中，這樣就可以切斷之前的血緣關係，因為checkpoint後的RDD不需要知道它的父RDDs了，它可以從checkpoint處拿到數據。

小結

總結起來，給定一個RDD我們至少可以知道如下幾點信息：1、分區數以及分區方式;2、由父RDDs衍生而來的相關依賴信息;3、計算每個分區的數據，計算步驟為：1)如果被緩存，則從緩存中取的分區的數據;2)如果被Checkpoint，則從Checkpoint處恢復數據;3)根據血緣關係計算分區的數據。

編程模型

在Spark中，RDD被表示為對象，通過對象上的方法調用來對RDD進行轉換。經過一系列的Transformations後，就可以調用Actions觸發RDD的計算，Action可以是嚮應用程序返回結果( count, collect 等)，或者是向存儲系統保存數據( saveAsTextFile 等)。在Spark中，只有遇到Action，才會執行RDD的計算(即懶執行)，這樣在運行時可以通過管道的方式傳輸多個轉換。

要使用Spark，開發者需要編寫一個Driver程序，它被提交到集羣以調度運行Worker，如下圖所示。Driver中定義了一個或多個RDD，並調用RDD上的action，Worker則執行RDD分區計算任務。

應用舉例

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

object WordCount {

def main(args: Array[String]) {

if (th < 2) {

tln("Usage: WordCount");

(1);

}

val conf = new SparkConf()ppName("WordCount")

val sc = new SparkContext(conf)

val result = File(args(0))

Map(line => t(" "))

(word => (word, 1))

ceByKey(_ + _)

AsTextFile(args(1))

}

}

結語

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

提供了豐富的操作算子，不再是隻有map和reduce兩個操作了，對於描述應用程序來説更加方便;

2.通過RDDs之間的轉換構建DAG，中間結果不用落地;

支持緩存，可以在內存中快速完成計算。