sparkCore-RDD詳解

時間 2019-11-11

原文原文鏈接

1.1 什麼是RDD

1.1.1 產生背景

當初設計RDD主要是爲了解決三個問題：html

Fast:Spark以前的Hadoop用的是MapReduce的編程模型，沒有很好的利用分佈式內存系統，中間結果都須要保存到external disk，運行效率很低。RDD模型是in-memory computing的，中間結果不須要被物化（materialized），它的persistence機制，能夠保存中間結果重複使用，對須要迭代運算的機器學習應用和交互式數據挖掘應用，加速顯著。Spark快還有一個緣由是開頭提到過的Delay Scheduling機制，它得益於RDD的Dependency設計。
General: MapReduce編程模型只能提供有限的運算種類（Map和Reduce），RDD但願支持更普遍更多樣的operators（map，flatMap，filter等等），而後用戶能夠任意地組合他們。

The ability of RDDs to accommodate computing needs that were previously met only by introducing new frameworks is, we believe, the most credible evidence of the power of the RDD abstraction.java

Fault tolerance: 其餘的in-memory storage on clusters，基本單元是可變的，用細粒度更新（fine-grained updates）方式改變狀態，如改變table/cell裏面的值，這種模型的容錯只能經過複製多個數據copy，須要傳輸大量的數據，容錯效率低下。而RDD是不可變的immutable，經過粗粒度變換（coarse-grained transformations），好比map，filter和join，能夠把相同的運算同時做用在許多數據單元上，這樣的變換隻會產生新的RDD而不改變舊的RDD。這種模型可讓Spark用Lineage很高效地容錯（後面會有介紹）。

1.1.2 RDD定義

A Resilient Distributed Dataset (RDD), the basic abstraction in Spark. Represents（表明） an immutable(不變的),partitioned collection of elements that can be operated on in parallel

RDD是spark的核心，也是整個spark的架構基礎，RDD是彈性分佈式集合（Resilient Distributed Datasets）的簡稱。node

1.1.3 RDD特色

immutable：只讀，任何操做都不會改變RDD自己，只會創造新的RDDgit
fault-tolerant：容錯，經過Lineage能夠高效容錯github
partitioned：分片，RDD以partition做爲最小存儲和計算單元，分佈在cluster的不一樣nodes上，一個node能夠有多個partitions，一個partition只能在一個node上web
in parallel：並行，一個Task對應一個partition，Tasks之間相互獨立能夠並行計算apache
persistence：持久化，用戶能夠把會被重複使用的RDDs保存到storage上（內存或者磁盤）編程
partitioning：分區，用戶能夠選擇RDD元素被partitioned的方式來優化計算，好比兩個須要被join的數據集能夠用相同的方式作hash-partitioned，這樣能夠減小shuffle提升性能緩存

1.1.4 RDD抽象概念

一個RDD定義了對數據的一個操做過程, 用戶提交的計算任務能夠由多個RDD構成。多個RDD能夠是對單個/多個數據的多個操做過程。多個RDD之間的關係使用依賴來表達。操做過程就是用戶自定義的函數。服務器

RDD(彈性分佈式數據集)去掉形容詞，主體爲：數據集。若是認爲RDD就是數據集，那就有點理解錯了。我的認爲：RDD是定義對partition數據項轉變的高階函數，應用到輸入源數據，輸出轉變後的數據，即：RDD是一個數據集到另一個數據集的映射，而不是數據自己。 這個概念相似數學裏的函數f(x) = ax^2 + bx + c。這個映射函數能夠被序列化，所要被處理的數據被分區後分布在不一樣的機器上，應用一下這個映射函數，得出結果，聚合結果。

這些集合是彈性的，若是數據集一部分丟失，則能夠對它們進行重建。具備自動容錯、位置感知調度和可伸縮性，而容錯性是最難實現的，大多數分佈式數據集的容錯性有兩種方式：數據檢查點和記錄數據的更新。對於大規模數據分析系統，數據檢查點操做成本高，主要緣由是大規模數據在服務器之間的傳輸帶來的各方面的問題，相比記錄數據的更新，RDD也只支持粗粒度的轉換共享狀態而非細粒度的更新共享狀態，也就是記錄如何從其餘RDD轉換而來(即lineage)，以便恢復丟失的分區。

RDDs 很是適合將相同操做應用在整個數據集的全部的元素上的批處理應用. 在這些場景下, RDDs 能夠利用血緣關係圖來高效的記住每個 transformations 的步驟, 而且不須要記錄大量的數據就能夠恢復丟失的分區數據. RDDs 不太適合用於須要異步且細粒度的更新共享狀態的應用, 好比一個 web 應用或者數據遞增的 web 爬蟲應用的存儲系統。

1.2 RDD特色

Internally, each RDD is characterized by five main properties:
- A list of partitions
- A function for computing each split
- A list of dependencies on other RDDs
- Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)
- Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

每一個特性都對應RDD.scala中的一個方法實現：

a list of partition 由多個機器裏面的partition組成的
a function for computing each split 並行計算
a list of dependencies on other RDDS rdd間存在依賴關係,記錄數據轉換間的依賴
a partitioner for key-vaue RDDS 可進行從新分區（只有key value的partition有）
a list of preferred locations to compute each spilt on 用最指望的位置進行計算

1.3 RDD操做

1.3.1 RDD建立

parallelize:從普通Scala集合建立

val data = Array(1, 2, 3, 4, 5)
val distData = sc.parallelize(data)

從Hadoop文件系統或與Hadoop兼容的其餘持久化存儲系統建立，如Hive、HBase

scala> val distFile = sc.textFile("data.txt")
distFile: org.apache.spark.rdd.RDD[String] = data.txt MapPartitionsRDD[10] at textFile at <console>:26

從父RDD轉換獲得新的RDD

val fromParent=distFile.map(s=>s.length)

1.3.2 操做方式

RDD在宏觀來看相似於java中對象的概念，咱們在java中對對象上做用一系列操做（方法）獲得最終結果。一樣的咱們在RDD上進行一系列操做（算子）將一個RDD轉換爲另外一個RDD，最終獲得咱們所須要的RDD。RDD算子主要包括：

Transformation算子：Transformation操做是延遲計算的，即從一個RDD轉換成另外一個RDD的轉換操做不是立刻執行，須要等到有Action操做時，才真正出發執行，如Map、Filter等操做
Action算子：Action算子會出發Spark提交做業（Job），並將數據輸出到Spark系統，如collect、count等

RDD操做特色 惰性求值：

transformation算子做用在RDD時，並非當即觸發計算，只是記錄須要操做的指令。等到有Action算子出現時才真正開始觸發計算。

textFile等讀取數據操做和persist和cache緩存操做也是惰性的

爲何要使用惰性求值呢：使用惰性求值能夠把一些操做合併到一塊兒來減小數據的計算步驟，提升計算效率。

從惰性求值角度看RDD就是一組spark計算指令的列表

1.3.4 緩存策略

RDD的緩存策略在StorageLevel中實現，經過對是否序列化，是否存儲多個副本等條件的組合造成了多種緩存方式。例如：MEMORY_ONLY_SER存儲在內存中並進行序列化，當內存不足時，不進行本地化；MEMORY_AND_DISK_2優先存儲內存中，內存中無空間時，存儲在本地磁盤，並有兩個副本。

class StorageLevel private(
    // 緩存方式
    private var _useDisk: Boolean, 		// 是否使用磁盤
    private var _useMemory: Boolean, 	// 是否使用內存
    private var _useOffHeap: Boolean,	// 是否使用堆外內存
    private var _deserialized: Boolean, // 是否序列化
    private var _replication: Int = 1)	// 存儲副本，默認一個
  extends Externalizable {
      
  // 條件組合結果
  val NONE = new StorageLevel(false, false, false, false)
  val DISK_ONLY = new StorageLevel(true, false, false, false)
  val DISK_ONLY_2 = new StorageLevel(true, false, false, false, 2)
  val MEMORY_ONLY = new StorageLevel(false, true, false, true)
  val MEMORY_ONLY_2 = new StorageLevel(false, true, false, true, 2)
  val MEMORY_ONLY_SER = new StorageLevel(false, true, false, false)
  val MEMORY_ONLY_SER_2 = new StorageLevel(false, true, false, false, 2)
  val MEMORY_AND_DISK = new StorageLevel(true, true, false, true)
  val MEMORY_AND_DISK_2 = new StorageLevel(true, true, false, true, 2)
  val MEMORY_AND_DISK_SER = new StorageLevel(true, true, false, false)
  val MEMORY_AND_DISK_SER_2 = new StorageLevel(true, true, false, false, 2)
  val OFF_HEAP = new StorageLevel(true, true, true, false, 1)

策略選擇順序：

默認選擇MEMORY_ONLY
若是內存不足，選擇MEMORY_ONLY_SER
若是須要作容錯,選擇MEMORY_ONLY_SER_2
若是中間計算RDD的代價比較大時，選擇MEMORY_AND_DISK

控制操做：

persist操做，能夠將RDD持久化到不一樣層次的存儲介質，以便後續操做重複使用。

　　 1)cache:RDD[T] 默認使用MEMORY_ONLY

　　 2)persist:RDD[T] 默認使用MEMORY_ONLY

　　 3)Persist(level:StorageLevel):RDD[T] eg: myRdd.persist(StorageLevels.MEMORY_ONLY_SER)

checkpoint

　　將RDD持久化到HDFS中，與persist操做不一樣的是checkpoint會切斷此RDD以前的依賴關係，而persist依然保留RDD的依賴關係。

1.3.5 RDD回收

Spark automatically monitors cache usage on each node and drops out old data partitions in a least-recently-used (LRU) fashion. If you would like to manually remove an RDD instead of waiting for it to fall out of the cache, use the `RDD.unpersist()` method.

spark有一個監控線程去檢測內存使用狀況，當內存不足時使用LRU進行淘汰old data,也能夠經過RDD.unpersist()方法手動移除緩存。

1.3.6 RDD保存

saveAsTextFile()將RDD中的元素保存在指定目錄中，這個目錄位於任何Hadoop支持的存儲系統中
saveAsObjectFile()將原RDD中的元素序列化成Java對象，存儲在指定目錄中
saveAsSequenceFile() 將鍵值對型RDD以SequenceFile的格式保存。鍵值對型RDD也能夠以文本形式保存

須要注意的是，上面的方法都把一個目錄名字做爲入參，而後在這個目錄爲每一個RDD分區建立一個文件夾。這種設計不只能夠高效並且可容錯。由於每一個分區被存成一個文件，因此Spark在保存RDD的時候能夠啓動多個任務，並行執行，將數據寫入文件系統中，這樣也保證了寫入數據的過程當中可容錯，一旦有一個分區寫入文件的任務失敗了，Spark能夠在重啓一個任務，重寫剛纔失敗任務建立的文件。

2. RDD詳解

2.1 RDD分區

RDD 表示並行計算的計算單元是使用分區（Partition）。

2.1.1 分區實現

RDD 內部的數據集合在邏輯上和物理上被劃分紅多個小子集合，這樣的每個子集合咱們將其稱爲分區，分區的個數會決定並行計算的粒度，而每個分區數值的計算都是在一個單獨的任務中進行，所以並行任務的個數，也是由 RDD（其實是一個階段的末 RDD，調度章節會介紹）分區的個數決定的。

RDD 只是數據集的抽象，分區內部並不會存儲具體的數據。Partition 類內包含一個 index 成員，表示該分區在 RDD 內的編號，經過 RDD 編號 + 分區編號能夠惟一肯定該分區對應的塊編號，利用底層數據存儲層提供的接口，就能從存儲介質（如：HDFS、Memory）中提取出分區對應的數據

怎麼切分是Partitioner定義的, Partitioner有兩個接口: numPartitions分區數, getPartition(key: Any): Int根據傳入的參數肯定分區號。實現了Partitioner的有：

HashPartitioner
RangePartitioner
GridPartitioner
PythonPartitioner

一個RDD有了Partitioner, 就能夠對當前RDD持有的數據進行劃分

2.1.2 分區個數

RDD 分區的一個分配原則是：儘量使得分區的個數等於集羣的CPU核數。

RDD 能夠經過建立操做或者轉換操做獲得。轉換操做中，分區的個數會根據轉換操做對應多個 RDD 之間的依賴關係肯定，窄依賴子 RDD 由父 RDD 分區個數決定，Shuffle 依賴由子 RDD 分區器決定。

建立操做中，程序開發者能夠手動指定分區的個數，例如 sc.parallelize (Array(1, 2, 3, 4, 5), 2) 表示建立獲得的 RDD 分區個數爲 2，在沒有指定分區個數的狀況下，Spark 會根據集羣部署模式，來肯定一個分區個數默認值。

對於 parallelize 方法，默認狀況下，分區的個數會受 Apache Spark 配置參數 spark.default.parallelism 的影響,不管是以本地模式、Standalone 模式、Yarn 模式或者是 Mesos 模式來運行 Apache Spark，分區的默認個數等於對 spark.default.parallelism 的指定值，若該值未設置，則 Apache Spark 會根據不一樣集羣模式的特徵，來肯定這個值。

本地模式，默認分區個數等於本地機器的 CPU 核心總數（或者是用戶經過 local[N] 參數指定分配給 Apache Spark 的核心數目),集羣模式（Standalone 或者 Yarn）默認分區個數等於集羣中全部核心數目的總和，或者 2，取二者中的較大值(conf.getInt("spark.default.parallelism", math.max(totalCoreCount.get(), 2)))

對於 textFile 方法，默認分區個數等於 min(defaultParallelism, 2)

2.1.3 分區內部記錄個數

分區分配的另外一個分配原則是：儘量使同一 RDD 不一樣分區內的記錄的數量一致。

對於轉換操做獲得的 RDD，若是是窄依賴，則分區記錄數量依賴於父 RDD 中相同編號分區是如何進行數據分配的，若是是 Shuffle 依賴，則分區記錄數量依賴於選擇的分區器，分區器有哈希分區和範圍分區。哈希分區器沒法保證數據被平均分配到各個分區，而範圍分區器則能作到這一點

對於textFile 方法分區內數據的大小則是由 Hadoop API 接口 FileInputFormat.getSplits 方法決定（見 HadoopRDD 類），獲得的每個分片即爲 RDD 的一個分區，分片內數據的大小會受文件大小、文件是否可分割、HDFS 中塊大小等因素的影響，但整體而言會是比較均衡的分配

2.2 RDD依賴

2.2.1 依賴與 RDD

RDD 的容錯機制是經過記錄更新來實現的，且記錄的是粗粒度的轉換操做。在外部，咱們將記錄的信息稱爲血統（Lineage）關係，而到了源碼級別，Apache Spark 記錄的則是 RDD 之間的依賴Dependency關係。在一次轉換操做中，建立獲得的新 RDD 稱爲子 RDD，提供數據的 RDD 稱爲父 RDD，父 RDD 可能會存在多個，咱們把子 RDD 與父 RDD 之間的關係稱爲依賴關係，或者能夠說是子 RDD 依賴於父 RDD。

依賴只保存父 RDD 信息，轉換操做的其餘信息，如數據處理函數，會在建立 RDD 時候，保存在新的 RDD 內。依賴在 Apache Spark 源碼中的對應實現是 Dependency 抽象類，每一個 Dependency 子類內部都會存儲一個 RDD 對象，對應一個父 RDD，若是一次轉換轉換操做有多個父 RDD，就會對應產生多個 Dependency 對象，全部的 Dependency 對象存儲在子 RDD 內部，經過遍歷 RDD 內部的 Dependency 對象，就能獲取該 RDD 全部依賴的父 RDD。

2.2.2 依賴分類

Apache Spark 將依賴進一步分爲兩類，分別是窄依賴Narrow Dependency和 Shuffle 依賴（Shuffle Dependency，在部分文獻中也被稱爲 Wide Dependency，即寬依賴。

窄依賴中，父 RDD 中的一個分區最多隻會被子 RDD 中的一個分區使用，換句話說，父 RDD 中，一個分區內的數據是不能被分割的，必須整個交付給子 RDD 中的一個分區。

窄依賴可進一步分類成一對一依賴和範圍依賴，對應實現分別是 OneToOneDependency 類和RangeDependency 類。一對一依賴表示子 RDD 分區的編號與父 RDD 分區的編號徹底一致的狀況，若兩個 RDD 之間存在着一對一依賴，則子 RDD 的分區個數、分區內記錄的個數都將繼承自父 RDD。範圍依賴是依賴關係中的一個特例，只被用於表示 UnionRDD 與父 RDD 之間的依賴關係。相比一對一依賴，除了第一個父 RDD，其餘父 RDD 和子 RDD 的分區編號再也不一致，Apache Spark 統一將unionRDD與父 RDD 之間（包含第一個 RDD）的關係都叫作範圍依賴。

依賴類圖：

graph TD
A[Dependency<br>依賴關係基類]--- B[NarrowDependency<br>窄依賴]
A---C[ShuffleDenpendency<br>shuffle依賴]
B---D[OneToOneDependency<br>一對一依賴]
B---E[RangeDependency<br>範圍依賴]

下圖展現了幾類常見的窄依賴及其對應的轉換操做。

Shuffle 依賴中，父 RDD 中的分區可能會被多個子 RDD 分區使用。由於父 RDD 中一個分區內的數據會被分割，發送給子 RDD 的全部分區，所以 Shuffle 依賴也意味着父 RDD 與子 RDD 之間存在着 Shuffle 過程。下圖展現了幾類常見的Shuffle依賴及其對應的轉換操做。

Shuffle 依賴的對應實現爲ShuffleDependency 類,其實現比較複雜，主要經過如下成員完成：

rdd：用於表示 Shuffle 依賴中，子 RDD 所依賴的父 RDD。
shuffleId：Shuffle 的 ID 編號，在一個 Spark 應用程序中，每一個 Shuffle 的編號都是惟一的。
shuffleHandle：Shuffle 句柄，ShuffleHandle 內部通常包含 Shuffle ID、Mapper 的個數以及對應的 Shuffle 依賴，在執行 ShuffleMapTask 時候，任務能夠經過 ShuffleManager 獲取獲得該句柄，並進一步獲得 Shuffle 相關信息。
partitioner：分區器，用於決定 Shuffle 過程當中 Reducer 的個數（其實是子 RDD 的分區個數）以及 Map 端的一條數據記錄應該分配給哪個 Reducer，也能夠被用在 CoGroupedRDD 中，肯定父 RDD 與子 RDD 之間的依賴關係類型。
serializer：序列化器。用於 Shuffle 過程當中 Map 端數據的序列化和 Reduce 端數據的反序列化。
KeyOrdering：鍵值排序策略，用於決定子 RDD 的一個分區內，如何根據鍵值對類型數據記錄進行排序。
Aggregator：聚合器，內部包含了多個聚合函數，比較重要的函數有 createCombiner：V => C，mergeValue: (C, V) => C 以及 mergeCombiners: (C, C) => C。例如，對於 groupByKey 操做，createCombiner 表示把第一個元素放入到集合中，mergeValue 表示一個元素添加到集合中，mergeCombiners 表示把兩個集合進行合併。這些函數被用於 Shuffle 過程當中數據的聚合。
mapSideCombine：用於指定 Shuffle 過程當中是否須要在 map 端進行 combine 操做。若是指定該值爲 true，因爲 combine 操做須要用到聚合器中的相關聚合函數，所以 Aggregator 不能爲空，不然 Apache Spark 會拋出異常。例如：groupByKey 轉換操做對應的ShuffleDependency中，mapSideCombine = false，而 reduceByKey 轉換操做中，mapSideCombine = true。

依賴關係是兩個 RDD 之間的依賴，所以若一次轉換操做中父 RDD 有多個，則可能會同時包含窄依賴和 Shuffle 依賴，下圖所示的 Join 操做，RDD a 和 RDD c 採用了相同的分區器，兩個 RDD 之間是窄依賴，Rdd b 的分區器與 RDD c 不一樣，所以它們之間是 Shuffle 依賴，具體實現可參見 CoGroupedRDD 類的 getDependencies 方法。這裏可以再次發現：一個依賴對應的是兩個 RDD，而不是一次轉換操做。

2.2.3 依賴與容錯機制

介紹完依賴的類別和實現以後，回過頭來，從分區的角度繼續探究 Apache Spark 是如何經過依賴關係來實現容錯機制的。下圖給出了一張依賴關係圖，fileRDD 經歷了 map、reduce 以及filter 三次轉換操做，獲得了最終的 RDD，其中，map、filter 操做對應的依賴爲窄依賴，reduce 操做對應的是 Shuffle 依賴。

假設最終 RDD 第一塊分區內的數據由於某些緣由丟失了，因爲 RDD 內的每個分區都會記錄其對應的父 RDD 分區的信息，所以沿着下圖所示的依賴關係往回走，咱們就能找到該分區數據最終來源於 fileRDD 的全部分區，再沿着依賴關係日後計算路徑中的每個分區數據，便可獲得丟失的分區數據。

這個例子並非特別嚴謹，按照咱們的思惟，只有執行了持久化，存儲在存儲介質中的 RDD 分區纔會出現數據丟失的狀況，可是上例中最終的 RDD 並無執行持久化操做。事實上，Apache Spark 將沒有被持久化數據從新被計算，以及持久化的數據第一次被計算，也等價視爲數據「丟失」，在 1.7 節中咱們會看到這一點。

2.2.4 依賴與並行計算

在上一節中咱們看到，在 RDD 中，能夠經過計算鏈Computing Chain來計算某個 RDD 分區內的數據，咱們也知道分區是並行計算的基本單位，這時候可能會有一種想法：可否把 RDD 每一個分區內數據的計算當成一個並行任務，每一個並行任務包含一個計算鏈，將一個計算鏈交付給一個 CPU 核心去執行，集羣中的 CPU 核心一塊兒把 RDD 內的全部分區計算出來。

答案是能夠，這得益於 RDD 內部分區的數據依賴相互之間並不會干擾，而 Apache Spark 也是這麼作的，但在實現過程當中，仍有不少實際問題須要去考慮。進一步觀察窄依賴、Shuffle 依賴在作並行計算時候的異同點。

先來看下方左側的依賴圖，依賴圖中全部的依賴關係都是窄依賴（包括一對一依賴和範圍依賴），能夠看到，不只計算鏈是獨立不干擾的（因此能夠並行計算），全部計算鏈內的每一個分區單元的計算工做也不會發生重複，如右側的圖所示。這意味着除非執行了持久化操做，不然計算過程當中產生的中間數據咱們沒有必要保留 —— 由於當前分區的數據只會給計算鏈中的下一個分區使用，而不用專門保留給其餘計算鏈使用。

再來觀察 Shuffle 依賴的計算鏈，如圖下方左側的圖中，既有窄依賴，又有 Shuffle 依賴，因爲 Shuffle 依賴中，子 RDD 一個分區的數據依賴於父 RDD 內全部分區的數據，當咱們想計算末 RDD 中一個分區的數據時，Shuffle 依賴處須要把父 RDD 全部分區的數據計算出來，如右側的圖所示（紫色表示最後兩個分區計算鏈共同通過的地方） —— 而這些數據，在計算末 RDD 另一個分區的數據時候，一樣會被用到。

若是咱們作到計算鏈的並行計算的話，這就意味着，要麼 Shuffle 依賴處父 RDD 的數據在每次須要使用的時候都重複計算一遍，要麼想辦法把父 RDD 數據保存起來，提供給其他分區的數據計算使用。

Apache Spark 採用的是第二種辦法，但保存數據的方法可能與想象中的會有所不一樣，Spark 把計算鏈從 Shuffle 依賴處斷開，劃分紅不一樣的階段Stage，階段之間存在依賴關係（其實就是 Shuffle 依賴），從而能夠構建一張不一樣階段之間的有向無環圖DAG。