簡書 Wwwwei
轉(zhuǎn)載請注明原創(chuàng)出處，謝謝！

前言

??Spark是專為大規(guī)模數(shù)據(jù)處理而設(shè)計(jì)的快速通用的計(jì)算引擎，隨著大數(shù)據(jù)的興起，應(yīng)用場景越來越豐富。而RDD作為一種新型的內(nèi)存抽象模型，是Spark核心功能的基礎(chǔ)，理解RDD的思想和實(shí)現(xiàn)不僅僅有助于了解和使用Spark計(jì)算引擎，更有助于應(yīng)用其設(shè)計(jì)思想解決其他大數(shù)據(jù)問題。
??同之前系列一樣，本文不會過多的講如何使用RDD，更注重說清楚為什么需要RDD、什么是RDD以及如何實(shí)現(xiàn)RDD。

為什么需要RDD

傳統(tǒng)分布式計(jì)算框架的局限性

??分布式計(jì)算框架是針對大數(shù)據(jù)應(yīng)用場景的計(jì)算框架，以分布式的形式把巨大的計(jì)算任務(wù)分成小的單機(jī)可以承受的計(jì)算任務(wù)，解決常規(guī)單機(jī)計(jì)算模式無法支撐巨大數(shù)據(jù)量的問題。
??大多數(shù)傳統(tǒng)的分布式計(jì)算框架都是基于非循環(huán)的數(shù)據(jù)流模型，即從穩(wěn)定的物理存儲（如分布式文件系統(tǒng)）中加載記錄，在執(zhí)行計(jì)算任務(wù)時，當(dāng)一組確定性操作完成后，通常將中間結(jié)果寫回磁盤中穩(wěn)定存儲，每次查詢時再重新加載。
??盡管非循環(huán)數(shù)據(jù)流是一種很強(qiáng)大的抽象方法，但仍然有些應(yīng)用無法使用這種方式描述，例如機(jī)器學(xué)習(xí)和圖應(yīng)用中常用的迭代算法（每一步對數(shù)據(jù)執(zhí)行相似的函數(shù)）和交互式數(shù)據(jù)挖掘工具（用戶反復(fù)查詢一個數(shù)據(jù)子集），這類應(yīng)用需要在多個并行操作之間重用工作數(shù)據(jù)集，即中間結(jié)果。
??因此，基于數(shù)據(jù)流的架構(gòu)并不明確支持工作集。因?yàn)?strong>每次查詢中間結(jié)果會產(chǎn)生大量的I/O消耗，極大程度的影響了計(jì)算的性能和效率。

基于內(nèi)存的分布式計(jì)算構(gòu)想

??針對這個問題，需要提供一種新的分布式計(jì)算構(gòu)想，它要求既能夠保持傳統(tǒng)分布式計(jì)算框架，例如MapReduce及其相關(guān)模型的優(yōu)勢特性，即自動容錯、位置感知性調(diào)度和可伸縮性，同時夠?qū)诠ぷ骷挠?jì)算任務(wù)也具有良好的描述能力，即支持中間結(jié)果的復(fù)用場景。
??因此，在傳統(tǒng)的分布式計(jì)算框架基礎(chǔ)上，開發(fā)人員希望將計(jì)算的中間結(jié)果存儲由磁盤轉(zhuǎn)為內(nèi)存，提供一種在大型集群上執(zhí)行基于內(nèi)存的計(jì)算構(gòu)想，消除磁盤中加載中間結(jié)果帶來的I/O損耗，最大程度的提高計(jì)算性能。
??Apache Spark由此誕生，作為專為大規(guī)模數(shù)據(jù)處理而設(shè)計(jì)的快速通用的計(jì)算引擎，擁有Hadoop MapReduce所具有的優(yōu)點(diǎn)；但不同于MapReduce的是——Job中間輸出結(jié)果可以保存在內(nèi)存中，從而不再需要讀寫HDFS，因此Spark能更好地適用于數(shù)據(jù)挖掘與機(jī)器學(xué)習(xí)等需要迭代的MapReduce的算法。

什么是RDD

RDD的提出

??為了滿足Spark基于內(nèi)存的分布式計(jì)算思想，需要定義一種分布式內(nèi)存抽象，保證在分布式環(huán)境中能夠正確、可靠、高效地完成計(jì)算任務(wù)。
??如何定義這種分布式內(nèi)存抽象，需要考慮多方面的因素。首先，分布式內(nèi)存抽象需要具有傳統(tǒng)分布式計(jì)算框架的優(yōu)點(diǎn)，即自動容錯、位置感知性調(diào)度和可伸縮性。其次，為了提高迭代計(jì)算的性能和分布式并行計(jì)算下共享數(shù)據(jù)的容錯性，在設(shè)計(jì)上還需要符合基于內(nèi)存的分布式計(jì)算的實(shí)際需求，例如將數(shù)據(jù)集分區(qū)存儲在節(jié)點(diǎn)的內(nèi)存中，減少迭代過程（如機(jī)器學(xué)習(xí)算法）反復(fù)的I/O操作從而提高性能以及數(shù)據(jù)集不可變，并記錄其轉(zhuǎn)換過程，從而實(shí)現(xiàn)無共享數(shù)據(jù)讀寫同步問題、以及出錯的可重算性。
??基于以上原則，伯克利的設(shè)計(jì)者提出了RDD，即彈性分布式數(shù)據(jù)集的概念。

RDD的定義

??RDD是一種分布式的內(nèi)存抽象，稱為彈性分布式數(shù)據(jù)集（Resilient Distributed Datasets），是Apache Spark中數(shù)據(jù)的核心抽象，是一種只讀的、分區(qū)的數(shù)據(jù)記錄集合。RDD支持基于工作集的應(yīng)用，同時具有數(shù)據(jù)流模型的特點(diǎn)，即自動容錯、位置感知性調(diào)度和可伸縮性，允許用戶在執(zhí)行多個查詢時顯式地將工作集緩存在內(nèi)存中，極大地加速了后期的工作集重用。

怎樣實(shí)現(xiàn)RDD

RDD的邏輯物理結(jié)構(gòu)

??RDD是一種內(nèi)存抽象，可以認(rèn)為是一個分布式的數(shù)組，數(shù)組的元素是RDD的分區(qū)（Partition），分布在集群上。在物理數(shù)據(jù)存儲上，RDD的每一個分區(qū)對應(yīng)的就是一個數(shù)據(jù)塊（Block），數(shù)據(jù)塊可以存儲在內(nèi)存中，當(dāng)內(nèi)存不夠時可以存儲在磁盤上。RDD的邏輯物理結(jié)構(gòu)如下圖所示：

RDD的邏輯物理結(jié)構(gòu)

RDD的數(shù)據(jù)存儲結(jié)構(gòu)

??每個RDD的數(shù)據(jù)都以數(shù)據(jù)塊（Block）的形式存儲于多臺機(jī)器上，下圖是Spark的RDD存儲架構(gòu)原理圖：

RDD存儲架構(gòu)原理

??其中每個Executor會啟動一個BlockManagerSlave，并管理一部分Block；而Block的元數(shù)據(jù)由Driver節(jié)點(diǎn)的BlockManagerMaster保存。BlockManagerSlave生成Block后向BlockManagerMaster注冊該Block，BlockManagerMaster管理RDD與Block的關(guān)系，當(dāng)RDD不再需要存儲的時候，將向BlockManagerSlave發(fā)送指令刪除相應(yīng)的Block。

RDD是Spark的計(jì)算單元

??RDD是Spark的計(jì)算單元，算子是RDD中定義的函數(shù)，可以對RDD中的數(shù)據(jù)進(jìn)行轉(zhuǎn)換和操作。
??RDD有兩種操作算子：轉(zhuǎn)換（Transformation）與行動（Action），其中轉(zhuǎn)換操作是從一個RDD轉(zhuǎn)換創(chuàng)建一個新的RDD，這種操作是延遲計(jì)算的，也就是說不是馬上執(zhí)行，需要等到有Action操作的時候才會真正觸發(fā)運(yùn)算；行動操作對RDD操作后把結(jié)果返回給Driver，會觸發(fā)Spark提交作業(yè)（Job），并將數(shù)據(jù)輸出Spark系統(tǒng)。下圖展示了RDD算子的執(zhí)行過程：

RDD算子的執(zhí)行過程

??Spark從外部空間（HDFS）讀取數(shù)據(jù)形成RDD_0，Transformation算子對數(shù)據(jù)進(jìn)行操作（如filter）并轉(zhuǎn)化為新的RDD_1、RDD_2，通過Action算子（如collect/count）觸發(fā)Spark提交作業(yè)。
??如上的分析過程可以看出，Transformation算子并不會觸發(fā)Spark提交作業(yè)，直至Action算子才提交作業(yè)，這是一個延遲計(jì)算的設(shè)計(jì)技巧，可以避免內(nèi)存過快被中間計(jì)算占滿，從而提高內(nèi)存的利用率。

RDD是Spark的運(yùn)行邏輯的載體

??Spark采用了分布式計(jì)算中的Master-Slave模型。Master作為整個集群的控制器，負(fù)責(zé)整個集群的正常運(yùn)行；Worker是計(jì)算節(jié)點(diǎn)，接受主節(jié)點(diǎn)命令以及進(jìn)行狀態(tài)匯報；Executor是Worker節(jié)點(diǎn)行的一個進(jìn)程，負(fù)責(zé)任務(wù)（Task）的調(diào)度和執(zhí)行；Client作為用戶的客戶端負(fù)責(zé)提交應(yīng)用；Driver同樣也是Worker節(jié)點(diǎn)的一個進(jìn)程，負(fù)責(zé)控制一個應(yīng)用的執(zhí)行。Spark架構(gòu)如下圖所示：

Spark架構(gòu)

??從Spark的架構(gòu)角度來看，RDD是Spark的運(yùn)行邏輯的載體。一個Spark應(yīng)用的執(zhí)行過程可以分為五個步驟，如下圖所示：

Spark應(yīng)用的執(zhí)行過程

??（1）Spark集群啟動時，需要從主節(jié)點(diǎn)和從節(jié)點(diǎn)分別啟動Master進(jìn)程和Worker進(jìn)程，對整個集群進(jìn)行控制。
??（2）啟動Driver進(jìn)程。在一個Spark應(yīng)用的執(zhí)行過程中，Driver是應(yīng)用的邏輯執(zhí)行起點(diǎn)，運(yùn)行應(yīng)用的main函數(shù)并創(chuàng)建SparkContext。
??（3）構(gòu)建RDD的有向無環(huán)圖DAG（Directed Acyclic Graph）。在Spark應(yīng)用的執(zhí)行流程中，邏輯運(yùn)算會使用許多轉(zhuǎn)換操作，而每個轉(zhuǎn)換操作都會生成新的RDD，使得新的RDD和原有的RDD之間產(chǎn)生了依賴關(guān)系，行動操作觸發(fā)之后會將所有累積的轉(zhuǎn)換操作產(chǎn)生的RDD之間的依賴關(guān)系形成一個有向無環(huán)圖DAG。RDD的有向無環(huán)圖DAG，記錄了RDD的更新過程，當(dāng)這個RDD的部分分區(qū)數(shù)據(jù)丟失時，它可以通過DAG獲取足夠的信息來重新運(yùn)算和恢復(fù)丟失的數(shù)據(jù)分區(qū)。
??（4）根據(jù)RDD的有向無環(huán)圖DAG劃分階段（Stage）。SparkContext中的DAGScheduler把應(yīng)用程序中每個Job中的RDD的有向無環(huán)圖根據(jù)依賴關(guān)系劃分為多個階段，一個階段包含一系列函數(shù)進(jìn)行流水線執(zhí)行。RDD之間的依賴關(guān)系分為兩種，分別是窄依賴（Narrow Dependency）與寬依賴（Wide Dependency），其中Wide Dependency為子RDD的每個Partition都依賴于父RDD的所有Partition，而Narrow Dependency則只依賴一個或部分的Partition。下圖說明了窄依賴與寬依賴之間的區(qū)別：

窄依賴與寬依賴

??下圖是一個根據(jù)RDD的有向無環(huán)圖DAG劃分階段（Stage）的示例。圖中的A、B、C、D、E、F、G，分別代表不同的RDD，其中RDD內(nèi)的一個方框代表一個數(shù)據(jù)塊。數(shù)據(jù)從HDFS輸入Spark，形成RDD A和RDD C，RDD C上執(zhí)行map操作，轉(zhuǎn)換為RDD D，RDD B和RDD F進(jìn)行join操作轉(zhuǎn)換為G，而在B到G的過程中又會進(jìn)行Shuffle。最后RDD G通過函數(shù)saveAsSequenceFile輸出保存到HDFS中。

根據(jù)DAG劃分stage

??（5）根據(jù)階段（Stage）劃分任務(wù)（Task），調(diào)度任務(wù)進(jìn)行運(yùn)算。每一個Stage是一個任務(wù)集合（Task Set），TaskScheduler把Task分發(fā)給Worker中的Executor，Worker啟動Executor，Executor啟動線程池用于執(zhí)行Task。

總結(jié)

??由于數(shù)據(jù)爆炸性增長和大數(shù)據(jù)需求日益增加，傳統(tǒng)分布式計(jì)算框架存在局限性，不能夠很好的應(yīng)用于需要反復(fù)復(fù)用中間結(jié)果的非循環(huán)模型應(yīng)用?；趦?nèi)存的分布式計(jì)算框架Apache Spark因此誕生，作為Spark的核心和架構(gòu)基礎(chǔ)，RDD(Resilient Distributed Datasets),即彈性分布式數(shù)據(jù)集，在設(shè)計(jì)上結(jié)合了傳統(tǒng)分布式計(jì)算框架的優(yōu)點(diǎn)，即自動容錯、位置感知性調(diào)度和可伸縮性，以及為了符合基于內(nèi)存的分布式計(jì)算的實(shí)際需求，提高迭代計(jì)算的性能和分布式并行計(jì)算下共享數(shù)據(jù)的容錯性，應(yīng)用了有向無環(huán)圖DAG記錄RDD的依賴關(guān)系、高度受限的共享內(nèi)存機(jī)制、延時加載等理念。

??最后，特別感謝以下論文和博文，寫作時作為參考借鑒。
??[1] Zaharia M, Chowdhury M, Franklin M J, et al. Spark: cluster computing with working sets[C]// Usenix Conference on Hot Topics in Cloud Computing. USENIX Association, 2010:10-10.
??[2] Zaharia M, Chowdhury M, Das T, et al. Resilient distributed datasets: a fault-tolerant abstraction for in-memory cluster computing[C]// Usenix Conference on Networked Systems Design and Implementation. USENIX Association, 2012:2-2.
??[3] spark RDD的原理
??[4] 理解Spark的核心RDD

簡書 Wwwwei
轉(zhuǎn)載請注明原創(chuàng)出處，謝謝！