Spark 中 RDD的執行機制

2020-06-16 16:35:18

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 的典型的執行過程如下：

讀入外部的資料來源（或者記憶體中的集合）進行 RDD 建立；
RDD 經過一系列的 “轉換” 操作，每一次都會產生不同的 RDD，供給下一個轉換使用；
最後一個 RDD 經過 “行動” 操作進行處理，並輸出指定的資料型別和值。

RDD 採用了惰性呼叫，即在 RDD 的執行過程中，所有的轉換操作都不會執行真正的操作，只會記錄依賴關係，而只有遇到了行動操作，才會觸發真正的計算，並根據之前的依賴關係得到最終的結果。

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

1.2. RDD 特性

總體而言，Spark 採用 RDD 以後能夠實現高效計算的主要原因如下：

高效的容錯性。在 RDD 的設計中，只能通過從父 RDD 轉換到子 RDD 的方式來修改資料，這也就是說我們可以直接利用 RDD 之間的依賴關係來重新計算得到丟失的分割區，而不需要通過資料冗餘的方式。而且也不需要記錄具體的資料和各種細粒度操作的紀錄檔，這大大降低了資料密集型應用中的容錯開銷。
中間結果持久化到記憶體。資料在記憶體中的多個 RDD 操作之間進行傳遞，不需要在磁碟上進行儲存和讀取，避免了不必要的讀寫磁碟開銷；
存放的資料可以是 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 架構中的執行過程（如下圖所示）：

建立 RDD 物件；
SparkContext 負責計算 RDD 之間的依賴關係，構建 DAG；
DAGSchedule 負責把 DAG 圖反向解析成多個階段，每個階段中包含多個任務，每個任務會被任務排程器分發給工作節點上的 Executor 上執行。