1、Spark介紹
Spark是起源于美國加州大學(xué)伯克利分校AMPLab的大數(shù)據(jù)計(jì)算平臺(tái),在2010年開源,目前是Apache軟件基金會(huì)的頂級(jí)項(xiàng)目。隨著 Spark在大數(shù)據(jù)計(jì)算領(lǐng)域的暫露頭角,越來越多的企業(yè)開始關(guān)注和使用。2014年11月,Spark在Daytona Gray Sort 100TB Benchmark競(jìng)賽中打破了由Hadoop MapReduce保持的排序記錄。Spark利用1/10的節(jié)點(diǎn)數(shù),把100TB數(shù)據(jù)的排序時(shí)間從72分鐘提高到了23分鐘。
Spark在架構(gòu)上包括內(nèi)核部分和4個(gè)官方子模塊--Spark SQL、Spark Streaming、機(jī)器學(xué)習(xí)庫MLlib和圖計(jì)算庫GraphX。圖1所示為Spark在伯克利的數(shù)據(jù)分析軟件棧BDAS(Berkeley Data Analytics Stack)中的位置。可見Spark專注于數(shù)據(jù)的計(jì)算,而數(shù)據(jù)的存儲(chǔ)在生產(chǎn)環(huán)境中往往還是由Hadoop分布式文件系統(tǒng)HDFS承擔(dān)。
圖1 Spark在BDAS中的位置
Spark被設(shè)計(jì)成支持多場(chǎng)景的通用大數(shù)據(jù)計(jì)算平臺(tái),它可以解決大數(shù)據(jù)計(jì)算中的批處理,交互查詢及流式計(jì)算等核心問題。Spark可以從多數(shù)據(jù)源的 讀取數(shù)據(jù),并且擁有不斷發(fā)展的機(jī)器學(xué)習(xí)庫和圖計(jì)算庫供開發(fā)者使用。數(shù)據(jù)和計(jì)算在Spark內(nèi)核及Spark的子模塊中是打通的,這就意味著Spark內(nèi)核 和子模塊之間成為一個(gè)整體。Spark的各個(gè)子模塊以Spark內(nèi)核為基礎(chǔ),進(jìn)一步支持更多的計(jì)算場(chǎng)景,例如使用Spark SQL讀入的數(shù)據(jù)可以作為機(jī)器學(xué)習(xí)庫MLlib的輸入。表1列舉了一些在Spark平臺(tái)上的計(jì)算場(chǎng)景。
表1 Spark的應(yīng)用場(chǎng)景舉例
在本文寫作是,Spark的最新版本為1.2.0,文中的示例代碼也來自于這個(gè)版本。
2、Spark內(nèi)核介紹
相信大數(shù)據(jù)工程師都非常了解Hadoop MapReduce一個(gè)最大的問題是在很多應(yīng)用場(chǎng)景中速度非常慢,只適合離線的計(jì)算任務(wù)。這是由于MapReduce需要將任務(wù)劃分成map和 reduce兩個(gè)階段,map階段產(chǎn)生的中間結(jié)果要寫回磁盤,而在這兩個(gè)階段之間需要進(jìn)行shuffle操作。Shuffle操作需要從網(wǎng)絡(luò)中的各個(gè)節(jié)點(diǎn) 進(jìn)行數(shù)據(jù)拷貝,使其往往成為最為耗時(shí)的步驟,這也是Hadoop MapReduce慢的根本原因之一,大量的時(shí)間耗費(fèi)在網(wǎng)絡(luò)磁盤IO中而不是用于計(jì)算。在一些特定的計(jì)算場(chǎng)景中,例如像邏輯回歸這樣的迭代式的計(jì) 算,MapReduce的弊端會(huì)顯得更加明顯。
那Spark是如果設(shè)計(jì)分布式計(jì)算的呢?首先我們需要理解Spark中最重要的概念--彈性分布數(shù)據(jù)集(Resilient Distributed Dataset),也就是RDD。
2.1 彈性分布數(shù)據(jù)集RDD
RDD是Spark中對(duì)數(shù)據(jù)和計(jì)算的抽象,是Spark中最核心的概念,它表示已被分片(partition),不可變的并能夠被并行操作的數(shù)據(jù)集 合。對(duì)RDD的操作分為兩種transformation和action。Transformation操作是通過轉(zhuǎn)換從一個(gè)或多個(gè)RDD生成新的 RDD。Action操作是從RDD生成最后的計(jì)算結(jié)果。在Spark最新的版本中,提供豐富的transformation和action操作,比起 MapReduce計(jì)算模型中僅有的兩種操作,會(huì)大大簡(jiǎn)化程序開發(fā)的難度。
RDD的生成方式只有兩種,一是從數(shù)據(jù)源讀入,另一種就是從其它RDD通過transformation操作轉(zhuǎn)換。一個(gè)典型的Spark程序就是通 過Spark上下文環(huán)境(SparkContext)生成一個(gè)或多個(gè)RDD,在這些RDD上通過一系列的transformation操作生成最終的 RDD,最后通過調(diào)用最終RDD的action方法輸出結(jié)果。
每個(gè)RDD都可以用下面5個(gè)特性來表示,其中后兩個(gè)為可選的:
分片列表(數(shù)據(jù)塊列表)計(jì)算每個(gè)分片的函數(shù)對(duì)父RDD的依賴列表對(duì)key-value類型的RDD的分片器(Partitioner)(可選)每個(gè)數(shù)據(jù)分片的預(yù)定義地址列表(如HDFS上的數(shù)據(jù)塊的地址)(可選)雖然Spark是基于內(nèi)存的計(jì)算,但RDD不光可以存儲(chǔ)在內(nèi)存中,根據(jù)useDisk、useMemory、useOffHeap, deserialized、replication五個(gè)參數(shù)的組合Spark提供了12種存儲(chǔ)級(jí)別,在后面介紹RDD的容錯(cuò)機(jī)制時(shí),我們會(huì)進(jìn)一步理解。值 得注意的是當(dāng)StorageLevel設(shè)置成OFF_HEAP時(shí),RDD實(shí)際被保存到Tachyon中。Tachyon是一個(gè)基于內(nèi)存的分布式文件系統(tǒng),目前正在快速發(fā)展,本文不做詳細(xì)介紹,可以通過其官方網(wǎng)站進(jìn)一步了解。
classStorageLevelprivate( privatevar _useDisk:Boolean, privatevar _useMemory:Boolean, privatevar _useOffHeap:Boolean, privatevar _deserialized:Boolean privatevar _replication:Int=1) extendsExternalizable{//… } val NONE =newStorageLevel(false,false,false,false) val DISK_ONLY =newStorageLevel(true,false,false,false) val DISK_ONLY_2 =newStorageLevel(true,false,false,false,2) val MEMORY_ONLY =newStorageLevel(false,true,false,true) val MEMORY_ONLY_2 =newStorageLevel(false,true,false,true,2) val MEMORY_ONLY_SER =newStorageLevel(false,true,false,false) val MEMORY_ONLY_SER_2 =newStorageLevel(false,true,false,false,2) val MEMORY_AND_DISK =newStorageLevel(true,true,false,true) val MEMORY_AND_DISK_2 =newStorageLevel(true,true,false,true,2) val MEMORY_AND_DISK_SER =newStorageLevel(true,true,false,false) val MEMORY_AND_DISK_SER_2 =newStorageLevel(true,true,false,false,2) val OFF_HEAP =newStorageLevel(false,false,true,false)2.2 DAG、Stage與任務(wù)的生成
Spark的計(jì)算發(fā)生在RDD的action操作,而對(duì)action之前的所有transformation,Spark只是記錄下RDD生成的軌跡,而不會(huì)觸發(fā)真正的計(jì)算。
Spark內(nèi)核會(huì)在需要計(jì)算發(fā)生的時(shí)刻繪制一張關(guān)于計(jì)算路徑的有向無環(huán)圖,也就是DAG。舉個(gè)例子,在圖2中,從輸入中邏輯上生成A和C兩個(gè) RDD,經(jīng)過一系列transformation操作,邏輯上生成了F,注意,我們說的是邏輯上,因?yàn)檫@時(shí)候計(jì)算沒有發(fā)生,Spark內(nèi)核做的事情只是記 錄了RDD的生成和依賴關(guān)系。當(dāng)F要進(jìn)行輸出時(shí),也就是F進(jìn)行了action操作,Spark會(huì)根據(jù)RDD的依賴生成DAG,并從起點(diǎn)開始真正的計(jì)算。
圖2 邏輯上的計(jì)算過程:DAG
有了計(jì)算的DAG圖,Spark內(nèi)核下一步的任務(wù)就是根據(jù)DAG圖將計(jì)算劃分成任務(wù)集,也就是Stage,這樣可以將任務(wù)提交到計(jì)算節(jié)點(diǎn)進(jìn)行真正的 計(jì)算。Spark計(jì)算的中間結(jié)果默認(rèn)是保存在內(nèi)存中的,Spark在劃分Stage的時(shí)候會(huì)充分考慮在分布式計(jì)算中可流水線計(jì)算(pipeline)的部 分來提高計(jì)算的效率,而在這個(gè)過程中,主要的根據(jù)就是RDD的依賴類型。根據(jù)不同的transformation操作,RDD的依賴可以分為窄依賴 (Narrow Dependency)和寬依賴(Wide Dependency,在代碼中為ShuffleDependency)兩種類型。窄依賴指的是生成的RDD中每個(gè)partition只依賴于父 RDD(s) 固定的partition。寬依賴指的是生成的RDD的每一個(gè)partition都依賴于父 RDD(s) 所有partition。窄依賴典型的操作有map, filter, union等,寬依賴典型的操作有g(shù)roupByKey, sortByKey等。可以看到,寬依賴往往意味著shuffle操作,這也是Spark劃分stage的主要邊界。對(duì)于窄依賴,Spark會(huì)將其盡量劃 分在同一個(gè)stage中,因?yàn)樗鼈兛梢赃M(jìn)行流水線計(jì)算。
圖3 RDD的寬依賴和窄依賴
我們?cè)偻ㄟ^圖4詳細(xì)解釋一下Spark中的Stage劃分。我們從HDFS中讀入數(shù)據(jù)生成3個(gè)不同的RDD,通過一系列 transformation操作后再將計(jì)算結(jié)果保存回HDFS。可以看到這幅DAG中只有join操作是一個(gè)寬依賴,Spark內(nèi)核會(huì)以此為邊界將其前 后劃分成不同的Stage. 同時(shí)我們可以注意到,在圖中Stage2中,從map到union都是窄依賴,這兩步操作可以形成一個(gè)流水線操作,通過map操作生成的 partition可以不用等待整個(gè)RDD計(jì)算結(jié)束,而是繼續(xù)進(jìn)行union操作,這樣大大提高了計(jì)算的效率。
圖4 Spark中的Stage劃分
Spark在運(yùn)行時(shí)會(huì)把Stage包裝成任務(wù)提交,有父Stage的Spark會(huì)先提交父Stage。弄清楚了Spark劃分計(jì)算的原理,我們?cè)俳Y(jié) 合源碼看一看這其中的過程。下面的代碼是DAGScheduler中的得到一個(gè)RDD父Stage的函數(shù),可以看到寬依賴為劃分Stage的邊界。
* Get or create the list of parent stages for a given RDD. The stages will be assigned the * provided jobId if they haven't already been created with a lower jobId. */ privatedef getParentStages(rdd: RDD[_], jobId:Int):List[Stage]={ val parents =newHashSet[Stage] val visited =newHashSet[RDD[_]] // We are manually maintaining a stack here to prevent StackOverflowError // caused by recursively visiting val waitingForVisit =newStack[RDD[_]] def visit(r: RDD[_]){ if(!visited(r)){ visited += r // Kind of ugly: need to register RDDs with the cache here since // we can't do it in its constructor because # of partitions is unknown for(dep <- r.dependencies){ dep match { case shufDep:ShuffleDependency[_, _, _]=> parents += getShuffleMapStage(shufDep, jobId) case _ => waitingForVisit.push(dep.rdd) } } } } waitingForVisit.push(rdd) while(!waitingForVisit.isEmpty){ visit(waitingForVisit.pop()) } parents.toList }上面提到Spark的計(jì)算是從RDD調(diào)用action操作時(shí)候觸發(fā)的,我們來看一個(gè)action的代碼
RDD的collect方法是一個(gè)action操作,作用是將RDD中的數(shù)據(jù)返回到一個(gè)數(shù)組中。可以看到,在此action中,會(huì)觸發(fā)Spark上下文環(huán)境SparkContext中的runJob方法,這是一系列計(jì)算的起點(diǎn)。
abstractclass RDD[T:ClassTag]( @transientprivatevar sc:SparkContext, @transientprivatevar deps:Seq[Dependency[_]] )extendsSerializablewithLogging{ //…./** * Return an array that contains all of the elements in this RDD. */ def collect():Array[T]={ val results = sc.runJob(this,(iter:Iterator[T])=> iter.toArray) Array.concat(results: _*) }}SparkContext擁有DAGScheduler的實(shí)例,在runJob方法中會(huì)進(jìn)一步調(diào)用DAGScheduler的runJob方法。在 此時(shí),DAGScheduler會(huì)生成DAG和Stage,將Stage提交給TaskScheduler。TaskSchduler將Stage包裝成 TaskSet,發(fā)送到Worker節(jié)點(diǎn)進(jìn)行真正的計(jì)算,同時(shí)還要監(jiān)測(cè)任務(wù)狀態(tài),重試失敗和長時(shí)間無返回的任務(wù)。整個(gè)過程如圖5所示。
圖5 Spark中任務(wù)的生成
2.3 RDD的緩存與容錯(cuò)
上文提到,Spark的計(jì)算是從action開始觸發(fā)的,如果在action操作之前邏輯上很多transformation操作,一旦中間發(fā)生計(jì) 算失敗,Spark會(huì)重新提交任務(wù),這在很多場(chǎng)景中代價(jià)過大。還有一些場(chǎng)景,如有些迭代算法,計(jì)算的中間結(jié)果會(huì)被重復(fù)使用,重復(fù)計(jì)算同樣增加計(jì)算時(shí)間和造 成資源浪費(fèi)。因此,在提高計(jì)算效率和更好支持容錯(cuò),Spark提供了基于RDDcache機(jī)制和checkpoint機(jī)制。
我們可以通過RDD的toDebugString來查看其遞歸的依賴信息,圖6展示了在spark shell中通過調(diào)用這個(gè)函數(shù)來查看wordCount RDD的依賴關(guān)系,也就是它的Lineage.
圖6 RDD wordCount的lineage
如果發(fā)現(xiàn)Lineage過長或者里面有被多次重復(fù)使用的RDD,我們就可以考慮使用cache機(jī)制或checkpoint機(jī)制了。
我們可以通過在程序中直接調(diào)用RDD的cache方法將其保存在內(nèi)存中,這樣這個(gè)RDD就可以被多個(gè)任務(wù)共享,避免重復(fù)計(jì)算。另外,RDD還提供了 更為靈活的persist方法,可以指定存儲(chǔ)級(jí)別。從源碼中可以看到RDD.cache就是簡(jiǎn)單的調(diào)用了 RDD.persist(StorageLevel.MEMORY_ONLY)。
/** Persist this RDD with the default storage level (`MEMORY_ONLY`). */ def persist():this.type = persist(StorageLevel.MEMORY_ONLY) def cache():this.type = persist()同樣,我們可以調(diào)用RDD的checkpoint方法將其保存到磁盤。我們需要在SparkContext中設(shè)置checkpoint的目錄,否則 調(diào)用會(huì)拋出異常。值得注意的是,在調(diào)用checkpoint之前建議先調(diào)用cache方法將RDD放入內(nèi)存,否則將RDD保存到文件的時(shí)候需要重新計(jì) 算。
/** * Mark this RDD for checkpointing. It will be saved to a file inside the checkpoint * directory set with SparkContext.setCheckpointDir() and all references to its parent * RDDs will be removed. This function must be called before any job has been * executed on this RDD. It is strongly recommended that this RDD is persisted in * memory, otherwise saving it on a file will require recomputation. */ def checkpoint(){ if(context.checkpointDir.isEmpty){ thrownewSparkException("Checkpoint directory has not been set in the SparkContext") }elseif(checkpointData.isEmpty){ checkpointData =Some(newRDDCheckpointData(this)) checkpointData.get.markForCheckpoint() } }Cache機(jī)制和checkpoint機(jī)制的差別在于cache將RDD保存到內(nèi)存,并保留Lineage,如果緩存失效RDD還可以通過Lineage重建。而checkpoint將RDD落地到磁盤并切斷Lineage,由文件系統(tǒng)保證其重建。
2.4 Spark任務(wù)的部署
Spark的集群部署分為Standalone、Mesos和Yarn三種模式,我們以Standalone模式為例,簡(jiǎn)單介紹Spark程序的部 署。如圖7示,集群中的Spark程序運(yùn)行時(shí)分為3種角色,driver, master和worker(slave)。在集群?jiǎn)?dòng)前,首先要配置master和worker節(jié)點(diǎn)。啟動(dòng)集群后,worker節(jié)點(diǎn)會(huì)向master節(jié) 點(diǎn)注冊(cè)自己,master節(jié)點(diǎn)會(huì)維護(hù)worker節(jié)點(diǎn)的心跳。Spark程序都需要先創(chuàng)建Spark上下文環(huán)境,也就是SparkContext。創(chuàng)建 SparkContext的進(jìn)程就成為了driver角色,上一節(jié)提到的DAGScheduler和TaskScheduler都在driver中運(yùn)行。 Spark程序在提交時(shí)要指定master的地址,這樣可以在程序啟動(dòng)時(shí)向master申請(qǐng)worker的計(jì)算資源。Driver,master和 worker之間的通信由Akka支持。Akka 也使用 Scala 編寫,用于構(gòu)建可容錯(cuò)的、高可伸縮性的Actor 模型應(yīng)用。關(guān)于Akka,可以訪問其官方網(wǎng)站進(jìn)行進(jìn)一步了解,本文不做詳細(xì)介紹。
3、更深一步了解Spark內(nèi)核
了解了Spark內(nèi)核的基本概念和實(shí)現(xiàn)后,更深一步理解其工作原理的最好方法就是閱讀源碼。最新的Spark源碼可以從Spark官方網(wǎng)站下載。 源碼推薦使用IntelliJ IDEA閱讀,會(huì)自動(dòng)安裝Scala插件。讀者可以從core工程,也就是Spark內(nèi)核工程開始閱讀,更可以設(shè)置斷點(diǎn)嘗試跟蹤一個(gè)任務(wù)的執(zhí)行。另外,讀 者還可以通過分析Spark的日志來進(jìn)一步理解Spark的運(yùn)行機(jī)制,Spark使用log4j記錄日志,可以在啟動(dòng)集群前修改log4j的配置文件來配 置日志輸出和格式。