Spark 中 RDD的運行機制

1. RDD 的設計與運行原理

Spark 的核心是創建在統一的抽象 RDD 之上，基於 RDD 的轉換和行動操做使得 Spark 的各個組件能夠無縫進行集成，從而在同一個應用程序中完成大數據計算任務。

在實際應用中，存在許多迭代式算法和交互式數據挖掘工具，這些應用場景的共同之處在於不一樣計算階段之間會重用中間結果，即一個階段的輸出結果會做爲下一個階段的輸入。而 Hadoop 中的 MapReduce 框架都是把中間結果寫入到 HDFS 中，帶來了大量的數據複製、磁盤 IO 和序列化開銷，而且一般只支持一些特定的計算模式。而 RDD 提供了一個抽象的數據架構，從而讓開發者沒必要擔憂底層數據的分佈式特性，只需將具體的應用邏輯表達爲一系列轉換處理，不一樣 RDD 之間的轉換操做造成依賴關係，能夠實現管道化，從而避免了中間結果的存儲，大大下降了數據複製、磁盤 IO 和序列化開銷。

1.1. RDD 概念

一個 RDD 就是一個分佈式對象集合，提供了一種高度受限的共享內存模型，其本質上是一個只讀的分區記錄集合，不能直接修改。每一個 RDD 能夠分紅多個分區，每一個分區就是一個數據集片斷，而且一個 RDD 的不一樣分區能夠保存到集羣中不一樣的節點上，從而能夠在集羣中的不一樣節點上進行並行計算。

RDD 提供了一組豐富的操做以支持常見的數據運算，分爲「行動」（Action）和「轉換」（Transformation）兩種類型，前者用於執行計算並指定輸出的形式，後者指定 RDD 之間的相互依賴關係。RDD 提供的轉換接口都很是簡單，都是相似 map 、filter 、groupBy 、join 等粗粒度的數據轉換操做，而不是針對某個數據項的細粒度修改。所以，RDD 比較適合對於數據集中元素執行相同操做的批處理式應用，而不適合用於須要異步、細粒度狀態的應用，好比 Web 應用系統、增量式的網頁爬蟲等。

RDD 的典型的執行過程以下：

1. 讀入外部的數據源（或者內存中的集合）進行 RDD 建立；
2. RDD 通過一系列的 「轉換」 操做，每一次都會產生不一樣的 RDD，供給下一個轉換使用；
3. 最後一個 RDD 通過 「行動」 操做進行處理，並輸出指定的數據類型和值。

RDD 採用了惰性調用，即在 RDD 的執行過程當中，全部的轉換操做都不會執行真正的操做，只會記錄依賴關係，而只有遇到了行動操做，纔會觸發真正的計算，並根據以前的依賴關係獲得最終的結果。

下面以一個實例來描述 RDD 的實際執行過程，以下圖所示，開始從輸入中建立了兩個 RDD，分別是 A 和 C，而後通過一系列的轉換操做，最終生成了一個 F，這也是一個 RDD。注意，這些轉換操做的執行過程當中並無執行真正的計算，基於建立的過程也沒有執行真正的計算，而只是記錄的數據流向軌跡。當 F 執行了行爲操做並生成輸出數據時，Spark 纔會根據 RDD 的依賴關係生成有向無環圖（DAG），並從起點開始執行真正的計算。正是 RDD 的這種惰性調用機制，使得轉換操做獲得的中間結果不須要保存，而是直接管道式的流入到下一個操做進行處理。

1.2. RDD 特性

整體而言，Spark 採用 RDD 之後可以實現高效計算的主要緣由以下：

1. 高效的容錯性。在 RDD 的設計中，只能經過從父 RDD 轉換到子 RDD 的方式來修改數據，這也就是說咱們能夠直接利用 RDD 之間的依賴關係來從新計算獲得丟失的分區，而不須要經過數據冗餘的方式。並且也不須要記錄具體的數據和各類細粒度操做的日誌，這大大下降了數據密集型應用中的容錯開銷。

2. 中間結果持久化到內存。數據在內存中的多個 RDD 操做之間進行傳遞，不須要在磁盤上進行存儲和讀取，避免了沒必要要的讀寫磁盤開銷；

3. 存放的數據能夠是 Java 對象，避免了沒必要要的對象序列化和反序列化開銷。

1.3. RDD 之間的依賴關係

RDD 中的不一樣的操做會使得不一樣 RDD 中的分區會產生不一樣的依賴關係，主要分爲窄依賴（Narrow Dependency）與寬依賴（Wide Dependency）。其中，窄依賴表示的是父 RDD 和子 RDD 之間的一對一關係或者多對一關係，主要包括的操做有 map、filter、union 等；而寬依賴則表示父 RDD 與子 RDD 之間的一對多關係，即一個父 RDD 轉換成多個子 RDD，主要包括的操做有 groupByKey、sortByKey 等。

對於窄依賴的 RDD，能夠以流水線的方式計算全部父分區，不會形成網絡之間的數據混合。對於寬依賴的 RDD，則一般伴隨着 Shuffle 操做，即首先須要計算好全部父分區數據，而後在節點之間進行 Shuffle。所以，在進行數據恢復時，窄依賴只須要根據父 RDD 分區從新計算丟失的分區便可，並且能夠並行地在不一樣節點進行從新計算。而對於寬依賴而言，單個節點失效一般意味着從新計算過程會涉及多個父 RDD 分區，開銷較大。此外，Spark 還提供了數據檢查點和記錄日誌，用於持久化中間 RDD，從而使得在進行失敗恢復時不須要追溯到最開始的階段。在進行故障恢復時，Spark 會對數據檢查點開銷和從新計算 RDD 分區的開銷進行比較，從而自動選擇最優的恢復策略。

1.4. 階段的劃分

Spark 經過分析各個 RDD 的依賴關係生成了 DAG ，再經過分析各個 RDD 中的分區之間的依賴關係來決定如何劃分階段，具體劃分方法是：在 DAG 中進行反向解析，遇到寬依賴就斷開，遇到窄依賴就把當前的 RDD 加入到當前的階段中；將窄依賴儘可能劃分在同一個階段中，能夠實現流水線計算。例如在下圖中，首先根據數據的讀取、轉化和行爲等操做生成 DAG。而後在執行行爲操做時，反向解析 DAG，因爲從 A 到 B 的轉換和從 B、F 到 G 的轉換都屬於寬依賴，則須要從在寬依賴處進行斷開，從而劃分爲三個階段。把一個 DAG 圖劃分紅多個 「階段」 之後，每一個階段都表明了一組關聯的、相互之間沒有 Shuffle 依賴關係的任務組成的任務集合。每一個任務集合會被提交給任務調度器（TaskScheduler）進行處理，由任務調度器將任務分發給 Executor 運行。

1.5. RDD 運行過程

經過上述對 RDD 概念、依賴關係和階段劃分的介紹，結合以前介紹的 Spark 運行基本流程，這裏再總結一下 RDD 在 Spark 架構中的運行過程（以下圖所示）：

1. 建立 RDD 對象；
2. SparkContext 負責計算 RDD 之間的依賴關係，構建 DAG；
3. DAGSchedule 負責把 DAG 圖反向解析成多個階段，每一個階段中包含多個任務，每一個任務會被任務調度器分發給工做節點上的 Executor 上執行。