RDD概念：Resilient Distributed Datasets

RDD（Resilient Distributed Dataset）叫做彈性分布式數據集，是Spark中最基本的數據抽象，它代表一個不可變、可分區、里面的元素可并行計算的集合。RDD具有數據流模型的特點：自動容錯、位置感知性調度和可伸縮性。RDD允許用戶在執行多個查詢時顯式地將工作集緩存在內存中，后續的查詢能夠重用工作集，這極大地提升了查詢速度。

同時，RDD還提供了一組豐富的操作來操作數據。RDD是只讀的記錄分區的集合，只能通過在其他RDD執行確定的轉換操作（transformation操作）而創建。RDD可看作一個spark的對象，它本身存在于內存中，如對文件計算是一個RDD，等。一個RDD可以包含多個分區，每個分區就是一個dataset片段。但是RDD抽象出來的東西里面實際的數據，是分散在各個節點上面的，RDD可分區，分區的個數是我們可以指定的。默認情況下，一個hdfs塊就是一個分區。

RDD的屬性

一組分片（Partition），即數據集的基本組成單位。對于RDD來說，每個分片都會被一個計算任務處理，并決定并行計算的粒度。用戶可以在創建RDD時指定RDD的分片個數，如果沒有指定，那么就會采用默認值。默認值就是程序所分配到的CPU Core的數目。

一個計算每個分區的函數。Spark中RDD的計算是以分片為單位的，每個RDD都會實現compute函數以達到這個目的。compute函數會對迭代器進行復合，不需要保存每次計算的結果。

RDD之間的依賴關系。RDD的每次轉換都會生成一個新的RDD，所以RDD之間就會形成類似于流水線一樣的前后依賴關系。在部分分區數據丟失時，Spark可以通過這個依賴關系重新計算丟失的分區數據，而不是對RDD的所有分區進行重新計算。

一個Partitioner，即RDD的分片函數。當前Spark中實現了兩種類型的分片函數，一個是基于哈希的HashPartitioner，另外一個是基于范圍的RangePartitioner。只有對于于key-value的RDD，才會有Partitioner，非key-value的RDD的Parititioner的值是None。Partitioner函數不但決定了RDD本身的分片數量，也決定了parent RDD Shuffle輸出時的分片數量。

一個列表，存儲存取每個Partition的優先位置（preferred location）。對于一個HDFS文件來說，這個列表保存的就是每個Partition所在的塊的位置。按照“移動數據不如移動計算”的理念，Spark在進行任務調度的時候，會盡可能地將計算任務分配到其所要處理數據塊的存儲位置。

wordcount粗解RDD

懶惰調用

RDD將操作分為兩類：transformation與action。無論執行了多少次transformation操作，RDD都不會真正執行運算，只有當action操作被執行時，運算才會觸發。這就是Spark的惰性調用機制。

正式這種執行機制為是spark的高效計算的基礎，正是因為懶惰執行，spark才能更有效的運行于于內存，使得高效的共享內存機制避免了大量中間結果，從而避免了磁盤寫入寫出帶來的性能消耗，同時內部的存儲對象可以是JAVA對象也避免了不必要的序列化和反序列化。

而在這基礎之上，spark又實現了高效的容錯機制，以及根據依賴關系進行工作優化。

寬依賴和窄依賴

由于RDD是粗粒度的操作數據集，每個Transformation操作都會生成一個新的RDD，所以RDD之間就會形成類似流水線的前后依賴關系；RDD和它依賴的父RDD（s）的關系有兩種不同的類型，即窄依賴（narrow dependency）和寬依賴（wide dependency）。如圖所示顯示了RDD之間的依賴關系。

從圖中可知：

窄依賴：是指每個父RDD的一個Partition最多被子RDD的一個Partition所使用，例如map、filter、union等操作都會產生窄依賴；（獨生子女）

寬依賴：是指一個父RDD的Partition會被多個子RDD的Partition所使用，例如groupByKey、reduceByKey、sortByKey等操作都會產生寬依賴；（超生）

需要特別說明的是對join操作有兩種情況：

（1）圖中左半部分join：如果兩個RDD在進行join操作時，一個RDD的partition僅僅和另一個RDD中已知個數的Partition進行join，那么這種類型的join操作就是窄依賴，例如圖1中左半部分的join操作(join with inputs co-partitioned)；

（2）圖中右半部分join：其它情況的join操作就是寬依賴,例如圖1中右半部分的join操作(join with inputs not co-partitioned)，由于是需要父RDD的所有partition進行join的轉換，這就涉及到了shuffle，因此這種類型的join操作也是寬依賴。

同時從這兩種依賴關系我們也可以明確的看出了，窄依賴在運行上要優于寬依賴，但是在很多執行中寬依賴又是不可避免的。為了能夠更好更高效的運行數據，spark基于這兩種依賴關系對工作劃分進行可設計（具體內容請見文末）

RDD的高效容錯機制

一般來說，分布式數據集的容錯性有兩種方式：數據檢查點和記錄數據的更新。 面向大規模數據分析，數據檢查點操作成本很高，需要通過數據中心的網絡連接在機器之間復制龐大的數據集，而網絡帶寬往往比內存帶寬低得多，同時還需要消耗更多的存儲資源。

因此，Spark選擇記錄更新的方式，簡單點來說就是RDD運行之后保存RDD之間的依賴關系，當出現錯誤時，通過重新計算的方式來恢復數據。但是，如果更新粒度太細太多，那么記錄更新成本也不低。因此，RDD只支持粗粒度轉換，即只記錄單個塊上執行的單個操作，然后將創建RDD的一系列變換序列（每個RDD都包含了他是如何由其他RDD變換過來的以及如何重建某一塊數據的信息。因此RDD的容錯機制又稱“血統(Lineage)”容錯）記錄下來，以便恢復丟失的分區。

Lineage本質上很類似于數據庫中的重做日志（Redo Log），只不過這個重做日志粒度很大，是對全局數據做同樣的重做進而恢復數據。

除此之外RDD中有時候還需要設置檢查點來提高性能：

DAG中的Lineage過長，如果重算，則開銷太大（如在PageRank中）。

在寬依賴上做Checkpoint獲得的收益更大。

Spark 的任務劃分

構建Spark Application的運行環境（啟動SparkContext），SparkContext向資源管理器（可以是Standalone、Mesos或YARN）注冊并申請運行Executor資源；

資源管理器分配Executor資源并啟動StandaloneExecutorBackend，Executor運行情況將隨著心跳發送到資源管理器上；

SparkContext構建成DAG圖，將DAG圖分解成Stage，并把Taskset發送給Task Scheduler。Executor向SparkContext申請Task，Task?（我們之前提到的寬依賴與窄依賴的執行優化問題就在這里處理）?Scheduler將Task發放給Executor運行同時SparkContext將應用程序代碼發放給Executor。

Task在Executor上運行，運行完畢釋放所有資源。

RDD運行原理

?

那么 RDD在Spark架構中是如何運行的呢？總高層次來看，主要分為三步：

創建 RDD 對象

DAGScheduler模塊介入運算，計算RDD之間的依賴關系。RDD之間的依賴關系就形成了DAG

每一個JOB被分為多個Stage，劃分Stage的一個主要依據是當前計算因子的輸入是否是確定的，如果是則將其分在同一個Stage，避免多個Stage之間的消息傳遞開銷

對于寬依賴與窄依賴進行的認任務分發優化

*在spark中，會根據RDD之間的依賴關系將DAG圖（有向無環圖）劃分為不同的階段，對于窄依賴，由于partition依賴關系的確定性，partition的轉換處理就可以在同一個線程里完成，窄依賴就被spark劃分到同一個stage中，而對于寬依賴，只能等父RDD shuffle處理完成后，下一個stage才能開始接下來的計算。

因此spark劃分stage的整體思路是：從后往前推，遇到寬依賴就斷開，劃分為一個stage；遇到窄依賴就將這個RDD加入該stage中。因此在圖2中RDD C,RDD D,RDD E,RDDF被構建在一個stage中,RDD A被構建在一個單獨的Stage中,而RDD B和RDD G又被構建在同一個stage中。

這種分發方式可以使不同分區實現不同的流水線操作，有利于高效的運行與容錯機制，想象一下，當運行時當C1-->D1運行結束時候，就可以直接運行D1-->F1的操作了，這樣進行的并行計算有效的提升了計算性能。同時當出現錯誤時候，良好的stage劃分也減少了重新計算所帶來的成本。

在spark中，Task的類型分為2種：ShuffleMapTask和ResultTask；

簡單來說，DAG的最后一個階段會為每個結果的partition生成一個ResultTask，即每個Stage里面的Task的數量是由該Stage中最后一個RDD的Partition的數量所決定的！而其余所有階段都會生成ShuffleMapTask；之所以稱之為ShuffleMapTask是因為它需要將自己的計算結果通過shuffle到下一個stage中；也就是說上圖中的stage1和stage2相當于mapreduce中的Mapper,而ResultTask所代表的stage3就相當于mapreduce中的reducer。

Hadoop中MapReduce操作中的Mapper和Reducer在spark中的基本等量算子是map和reduceByKey;不過區別在于：Hadoop中的MapReduce天生就是排序的；而reduceByKey只是根據Key進行reduce，但spark除了這兩個算子還有其他的算子；因此從這個意義上來說，Spark比Hadoop的計算算子更為豐富。

參考

本文原為個人筆記，參考了眾多資料，包括但不止于以下內容。

Spark的容錯機制總結

Spark中的寬依賴和窄依賴

RDD理解與寬依賴窄依賴

Spark運行流程

總結

以上是生活随笔為你收集整理的RDD浅谈的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。