何煒杰

評審人：韓晶晶 徐江河

1、Spark 簡介

? Spark 是一種用于大規(guī)模數(shù)據(jù)處理的統(tǒng)一計(jì)算引擎。它是加州大學(xué)伯克利分校AMP 實(shí)驗(yàn)室所開發(fā)，后又成為Apache 頂級項(xiàng)目。圍繞著Spark 還推出了Spark SQL、Spark Streaming、MLlib 和GraphX 等組件。

? Spark使用Scala語言實(shí)現(xiàn)，它是一種面向?qū)ο蟮暮瘮?shù)式編程語言，能夠像操作本地集合對象一樣輕松地操作分布式數(shù)據(jù)集。

2、 Spark 特點(diǎn)

2.1 運(yùn)行速度快

Spark在批數(shù)據(jù)處理和流式數(shù)據(jù)處理方面，都有很強(qiáng)勁的表現(xiàn)，因?yàn)樗褂昧俗钕冗M(jìn)的DAG（有向無環(huán)圖）調(diào)度器、查詢優(yōu)化器和物理執(zhí)行引擎。

官方數(shù)據(jù)表明：在邏輯回歸算法中，Spark的處理速度能達(dá)到Hadoop的100多倍，這最主要是因?yàn)镾park基于內(nèi)存計(jì)算和引入了DAG調(diào)度器。

2.2. 易用性好

Spark提供了超過80個(gè)高級的算子，這使得構(gòu)建并行應(yīng)用程序變得很容易。Spark支持多種語言進(jìn)行開發(fā)，包括Scala、Java、Python等。尤其是用Scala或Python進(jìn)行開發(fā)時(shí)，能用簡潔的代碼實(shí)現(xiàn)較為復(fù)雜的功能。

2.3 通用性強(qiáng)

Spark 提供了一系列的組件，其中Spark Core 提供內(nèi)存計(jì)算框架、Spark SQL用于處理結(jié)構(gòu)化數(shù)據(jù)、SparkStreaming用于實(shí)時(shí)流計(jì)算、Graphx用于圖處理、MLlib用于機(jī)器學(xué)習(xí)。你能夠在同一個(gè)應(yīng)用程序中，組合使用這些組件，進(jìn)行一站式處理。

2.4 隨處運(yùn)行

Spark能運(yùn)行在Hadoop, Apache Mesos, Kubernetes, standalone或云端。它也能夠連接各種各樣的數(shù)據(jù)源，例如HDFS、HBase、Hive。

3、Spark 編程模型

3.1 RDD簡介

Spark 提供的最主要的編程模型是彈性分布式數(shù)據(jù)集RDD（resilient distributed dataset），這是一種不可變的、分區(qū)的、可以并行計(jì)算的元素集合。你可以通過Hadoop文件系統(tǒng)上的文件或是已經(jīng)存在的Scala集合來創(chuàng)建一個(gè)RDD，也可以通過已有的RDD來構(gòu)造一個(gè)新的RDD。用戶可以在內(nèi)存中緩存RDD，這樣可以使得它在并行計(jì)算中有效地重用。此外，RDD還能自動(dòng)從節(jié)點(diǎn)故障中恢復(fù)。

3.2 RDD五大特性

3.2.1 分區(qū)列表

? RDD被劃分為很多分區(qū)分布到集群的頂點(diǎn)中，每一個(gè)分區(qū)都會(huì)被一個(gè)計(jì)算任務(wù)TASK處理，分區(qū)的多少?zèng)Q定了對這個(gè)RDD進(jìn)行并行計(jì)算的粒度。

3.2.2 每個(gè)分區(qū)都有一個(gè)計(jì)算函數(shù)

? Spark的RDD的計(jì)算函數(shù)是以分區(qū)為基本單位的，每個(gè)RDD都會(huì)實(shí)現(xiàn) compute函數(shù)，對具體的分區(qū)進(jìn)行計(jì)算。

3.2.3 依賴于其他RDD

? RDD每次轉(zhuǎn)換都會(huì)生成新的RDD，所以RDD之間會(huì)有前后依賴關(guān)系。正是因?yàn)橛辛饲昂蟮囊蕾囮P(guān)系，所以當(dāng)有分區(qū)的數(shù)據(jù)丟失時(shí)，Spark 可以通過依賴關(guān)系對該分區(qū)進(jìn)行重新計(jì)算，從而得出丟失的數(shù)據(jù)。而不必對所有分區(qū)都進(jìn)行重新計(jì)算。

3.2.4 鍵值對類型的RDD有一個(gè)分區(qū)器

? 分區(qū)器決定了RDD如何分區(qū)，可以傳入相關(guān)的參數(shù)。

3.2.5 每個(gè)分區(qū)都有一個(gè)優(yōu)先位置列表

? 在Spark形成任務(wù)有向無環(huán)圖的時(shí)候，會(huì)盡可能地把計(jì)算分配到靠近數(shù)據(jù)的位置，以減少數(shù)據(jù)網(wǎng)絡(luò)傳輸。例如Hadoop 分區(qū)的首選位置就是HDFS塊所在的節(jié)點(diǎn)。

3.3 RDD依賴

3.3.1 窄依賴

? 窄依賴表示每一個(gè)父RDD 中的分區(qū)最多被子RDD 的一個(gè)分區(qū)所使用。常見的窄依賴算子有：map、filter、union等。

[圖片上傳失敗...(image-1d29c7-1542849324248)]

<center>圖3-1 RDD窄依賴</center>

3.3.2 寬依賴

? 寬依賴表示子RDD分區(qū)依賴于父RDD的所有分區(qū)。相對于窄依賴，寬依賴付出的代價(jià)要高很多，應(yīng)該盡量少使用。常見的寬依賴算子有：groupByKey、reduceByKey等。
? [圖片上傳失敗...(image-f87aa7-1542849324248)]

<center>圖3-2 RDD寬依賴</center>

4、Spark 核心原理

4.1 消息通信原理

4.1.1 Spark啟動(dòng)消息通信

? Spark啟動(dòng)過程中主要是進(jìn)行Master和Worker之間的通信，其消息發(fā)送關(guān)系如圖4-1所示。首先由Worker節(jié)點(diǎn)向Master節(jié)點(diǎn)發(fā)送注冊消息。Master處理完畢后，返回注冊成功或失敗的消息。如果注冊成功，則Worker定時(shí)發(fā)送心跳信息給Master。

avatar

<center>圖4-1 Spark啟動(dòng)消息通信交互過程</center>

4.1.2 Spark運(yùn)行時(shí)消息通信

? 圖4-2 展示了Spark運(yùn)行時(shí)消息通信交互過程。用戶提交應(yīng)用程序時(shí)，應(yīng)用程序的SparkContext會(huì)向Master發(fā)送應(yīng)用注冊消息，并由Master給該應(yīng)用分配Executor，Executor啟動(dòng)后會(huì)向SparkContext發(fā)送注冊成功消息。接著，TaskScheduler 會(huì)向注冊的Executor發(fā)送執(zhí)行消息。Executor接收到任務(wù)消息后啟動(dòng)并運(yùn)行。最后當(dāng)所有任務(wù)完成時(shí)，由Driver處理結(jié)果并回收資源。[圖片上傳失敗...(image-7e0dd5-1542849324248)]

<center>圖4-2 Spark運(yùn)行時(shí)消息通信交互過程</center>

4.2 作業(yè)執(zhí)行原理

? Spark的作業(yè)調(diào)度主要是指：基于RDD的一系列操作構(gòu)成一個(gè)作業(yè)，然后在Executor中執(zhí)行。而對RDD的操作主要分為轉(zhuǎn)換操作(Transformation)和行動(dòng)操作(Action)。對于轉(zhuǎn)換操作的計(jì)算是lazy級別的，也就是說它只會(huì)記錄轉(zhuǎn)換關(guān)系，并不會(huì)立即觸發(fā)計(jì)算。只有出現(xiàn)了行動(dòng)操作才會(huì)真正觸發(fā)計(jì)算。在Spark作業(yè)中最重要的是DAGScheduler 和 TaskScheduler 兩個(gè)調(diào)度器。其中DAGScheduler 負(fù)責(zé)任務(wù)的邏輯調(diào)度，而TaskScheduler 負(fù)責(zé)具體任務(wù)的調(diào)度執(zhí)行。下面通過圖4-1對Spark的作業(yè)和任務(wù)調(diào)度系統(tǒng)進(jìn)行具體介紹。

[圖片上傳失敗...(image-1cb0c2-1542849324248)]
<center>圖4-3 Spark的作業(yè)和任務(wù)調(diào)度系統(tǒng)</center>

（1）Spark應(yīng)用程序進(jìn)行各種轉(zhuǎn)換操作，通過行動(dòng)操作觸發(fā)作業(yè)運(yùn)行。作業(yè)提交之后根據(jù)RDD之間的依賴關(guān)系構(gòu)建DAG圖，DAG圖提交給DAGScheduler 解析。

（2）DAGScheduler把 DAG拆分為互相依賴的調(diào)度階段（Stage）。拆分的原則是：根據(jù)RDD的依賴是否為寬依賴，當(dāng)遇到寬依賴就拆分為一個(gè)新的調(diào)度階段。每個(gè)階段包含的任務(wù)構(gòu)成了任務(wù)集，DAGScheduler 會(huì)把這些任務(wù)集提交給TaskScheduler 進(jìn)行調(diào)度。

（3）TaskScheduler 收到任務(wù)集后以任務(wù)的形式一個(gè)個(gè)分發(fā)到集群Worker節(jié)點(diǎn)的Executor中去運(yùn)行。如果某個(gè)任務(wù)運(yùn)行失敗，TaskScheduler 會(huì)將其重新提交。如果某個(gè)任務(wù)運(yùn)行的很慢，TaskScheduler 會(huì)再啟動(dòng)一個(gè)同樣的任務(wù)，哪個(gè)任務(wù)先執(zhí)行完就用哪個(gè)任務(wù)的結(jié)果。

（4）Worker節(jié)點(diǎn)中的Executor收到任務(wù)后，以多線程的方式運(yùn)行，每個(gè)線程負(fù)責(zé)一個(gè)任務(wù)。

參考文獻(xiàn)

1. 郭景瞻 《圖解Spark：核心技術(shù)與案例實(shí)戰(zhàn)》電子工業(yè)出版社 2017-1
2. 王家林 ， 段智華 《Spark內(nèi)核機(jī)制解析及性能調(diào)優(yōu)》機(jī)械工業(yè)出版社 2017-1
3. spark官網(wǎng) http://spark.apache.org/