Spark分布式計算和RDD模型研究 · 內存計算

### 1背景介紹現今分布式計算框架像MapReduce和Dryad都提供了高層次的原語，使用戶不用操心任務分發和錯誤容忍，非常容易地編寫出并行計算程序。然而這些框架都缺乏對分布式內存的抽象和支持，使其在某些應用場景下不夠高效和強大。RDD(Resilient Distributed Datasets彈性分布式數據集)模型的產生動機主要來源于兩種主流的應用場景： ???迭代式算法：迭代式機器學習、圖算法，包括PageRank、K-means聚類和邏輯回歸(logistic regression) ???交互式數據挖掘工具：用戶在同一數據子集上運行多個Adhoc查詢。不難看出，這兩種場景的共同之處是：**在多個計算或計算的多個階段間，重用中間結果**。不幸的是，在目前框架如MapReduce中，要想在計算之間重用數據，唯一的辦法就是把數據保存到外部存儲系統中，例如分布式文件系統。這就導致了巨大的數據復制、磁盤I/O、序列化的開銷，甚至會占據整個應用執行時間的一大部分。為了解決這種問題，研究人員為有這種數據重用需要的應用開發了特殊的框架。例如將中間結果保存在內存中的迭代式圖計算框架Pregel。然而這些框架只支持一些特定的計算模式，而沒有提供一種通用的數據重用的抽象。于是，RDD橫空出世，它的主要功能有： ???高效的錯誤容忍 ???中間結果持久化到內存的并行數據結構 ???可控制數據分區來優化數據存儲 ???豐富的操作方法對于設計RDD來說，最大的挑戰在于**如何提供高效的錯誤容忍(fault-tolerance)**。現有的集群上的內存存儲抽象，如分布式共享內存、key-value存儲、內存數據庫以及Piccolo等，都提供了對可變狀態(如數據庫表里的Cell)的細粒度更新。在這種設計下為了容錯,就必須在集群結點間進行數據復制(data replicate)或者記錄日志。這兩種方法對于數據密集型的任務來說開銷都非常大，因為需要在結點間拷貝大量的數據，而網絡帶寬遠遠低于RAM。 ??????與這些框架不同，RDD提供**基于粗粒度轉換(coarse-grained transformation)的接口**,例如map、filter、join，能夠將同一操作施加到許多數據項上。于是通過記錄這些構建數據集(lineage世族)的粗粒度轉換的日志，而非實際數據，就能夠實現高效的容錯。當某個RDD丟失時，RDD有充足的關于丟失的那個RDD是如何從其他RDD產生的信息，從而通過重新計算來還原丟失的數據，避免了數據復制的高開銷。 ??????盡管基于粗粒度轉換的接口第一眼看起來有些受限、不夠強大，但實際上RDD卻能很好地用于許多并行計算應用，因為**這些應用本身自然而然地就是在多個數據項上運用相同的操作**。事實上，RDD能夠高效地表達許多框架的編程模型，如MapReduce、DryadLINQ、SQL、Pregel和HaLoop，以及它們處理不了的交互式數據挖掘應用。 ### 2 RDD簡介 ### 2.1概念 RDD是一種只讀的、分區的記錄集合。具體來說，RDD具有以下一些特點： ???創建：只能通過轉換(**transformation**，如map/filter/groupBy/join等，區別于動作action)從兩種數據源中創建RDD：1）穩定存儲中的數據；2）其他RDD。 ???只讀：狀態不可變，不能修改 ???分區：支持使RDD中的元素根據那個key來分區(**partitioning**)，保存到多個結點上。還原時只會重新計算丟失分區的數據，而不會影響整個系統。 ???路徑：在RDD中叫世族或血統(**lineage**)，即RDD有充足的信息關于它是如何從其他RDD產生而來的。 ???持久化：支持將會·被重用的RDD緩存(如in-memory或溢出到磁盤) ???延遲計算：像DryadLINQ一樣，Spark也會延遲計算RDD，使其能夠將轉換管道化(pipeline transformation) ???操作：豐富的動作(**action**)，count/reduce/collect/save等。關于轉換(transformation)與動作(action)的區別，前者會生成新的RDD，而后者只是將RDD上某項操作的結果返回給程序，而不會生成新的RDD： ![](https://box.kancloud.cn/2016-08-31_57c6b1347ec5b.jpg) ### 2.2例子假設網站中的一個WebService出現錯誤，我們想要從數以TB的HDFS日志文件中找到問題的原因，此時我們就可以用Spark加載日志文件到一組結點組成集群的RAM中，并交互式地進行查詢。以下是代碼示例： ![](https://box.kancloud.cn/2016-08-31_57c6b134a473d.jpg) ??????首先行1從HDFS文件中創建出一個RDD，而行2則衍生出一個經過某些條件過濾后的RDD。行3將這個RDD errors緩存到內存中，然而第一個RDD lines不會駐留在內存中。這樣做很有必要，因為errors可能非常小，足以全部裝進內存，而原始數據則會非常龐大。經過緩存后，現在就可以反復重用errors數據了。我們這里做了兩個操作，第一個是統計errors中包含MySQL字樣的總行數，第二個則是取出包含HDFS字樣的行的第三列時間，并保存成一個集合。 ![](https://box.kancloud.cn/2016-08-31_57c6b134ba3ab.jpg) ??????這里要注意的是前面曾經提到過的Spark的延遲處理。Spark調度器會將filter和map這兩個轉換保存到管道，然后一起發送給結點去計算。 ### 2.3優勢 RDD與DSM(distributed shared memory)的最大不同是：RDD只能通過粗粒度轉換來創建，而DSM則允許對每個內存位置上數據的讀和寫。在這種定義下，DSM不僅包括了傳統的共享內存系統，也包括了像提供了共享DHT(distributed hash table)的Piccolo以及分布式數據庫等。所以RDD相比DSM有著下面這些優勢： ???高效的容錯機制：沒有檢查點(checkpoint)開銷，能夠通過世族關系還原。而且還原只涉及了丟失數據分區的重計算，并且重算過程可以在不同結點并行進行，而無需回滾整個系統。 ???結點落后問題的緩和(mitigate straggler)：RDD的不可變性使得系統能夠運行類似MapReduce備份任務，來緩和慢結點。這在DSM系統中卻難以實現，因為多個相同任務一起運行會訪問同樣的內存數據而相互干擾。 ???批量操作：任務能夠根據數據本地性(data locality)被分配，從而提高性能。 ???優雅降級(degrade gracefully)：當內存不足時，大分區會被溢出到磁盤，提供與其他現今的數據并行計算系統類似的性能。 ### 2.4應用場景 RDD最適合那種在數據集上的所有元素都執行相同操作的批處理式應用。在這種情況下，RDD只需記錄世族圖譜中的每個轉換就能還原丟失的數據分區，而無需記錄大量的數據操作日志。所以**RDD不適合那些需要異步、細粒度更新狀態的應用**，比如Web應用的存儲系統，或增量式的Web爬蟲等。對于這些應用，使用具有事務更新日志和數據檢查點的數據庫系統更為高效。 ### 3 RDD表現形式 ### 3.1深入RDD 使用RDD作為抽象的一個挑戰就是：選擇一種合適的表現形式，來追蹤橫跨眾多轉換的RDD世族關系。在Spark中，我們使用一種簡單的、基于圖的表現形式，使得Spark在無需為每個轉換都增加特殊的處理邏輯的情況下，就能支持大量的轉換類型，這大大簡化了系統的設計。 ??????總的來說，對于每個RDD都包含五部分信息，即數據分區的集合，能根據本地性快速訪問到數據的偏好位置，依賴關系，計算方法，是否是哈希/范圍分區的元數據： ![](https://box.kancloud.cn/2016-08-31_57c6b134d1857.jpg) 以Spark中內建的幾個RDD舉例來說： <table border="1" cellpadding="0" cellspacing="0" style="color: rgb(0, 0, 0); orphans: 2; text-align: -webkit-auto; widows: 2; font-family: Simsun; border-collapse: collapse; border: none;"><tbody><tr><td style="width: 84.85pt; border: 1pt solid windowtext; background-color: rgb(91, 155, 213); padding: 0cm 5.4pt;" valign="top" width="113">信息/RDD</td><td style="width: 99.2pt; border-top-style: solid; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-top-color: windowtext; border-right-color: windowtext; border-bottom-color: windowtext; border-top-width: 1pt; border-right-width: 1pt; border-bottom-width: 1pt; background-color: rgb(91, 155, 213); padding: 0cm 5.4pt;" valign="top" width="132">HadoopRDD</td><td style="width: 106.3pt; border-top-style: solid; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-top-color: windowtext; border-right-color: windowtext; border-bottom-color: windowtext; border-top-width: 1pt; border-right-width: 1pt; border-bottom-width: 1pt; background-color: rgb(91, 155, 213); padding: 0cm 5.4pt;" valign="top" width="142">FilteredRDD</td><td style="width: 124.45pt; border-top-style: solid; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-top-color: windowtext; border-right-color: windowtext; border-bottom-color: windowtext; border-top-width: 1pt; border-right-width: 1pt; border-bottom-width: 1pt; background-color: rgb(91, 155, 213); padding: 0cm 5.4pt;" valign="top" width="166">JoinedRDD</td></tr><tr><td style="width: 84.85pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: solid; border-right-color: rgb(156, 194, 229); border-bottom-color: rgb(156, 194, 229); border-left-color: rgb(156, 194, 229); border-right-width: 1pt; border-bottom-width: 1pt; border-left-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="113">Partitions</td><td style="width: 99.2pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="132">每個HDFS塊一個分區，組成集合</td><td style="width: 106.3pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="142">與父RDD相同</td><td style="width: 124.45pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="166">每個Reduce任務一個分區</td></tr><tr><td style="width: 84.85pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: solid; border-right-color: rgb(156, 194, 229); border-bottom-color: rgb(156, 194, 229); border-left-color: rgb(156, 194, 229); border-right-width: 1pt; border-bottom-width: 1pt; border-left-width: 1pt; padding: 0cm 5.4pt;" valign="top" width="113">PreferredLoc</td><td style="width: 99.2pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; padding: 0cm 5.4pt;" valign="top" width="132">HDFS塊位置</td><td style="width: 106.3pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; padding: 0cm 5.4pt;" valign="top" width="142">無(或詢問父RDD)</td><td style="width: 124.45pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; padding: 0cm 5.4pt;" valign="top" width="166">無</td></tr><tr><td style="width: 84.85pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: solid; border-right-color: rgb(156, 194, 229); border-bottom-color: rgb(156, 194, 229); border-left-color: rgb(156, 194, 229); border-right-width: 1pt; border-bottom-width: 1pt; border-left-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="113">Dependencies</td><td style="width: 99.2pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="132">無(父RDD)</td><td style="width: 106.3pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="142">與父RDD一對一</td><td style="width: 124.45pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="166">對每個RDD進行混排</td></tr><tr><td style="width: 84.85pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: solid; border-right-color: rgb(156, 194, 229); border-bottom-color: rgb(156, 194, 229); border-left-color: rgb(156, 194, 229); border-right-width: 1pt; border-bottom-width: 1pt; border-left-width: 1pt; padding: 0cm 5.4pt;" valign="top" width="113">Iterator</td><td style="width: 99.2pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; padding: 0cm 5.4pt;" valign="top" width="132">讀取對應的塊數據</td><td style="width: 106.3pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; padding: 0cm 5.4pt;" valign="top" width="142">過濾</td><td style="width: 124.45pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; padding: 0cm 5.4pt;" valign="top" width="166">聯接混排的數據</td></tr><tr><td style="width: 84.85pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: solid; border-right-color: rgb(156, 194, 229); border-bottom-color: rgb(156, 194, 229); border-left-color: rgb(156, 194, 229); border-right-width: 1pt; border-bottom-width: 1pt; border-left-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="113">Partitioner</td><td style="width: 99.2pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="132">無</td><td style="width: 106.3pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="142">無</td><td style="width: 124.45pt; border-top-style: none; border-right-style: solid; border-bottom-style: solid; border-left-style: none; border-bottom-color: rgb(156, 194, 229); border-bottom-width: 1pt; border-right-color: rgb(156, 194, 229); border-right-width: 1pt; background-color: rgb(222, 234, 246); padding: 0cm 5.4pt;" valign="top" width="166">HashPartitioner</td></tr></tbody></table> ### 3.2工作原理在了解了RDD的概念和內部表現形式之后，那么RDD是如何運行的呢？總高層次來看，主要分為三步：創建RDD對象，DAG調度器創建執行計劃，Task調度器分配任務并調度Worker開始運行。 ![](https://box.kancloud.cn/2016-08-31_57c6b134e83d0.jpg) 以下面一個按A-Z首字母分類，查找相同首字母下不同姓名總個數的例子來看一下RDD是如何運行起來的。 ![](https://box.kancloud.cn/2016-08-31_57c6b135079e3.jpg) **步驟1：創建RDD。**上面的例子除去最后一個collect是個動作，不會創建RDD之外，前面四個轉換都會創建出新的RDD。因此第一步就是創建好所有RDD(內部的五項信息)。 **步驟2：創建執行計劃。**Spark會盡可能地管道化，并基于是否要重新組織數據來劃分**階段(stage)**，例如本例中的groupBy()轉換就會將整個執行計劃劃分成兩階段執行。最終會產生一個**DAG(directed acyclic graph，有向無環圖)**作為邏輯執行計劃。 ![](https://box.kancloud.cn/2016-08-31_57c6b1351c9e0.jpg) **步驟3：調度任務。**將各階段劃分成不同的**任務(task)**，每個任務都是數據和計算的合體。在進行下一階段前，當前階段的所有任務都要執行完成。因為下一階段的第一個轉換一定是重新組織數據的，所以必須等當前階段所有結果數據都計算出來了才能繼續。 ??????假設本例中的hdfs://names下有四個文件塊，那么HadoopRDD中partitions就會有四個分區對應這四個塊數據，同時preferedLocations會指明這四個塊的最佳位置。現在，就可以創建出四個任務，并調度到合適的集群結點上。 ![](https://box.kancloud.cn/2016-08-31_57c6b135304cf.jpg) ### 3.3混排 (待補充：關于混排(Shuffle)是如何執行的) ### 3.4寬窄依賴在設計RDD的接口時，一個有意思的問題是如何表現RDD之間的依賴。在RDD中將依賴劃分成了兩種類型：窄依賴(narrow dependencies)和寬依賴(wide dependencies)。窄依賴是指**父RDD的每個分區都只被子RDD的一個分區所使用**。相應的，那么寬依賴就是指父RDD的分區被多個子RDD的分區所依賴。例如，map就是一種窄依賴，而join則會導致寬依賴(除非父RDD是hash-partitioned，見下圖)。 ![](https://box.kancloud.cn/2016-08-31_57c6b13545828.jpg) ??????這種劃分有兩個用處。首先，窄依賴支持在一個結點上管道化執行。例如基于一對一的關系，可以在filter之后執行map。其次，窄依賴支持更高效的故障還原。因為對于窄依賴，只有丟失的父RDD的分區需要重新計算。而對于寬依賴，一個結點的故障可能導致來自所有父RDD的分區丟失，因此就需要完全重新執行。因此對于寬依賴，Spark會在持有各個父分區的結點上，將中間數據持久化來簡化故障還原，就像MapReduce會持久化map的輸出一樣。 ### 4內部實現 ### 4.1調度器 Spark的調度器類似于Dryad的，但是增加了對持久化RDD分區是否在內存里的考慮。重溫一下前面例子里介紹過的：調度器會根據RDD的族譜創建出分階段的DAG；每個階段都包含盡可能多的具有窄依賴的變換；具有寬依賴的混排操作是階段的邊界；調度器根據數據本地性分派任務到集群結點上。 ### 4.2解釋器集成 (待補充) ### 4.3內存管理 Spark支持三種內存管理方式：Java對象的內存存儲，序列化數據的內存存儲，磁盤存儲。第一種能提供最快的性能，因為JVM能夠直接訪問每個RDD對象。第二種使用戶在內存空間有限時，能選擇一種比Java對象圖更加高效的存儲方式。第三種則對大到無法放進內存，但每次重新計算又很耗時的RDD非常有用。同時，當有新的RDD分區被計算出來而內存空間又不足時，Spark使用LRU策略將老分區移除到磁盤上。 ### 4.4檢查點支持盡管RDD的Lineage可以用來還原數據，但這通常會非常耗時。所以將某些RDD持久化到磁盤上會非常有用，例如前面提到過的，寬依賴的中間數據。對于Spark來說，對檢查點的支持非常簡單，因為RDD都是不可變的。所以完全可以在后臺持久化RDD，而無需暫停整個系統。 ### 5高級特性 (待補充：Broadcast…) ### 6參考資料本文內容主要來源于：1）RDD論文《Resilient Distributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing》；2）Spark峰會ppt資料：《A-Deeper-Understanding-of-Spark-Internals》和《Introduction to Spark Internals》。感興趣的可以自行查找。