[[174205]]

與許多專(zhuān)有的大數(shù)據(jù)處理平臺(tái)不同，Spark建立在統(tǒng)一抽象的RDD之上，使得它可以以基本一致的方式應(yīng)對(duì)不同的大數(shù)據(jù)處理場(chǎng)景，包括MapReduce，Streaming，SQL，Machine Learning以及Graph等。這即Matei Zaharia所謂的“設(shè)計(jì)一個(gè)通用的編程抽象(Unified Programming Abstraction)。這正是Spark這朵小火花讓人著迷的地方。要理解Spark，就需得理解RDD。

RDD是什么?

RDD，全稱(chēng)為Resilient Distributed Datasets，是一個(gè)容錯(cuò)的、并行的數(shù)據(jù)結(jié)構(gòu)，可以讓用戶(hù)顯式地將數(shù)據(jù)存儲(chǔ)到磁盤(pán)和內(nèi)存中，并能控制數(shù)據(jù)的分區(qū)。同時(shí)，RDD還提供了一組豐富的操作來(lái)操作這些數(shù)據(jù)。在這些操作中，諸如map、flatMap、filter等轉(zhuǎn)換操作實(shí)現(xiàn)了monad模式，很好地契合了Scala的集合操作。

除此之外，RDD還提供了諸如join、groupBy、reduceByKey等更為方便的操作(注意，reduceByKey是action，而非transformation)，以支持常見(jiàn)的數(shù)據(jù)運(yùn)算。 通常來(lái)講，針對(duì)數(shù)據(jù)處理有幾種常見(jiàn)模型，包括：Iterative Algorithms，Relational Queries，MapReduce，Stream Processing。例如Hadoop MapReduce采用了MapReduces模型，Storm則采用了Stream Processing模型。

RDD混合了這四種模型，使得Spark可以應(yīng)用于各種大數(shù)據(jù)處理場(chǎng)景。RDD作為數(shù)據(jù)結(jié)構(gòu)，本質(zhì)上是一個(gè)只讀的分區(qū)記錄集合。一個(gè)RDD可以包含多個(gè)分區(qū)，每個(gè)分區(qū)就是一個(gè)dataset片段。RDD可以相互依賴(lài)。 如果RDD的每個(gè)分區(qū)最多只能被一個(gè)Child RDD的一個(gè)分區(qū)使用，則稱(chēng)之為narrow dependency;若多個(gè)Child RDD分區(qū)都可以依賴(lài)，則稱(chēng)之為wide dependency。不同的操作依據(jù)其特性，可能會(huì)產(chǎn)生不同的依賴(lài)。

例如map操作會(huì)產(chǎn)生narrow dependency，而join操作則產(chǎn)生wide dependency。Spark之所以將依賴(lài)分為narrow與wide，基于兩點(diǎn)原因。 首先，narrow dependencies可以支持在同一個(gè)cluster node上以管道形式執(zhí)行多條命令，例如在執(zhí)行了map后，緊接著執(zhí)行filter。相反，wide dependencies需要所有的父分區(qū)都是可用的，可能還需要調(diào)用類(lèi)似MapReduce之類(lèi)的操作進(jìn)行跨節(jié)點(diǎn)傳遞。 其次，則是從失敗恢復(fù)的角度考慮。

narrow dependencies的失敗恢復(fù)更有效，因?yàn)樗恍枰匦掠?jì)算丟失的parent partition即可，而且可以并行地在不同節(jié)點(diǎn)進(jìn)行重計(jì)算。而wide dependencies牽涉到RDD各級(jí)的多個(gè)Parent Partitions。下圖說(shuō)明了narrow dependencies與wide dependencies之間的區(qū)別：

本圖來(lái)自Matei Zaharia撰寫(xiě)的論文An Architecture for Fast and General Data Processing on Large Clusters。圖中，一個(gè)box代表一個(gè)RDD，一個(gè)帶陰影的矩形框代表一個(gè)partition。RDD如何保障數(shù)據(jù)處理效率?RDD提供了兩方面的特性persistence和patitioning，用戶(hù)可以通過(guò)persist與patitionBy函數(shù)來(lái)控制RDD的這兩個(gè)方面。RDD的分區(qū)特性與并行計(jì)算能力(RDD定義了parallerize函數(shù))，使得Spark可以更好地利用可伸縮的硬件資源。若將分區(qū)與持久化二者結(jié)合起來(lái)，就能更加高效地處理海量數(shù)據(jù)。 例如：

partitionBy函數(shù)需要接受一個(gè)Partitioner對(duì)象，如：

RDD本質(zhì)上是一個(gè)內(nèi)存數(shù)據(jù)集，在訪問(wèn)RDD時(shí)，指針只會(huì)指向與操作相關(guān)的部分。例如存在一個(gè)面向列的數(shù)據(jù)結(jié)構(gòu)，其中一個(gè)實(shí)現(xiàn)為Int的數(shù)組，另一個(gè)實(shí)現(xiàn)為Float的數(shù)組。如果只需要訪問(wèn)Int字段，RDD的指針可以只訪問(wèn)Int數(shù)組，避免了對(duì)整個(gè)數(shù)據(jù)結(jié)構(gòu)的掃描。RDD將操作分為兩類(lèi)：transformation與action。無(wú)論執(zhí)行了多少次transformation操作，RDD都不會(huì)真正執(zhí)行運(yùn)算，只有當(dāng)action操作被執(zhí)行時(shí)，運(yùn)算才會(huì)觸發(fā)。

而在RDD的內(nèi)部實(shí)現(xiàn)機(jī)制中，底層接口則是基于迭代器的，從而使得數(shù)據(jù)訪問(wèn)變得更高效，也避免了大量中間結(jié)果對(duì)內(nèi)存的消耗。 在實(shí)現(xiàn)時(shí)，RDD針對(duì)transformation操作，都提供了對(duì)應(yīng)的繼承自RDD的類(lèi)型，例如map操作會(huì)返回MappedRDD，而flatMap則返回FlatMappedRDD。當(dāng)我們執(zhí)行map或flatMap操作時(shí)，不過(guò)是將當(dāng)前RDD對(duì)象傳遞給對(duì)應(yīng)的RDD對(duì)象而已。 例如：

這些繼承自RDD的類(lèi)都定義了compute函數(shù)。該函數(shù)會(huì)在action操作被調(diào)用時(shí)觸發(fā)，在函數(shù)內(nèi)部是通過(guò)迭代器進(jìn)行對(duì)應(yīng)的轉(zhuǎn)換操作：

RDD對(duì)容錯(cuò)的支持

支持容錯(cuò)通常采用兩種方式： 數(shù)據(jù)復(fù)制或日志記錄。對(duì)于以數(shù)據(jù)為中心的系統(tǒng)而言，這兩種方式都非常昂貴，因?yàn)樗枰缂壕W(wǎng)絡(luò)拷貝大量數(shù)據(jù)，畢竟帶寬的數(shù)據(jù)遠(yuǎn)遠(yuǎn)低于內(nèi)存。RDD天生是支持容錯(cuò)的。首先，它自身是一個(gè)不變的(immutable)數(shù)據(jù)集，其次，它能夠記住構(gòu)建它的操作圖(Graph of Operation)，因此當(dāng)執(zhí)行任務(wù)的Worker失敗時(shí)，完全可以通過(guò)操作圖獲得之前執(zhí)行的操作，進(jìn)行重新計(jì)算。

由于無(wú)需采用replication方式支持容錯(cuò)，很好地降低了跨網(wǎng)絡(luò)的數(shù)據(jù)傳輸成本。不過(guò)，在某些場(chǎng)景下，Spark也需要利用記錄日志的方式來(lái)支持容錯(cuò)。例如，在Spark Streaming中，針對(duì)數(shù)據(jù)進(jìn)行update操作，或者調(diào)用Streaming提供的window操作時(shí)，就需要恢復(fù)執(zhí)行過(guò)程的中間狀態(tài)。 此時(shí)，需要通過(guò)Spark提供的checkpoint機(jī)制，以支持操作能夠從checkpoint得到恢復(fù)。

針對(duì)RDD的wide dependency，最有效的容錯(cuò)方式同樣還是采用checkpoint機(jī)制。不過(guò)，似乎Spark的***版本仍然沒(méi)有引入auto checkpointing機(jī)制。總結(jié)RDD是Spark的核心，也是整個(gè)Spark的架構(gòu)基礎(chǔ)。 它的特性可以總結(jié)如下：

• 它是不變的數(shù)據(jù)結(jié)構(gòu)存儲(chǔ)
• 它是支持跨集群的分布式數(shù)據(jù)結(jié)構(gòu)
• 可以根據(jù)數(shù)據(jù)記錄的key對(duì)結(jié)構(gòu)進(jìn)行分區(qū)
• 提供了粗粒度的操作，且這些操作都支持分區(qū)
• 它將數(shù)據(jù)存儲(chǔ)在內(nèi)存中，從而提供了低延遲性