1 簡(jiǎn)介

Apache Spark 是專為大規(guī)模數(shù)據(jù)處理而設(shè)計(jì)的快速通用的計(jì)算引擎。它產(chǎn)生于 UC Berkeley AMP Lab，繼承了 MapReduce 的優(yōu)點(diǎn)，但是不同于 MapReduce 的是，Spark 可以將結(jié)果保存在內(nèi)存中，一直迭代計(jì)算下去，除非遇到 shuffle 。因此 Spark 能更好的適用于數(shù)據(jù)挖掘與機(jī)器學(xué)習(xí)等要迭代的算法。值得注意的是，官網(wǎng)說(shuō)的 Spark 是 MR 計(jì)算速度的 100 倍。僅僅適用于邏輯回歸等這樣的迭代計(jì)算。

​

​

2 Spark 的運(yùn)行模式

• Local 模式：多用于本機(jī)編寫、測(cè)試代碼。
• Standalone 模式：這是 Spark 自帶的資源調(diào)度框架，它支持完全分布式。
• Yarn 模式：這是 hadoop 里面的一個(gè)資源調(diào)度框架，Spark 同樣也可以使用。
• Mesos 模式：為應(yīng)用程序(如Hadoop、Spark、Kafka、ElasticSearch)提供API的整個(gè)數(shù)據(jù)中心和云環(huán)境中的資源管理和調(diào)度。

下面分別介紹一下 Standalone 和 Yarn 模式下任務(wù)流程。

Standalone-client 提交方式

提交命令如下：以官方給的計(jì)算 PI 的代碼為例。

./spark-submit 
--master spark://node1:7077 
--class org.apache.spark.example.SaprkPi 
../lib/spark-examples-1.6.0-hadoop2.6.0.jar 
1000 

執(zhí)行流程圖以及原理：​

​

Standalone-cluster 提交方式

提交命令如下：以官方給的計(jì)算 PI 的代碼為例。

./spark-submit 
--master spark://node1:7077 
--deploy-mode cluster 
--class org.apache.spark.example.SaprkPi 
../lib/spark-examples-1.6.0-hadoop2.6.0.jar 
1000 

執(zhí)行流程圖以及原理：​

​

Yarn-cluster 提交方式

提交命令如下：以官方給的計(jì)算 PI 的代碼為例。

./spark-submit 
--master yarn 
--deploy-mode client 
--class org.apache.spark.example.SaprkPi 
../lib/spark-examples-1.6.0-hadoop2.6.0.jar 
1000 

執(zhí)行流程圖以及原理：​

​

Yarn-cluster 提交方式

提交命令如下：以官方給的計(jì)算 PI 的代碼為例。

​./spark-submit 
--master yarn 
--deploy-mode cluster 
--class org.apache.spark.example.SaprkPi 
../lib/spark-examples-1.6.0-hadoop2.6.0.jar 
1000 

執(zhí)行流程圖以及原理：​

​

3 RDD

Spark core 最核心的就是 Resilient Distributed Dataset (RDD) 了，RDD 比較抽象了。源碼中 RDD.scala 中對(duì) RDD 進(jìn)行了一段描述。最主要的是下面的五個(gè)方面;

/** 
* Internally, each RDD is characterized by five main properties: 
* 
* - A list of partitions 
* - A function for computing each split 
* - A list of dependencies on other RDDs 
* - Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned) 
* - Optionally, a list of preferred locations to compute each split on (e.g. block locations for 
* an HDFS file) 
* 
* All of the scheduling and execution in Spark is done based on these methods, allowing each RDD 
* to implement its own way of computing itself. Indeed, users can implement custom RDDs (e.g. for 
* reading data from a new storage system) by overriding these functions.  
*/ 

RDD 的五大特性：

1. RDD 是由一系列的 Partition 組成的。
2. 函數(shù)作用在每一個(gè) split 上。
3. RDD 之間有一系列依賴關(guān)系。
4. 分區(qū)器是作用在 K,V 格式的 RDD 上。
5. RDD 提供一系列***的位置

先記住這五個(gè)特性，之后的學(xué)習(xí)會(huì)慢慢體會(huì)到這樣設(shè)計(jì)的好處。下面是理解 RDD 的邏輯圖;​

​

看這個(gè)圖再回頭理解一下上面的五個(gè) RDD 的特性。

RDD 的彈性表現(xiàn)在 Partition 的數(shù)量上，并且大小沒(méi)有限制。RDD 的依賴關(guān)系，可以基于上一個(gè) RDD 計(jì)算出下一個(gè) RDD。RDD 的每個(gè) partition 是分布在不同數(shù)據(jù)節(jié)點(diǎn)上的，所有 RDD 的分布式的。RDD 提供了一些列的***的計(jì)算位置，體現(xiàn)了數(shù)據(jù)的本地化，我之前的這篇文章寫過(guò)：一文搞懂?dāng)?shù)據(jù)本地化級(jí)別

RDD 還有一個(gè) Lineage 的東西，叫做血統(tǒng)。

Lineage 簡(jiǎn)介：利用內(nèi)存加快數(shù)據(jù)加載，在其它的In-Memory類數(shù)據(jù)庫(kù)或Cache類系統(tǒng)中也有實(shí)現(xiàn)。Spark的主要區(qū)別在于它采用血統(tǒng)來(lái)實(shí)現(xiàn)分布式運(yùn)算環(huán)境下的數(shù)據(jù)容錯(cuò)性(節(jié)點(diǎn)失效、數(shù)據(jù)丟失)問(wèn)題。

RDD Lineage被稱為RDD運(yùn)算圖或RDD依賴關(guān)系圖，是RDD所有父RDD的圖。它是在RDD上執(zhí)行transformations函數(shù)并創(chuàng)建邏輯執(zhí)行計(jì)劃(logical execution plan)的結(jié)果，是RDD的邏輯執(zhí)行計(jì)劃。

相比其它系統(tǒng)的細(xì)顆粒度的內(nèi)存數(shù)據(jù)更新級(jí)別的備份或者LOG機(jī)制，RDD 的 Lineage 記錄的是粗顆粒度的特定數(shù)據(jù)轉(zhuǎn)換(Transformation)操作(filter, map, join etc.)行為。當(dāng)這個(gè) RDD 的部分分區(qū)數(shù)據(jù)丟失時(shí)，它可以通過(guò)Lineage找到丟失的父RDD的分區(qū)進(jìn)行局部計(jì)算來(lái)恢復(fù)丟失的數(shù)據(jù)，這樣可以節(jié)省資源提高運(yùn)行效率。這種粗顆粒的數(shù)據(jù)模型，限制了Spark的運(yùn)用場(chǎng)合，但同時(shí)相比細(xì)顆粒度的數(shù)據(jù)模型，也帶來(lái)了性能的提升。

4 控制算子

控制算子有三種：cache, persist, checkpoint, 以上算子都可以將 RDD 持久化、持久化的單位是 Partition。

cache 和 persist 都是懶執(zhí)行的，必須有一個(gè) action 算子來(lái)觸發(fā)他們執(zhí)行。checkpoint 不僅可以將 RDD 持久化到磁盤，還能切斷 RDD 之間的依賴關(guān)系。

說(shuō)幾點(diǎn)區(qū)別：

• cache 的持久化級(jí)別是 Memory_Only，就這一個(gè)。
• persist 的持久化級(jí)別：常用的有Memory_Only 和Memory_and_Disk_2， 數(shù)字 2 表示副本數(shù)。
• checkpoint 主要是用來(lái)做容錯(cuò)的。

checkpoint 的執(zhí)行原理是：當(dāng) RDD 的 job 執(zhí)行完畢之后，會(huì)從 finalRDD 進(jìn)行回溯。當(dāng)回溯到某一個(gè) RDD 調(diào)用了 checkpoint 方法，會(huì)對(duì)當(dāng)前的 RDD 做一個(gè)標(biāo)記。Spark 框架會(huì)自動(dòng)啟動(dòng)一個(gè)新的 Job ，重新計(jì)算這個(gè) RDD 的數(shù)據(jù)，將數(shù)據(jù)持久化到 HDFS 上。根據(jù)這個(gè)原理，我們可以進(jìn)行優(yōu)化，對(duì) RDD 進(jìn)行 checkpoint 之前，***先對(duì)這個(gè) RDD 進(jìn)行 cache, 這樣啟動(dòng)新的 job 只需要將內(nèi)存中的數(shù)據(jù)拷貝到 HDFS 上就可以了，節(jié)省了重新計(jì)算這一步。

5 RDD 的依賴關(guān)系

窄依賴：指父RDD的每一個(gè)分區(qū)最多被一個(gè)子RDD的分區(qū)所用，表現(xiàn)為一個(gè)父RDD的分區(qū)對(duì)應(yīng)于一個(gè)子RDD的分區(qū)，和兩個(gè)父RDD的分區(qū)對(duì)應(yīng)于一個(gè)子RDD 的分區(qū)。圖中，map/filter/union屬于***類，對(duì)輸入進(jìn)行協(xié)同劃分(co-partitioned)的join屬于第二類。窄依賴不會(huì)產(chǎn)生 shuffle。

寬依賴：指子RDD的分區(qū)依賴于父RDD的所有分區(qū)，這是因?yàn)?shuffle 類操作，如圖中的 groupByKey 和未經(jīng)協(xié)同劃分的 join。 遇到寬依賴會(huì)產(chǎn)生 shuffle 。

上面我們說(shuō)到了 RDD 之間的依賴關(guān)系，這些依賴關(guān)系形成了一個(gè)人 DAG 有向無(wú)環(huán)圖。DAG 創(chuàng)建完成之后，會(huì)被提交給 DAGScheduler, 它負(fù)責(zé)把 DAG 劃分相互依賴的多個(gè) stage ，劃分依據(jù)就是 RDD 之間的窄寬依賴。換句話說(shuō)就是，遇到一個(gè)寬依賴就劃分一個(gè) stage，每一個(gè) stage 包含一個(gè)或多個(gè) stask 任務(wù)。然后將這些 task 以 taskset 的方式提交給 TaskScheduler 運(yùn)行。也可以說(shuō) stage 是由一組并行的 task 組成。下圖很清楚的描述了 stage 的劃分。​

​

6 Stage劃分思路

接上圖，Spark 劃分 stage 的整體思路是：從后往前推，遇到寬依賴就斷開(kāi)，劃分為一個(gè)stage;遇到窄依賴就將這個(gè) RDD 加入該 stage 中。

因此在圖中 RDD C, RDD D, RDD E, RDD F 被構(gòu)建在一個(gè) stage 中, RDD A被構(gòu)建在一個(gè)單獨(dú)的Stage中,而 RDD B 和 RDD G 又被構(gòu)建在同一個(gè) stage中。

另一個(gè)角度

一個(gè) Job 會(huì)被拆分為多組 Task，每組任務(wù)被稱為一個(gè)Stage就像 Map Stage，Reduce Stage。

Stage 的劃分簡(jiǎn)單的說(shuō)是以 shuffle 和 result 這兩種類型來(lái)劃分。在 Spark中有兩類 task，一類是 shuffleMapTask，一類是 resultTask，***類 task的輸出是 shuffle 所需數(shù)據(jù)，第二類 task 的輸出是 result，stage的劃分也以此為依據(jù)，shuffle 之前的所有變換是一個(gè) stage，shuffle之后的操作是另一個(gè)stage。

如果 job 中有多次 shuffle，那么每個(gè) shuffle 之前都是一個(gè) stage. 會(huì)根據(jù) RDD 之間的依賴關(guān)系將 DAG圖劃分為不同的階段，對(duì)于窄依賴，由于 partition 依賴關(guān)系的確定性，partition 的轉(zhuǎn)換處理就可以在同一個(gè)線程里完成，窄依賴就被 spark 劃分到同一個(gè) stage 中，而對(duì)于寬依賴，只能等父 RDD shuffle 處理完成后，下一個(gè) stage 才能開(kāi)始接下來(lái)的計(jì)算。之所以稱之為 ShuffleMapTask 是因?yàn)樗枰獙⒆约旱挠?jì)算結(jié)果通過(guò) shuffle 到下一個(gè) stage 中。