一、Spark 介紹及生態(tài)

Spark是UC Berkeley AMP Lab開源的通用分布式并行計(jì)算框架，目前已成為Apache軟件基金會(huì)的頂級(jí)開源項(xiàng)目。至于為什么我們要學(xué)習(xí)Spark，可以總結(jié)為下面三點(diǎn)：

1. Spark相對(duì)于hadoop的優(yōu)勢(shì)

（1）高性能

Spark具有hadoop MR所有的優(yōu)點(diǎn)，hadoop MR每次計(jì)算的中間結(jié)果都會(huì)存儲(chǔ)到HDFS的磁盤上，而Spark的中間結(jié)果可以保存在內(nèi)存，在內(nèi)存中進(jìn)行數(shù)據(jù)處理。

（2）高容錯(cuò)

• 基于“血統(tǒng)”(Lineage)的數(shù)據(jù)恢復(fù)：spark引入了彈性分布式數(shù)據(jù)集RDD的抽象，它是分布在一組節(jié)點(diǎn)中的只讀的數(shù)據(jù)的集合，這些集合是彈性的且是相互依賴的，如果數(shù)據(jù)集中的一部分的數(shù)據(jù)發(fā)生丟失可以根據(jù)“血統(tǒng)”關(guān)系進(jìn)行重建。

• CheckPoint容錯(cuò)：RDD計(jì)算時(shí)可以通過(guò)checkpoint進(jìn)行容錯(cuò)，checkpoint有兩種檢測(cè)方式：通過(guò)冗余數(shù)據(jù)和日志記錄更新操作。在RDD中的doCheckPoint方法相當(dāng)于通過(guò)冗余數(shù)據(jù)來(lái)緩存數(shù)據(jù)，而“血統(tǒng)”是通過(guò)粗粒度的記錄更新操作來(lái)實(shí)現(xiàn)容錯(cuò)的。CheckPoint容錯(cuò)是對(duì)血統(tǒng)檢測(cè)進(jìn)行的容錯(cuò)輔助，避免“血統(tǒng)”(Lineage)過(guò)長(zhǎng)造成的容錯(cuò)成本過(guò)高。

（3）spark的通用性

spark 是一個(gè)通用的大數(shù)據(jù)計(jì)算框架，相對(duì)于hadoop它提供了更豐富的使用場(chǎng)景。

spark相對(duì)于hadoop map reduce兩種操作還提供了更為豐富的操作，分為action(collect,reduce,save…)和transformations（map,union,join,filter…）,同時(shí)在各節(jié)點(diǎn)的通信模型中相對(duì)于hadoop的shuffle操作還有分區(qū)，控制中間結(jié)果存儲(chǔ)，物化視圖等。

2. spark 生態(tài)介紹

 Spark支持多種編程語(yǔ)言，包括Java、Python、R和Scala。在計(jì)算資源調(diào)度層支持local模式，standalone模式，yarn模式以及k8s等。

同時(shí)spark有多組件的支持應(yīng)用場(chǎng)景，在spark core的基礎(chǔ)上提供了spark Streaming，spark SQL，spark Mllib，spark R，GraphX等組件。

spark Streaming用于實(shí)時(shí)流計(jì)算，spark SQL旨在將熟悉的SQL數(shù)據(jù)庫(kù)查詢與更復(fù)雜的基于算法的分析相結(jié)合，GraphX用于圖計(jì)算，spark Mllib用于機(jī)器學(xué)習(xí)，spark R用于對(duì)R語(yǔ)言的數(shù)據(jù)計(jì)算。

spark 支持多種的存儲(chǔ)介質(zhì)，在存儲(chǔ)層spark支持從hdfs，hive，aws等讀入和寫出數(shù)據(jù)，也支持從hbase，es等大數(shù)據(jù)庫(kù)中讀入和寫出數(shù)據(jù)，同時(shí)也支持從mysql，pg等關(guān)系型數(shù)據(jù)庫(kù)中讀入寫出數(shù)據(jù)，在實(shí)時(shí)流計(jì)算在可以從flume，kafka等多種數(shù)據(jù)源獲取數(shù)據(jù)并執(zhí)行流式計(jì)算。

在數(shù)據(jù)格式上spark也支持的非常豐富，比如常見的txt，json，csv等格式。同時(shí)也支持parquet，orc，avro等格式，這幾種格式在數(shù)據(jù)壓縮和海量數(shù)據(jù)查詢上優(yōu)勢(shì)也較為明顯。

二、spark 原理及特點(diǎn)

1. spark core

Spark Core是Spark的核心，其包含如下幾個(gè)部分：

（1）spark 基礎(chǔ)配置

sparkContext是spark應(yīng)用程序的入口，spark應(yīng)用程序的提交和執(zhí)行離不開sparkContext，它隱藏了網(wǎng)絡(luò)通信，分布式部署，消息通信，存儲(chǔ)體系，計(jì)算存儲(chǔ)等，開發(fā)人員只需要通過(guò)sparkContext等api進(jìn)行開發(fā)即可。

sparkRpc 基于netty實(shí)現(xiàn)，分為異步和同步兩種方式。事件總線主要用于sparkContext組件間的交換，它屬于監(jiān)聽者模式，采用異步調(diào)用。度量系統(tǒng)主要用于系統(tǒng)的運(yùn)行監(jiān)控。

（2）spark 存儲(chǔ)系統(tǒng)

它用于管理spark運(yùn)行中依賴的數(shù)據(jù)存儲(chǔ)方式和存儲(chǔ)位置，spark的存儲(chǔ)系統(tǒng)優(yōu)先考慮在各節(jié)點(diǎn)以內(nèi)存的方式存儲(chǔ)數(shù)據(jù)，內(nèi)存不足時(shí)將數(shù)據(jù)寫入磁盤中，這也是spark計(jì)算性能高的重要原因。

我們可以靈活的控制數(shù)據(jù)存儲(chǔ)在內(nèi)存還是磁盤中，同時(shí)可以通過(guò)遠(yuǎn)程網(wǎng)絡(luò)調(diào)用將結(jié)果輸出到遠(yuǎn)程存儲(chǔ)中，比如hdfs，hbase等。

（3）spark 調(diào)度系統(tǒng)

spark 調(diào)度系統(tǒng)主要由DAGScheduler和TaskScheduler組成。

DAGScheduler 主要是把一個(gè)Job根據(jù)RDD間的依賴關(guān)系，劃分為多個(gè)Stage，對(duì)于劃分后的每個(gè)Stage都抽象為一個(gè)或多個(gè)Task組成的任務(wù)集，并交給TaskScheduler來(lái)進(jìn)行進(jìn)一步的任務(wù)調(diào)度。而TaskScheduler 負(fù)責(zé)對(duì)每個(gè)具體的Task進(jìn)行調(diào)度。

具體調(diào)度算法有FIFO，F(xiàn)AIR：

• FIFO調(diào)度：先進(jìn)先出，這是Spark默認(rèn)的調(diào)度模式。

• FAIR調(diào)度：支持將作業(yè)分組到池中，并為每個(gè)池設(shè)置不同的調(diào)度權(quán)重，任務(wù)可以按照權(quán)重來(lái)決定執(zhí)行順序。

2. spark sql

spark sql提供了基于sql的數(shù)據(jù)處理方法，使得分布式的數(shù)據(jù)集處理變的更加簡(jiǎn)單，這也是spark 廣泛使用的重要原因。

目前大數(shù)據(jù)相關(guān)計(jì)算引擎一個(gè)重要的評(píng)價(jià)指標(biāo)就是：是否支持sql，這樣才會(huì)降低使用者的門檻。spark sql提供了兩種抽象的數(shù)據(jù)集合DataFrame和DataSet。

DataFrame 是spark Sql 對(duì)結(jié)構(gòu)化數(shù)據(jù)的抽象，可以簡(jiǎn)單的理解為spark中的表，相比較于RDD多了數(shù)據(jù)的表結(jié)構(gòu)信息(schema).DataFrame = Data + schema

RDD是分布式對(duì)象集合，DataFrame是分布式Row的集合，提供了比RDD更豐富的算子，同時(shí)提升了數(shù)據(jù)的執(zhí)行效率。

DataSet 是數(shù)據(jù)的分布式集合 ，它具有RDD強(qiáng)類型的優(yōu)點(diǎn) 和Spark SQL優(yōu)化后執(zhí)行的優(yōu)點(diǎn)。DataSet可以由jvm對(duì)象構(gòu)建，然后使用map，filter，flatmap等操作函數(shù)操作。

3. spark streaming

這個(gè)模塊主要是對(duì)流數(shù)據(jù)的處理，支持流數(shù)據(jù)的可伸縮和容錯(cuò)處理，可以與Flume和Kafka等已建立的數(shù)據(jù)源集成。Spark Streaming的實(shí)現(xiàn)，也使用RDD抽象的概念，使得在為流數(shù)據(jù)編寫應(yīng)用程序時(shí)更為方便。

4. spark特點(diǎn)

（1）spark 計(jì)算速度快

spark將每個(gè)任務(wù)構(gòu)建成DAG進(jìn)行計(jì)算，內(nèi)部的計(jì)算過(guò)程通過(guò)彈性式分布式數(shù)據(jù)集RDD在內(nèi)存在進(jìn)行計(jì)算，相比于hadoop的mapreduce效率提升了100倍。

（2）易于使用

spark 提供了大量的算子，開發(fā)只需調(diào)用相關(guān)api進(jìn)行實(shí)現(xiàn)無(wú)法關(guān)注底層的實(shí)現(xiàn)原理。
通用的大數(shù)據(jù)解決方案

相較于以前離線任務(wù)采用mapreduce實(shí)現(xiàn)，實(shí)時(shí)任務(wù)采用storm實(shí)現(xiàn)，目前這些都可以通過(guò)spark來(lái)實(shí)現(xiàn)，降低來(lái)開發(fā)的成本。同時(shí)spark 通過(guò)spark SQL降低了用戶的學(xué)習(xí)使用門檻，還提供了機(jī)器學(xué)習(xí)，圖計(jì)算引擎等。

（3）支持多種的資源管理模式

學(xué)習(xí)使用中可以采用local 模型進(jìn)行任務(wù)的調(diào)試，在正式環(huán)境中又提供了standalone，yarn等模式，方便用戶選擇合適的資源管理模式進(jìn)行適配。

（4）社區(qū)支持

spark 生態(tài)圈豐富，迭代更新快，成為大數(shù)據(jù)領(lǐng)域必備的計(jì)算引擎。

三、spark 運(yùn)行模式及集群角色

1. spark運(yùn)行模式

2. spark集群角色

下圖是spark的集群角色圖，主要有集群管理節(jié)點(diǎn)cluster manager，工作節(jié)點(diǎn)worker，執(zhí)行器executor，驅(qū)動(dòng)器driver和應(yīng)用程序application 五部分組成，下面詳細(xì)說(shuō)明每部分的特點(diǎn)。

（1）Cluster Manager

集群管理器，它存在于Master進(jìn)程中，主要用來(lái)對(duì)應(yīng)用程序申請(qǐng)的資源進(jìn)行管理，根據(jù)其部署模式的不同，可以分為local，standalone，yarn，mesos等模式。

（2）worker

worker是spark的工作節(jié)點(diǎn)，用于執(zhí)行任務(wù)的提交，主要工作職責(zé)有下面四點(diǎn)：

• worker節(jié)點(diǎn)通過(guò)注冊(cè)機(jī)向cluster manager匯報(bào)自身的cpu，內(nèi)存等信息。

• worker 節(jié)點(diǎn)在spark master作用下創(chuàng)建并啟用executor，executor是真正的計(jì)算單元。

• spark master將任務(wù)Task分配給worker節(jié)點(diǎn)上的executor并執(zhí)行運(yùn)用。

• worker節(jié)點(diǎn)同步資源信息和executor狀態(tài)信息給cluster manager。

在yarn 模式下運(yùn)行worker節(jié)點(diǎn)一般指的是NodeManager節(jié)點(diǎn)，standalone模式下運(yùn)行一般指的是slave節(jié)點(diǎn)。

（3）executor

executor 是真正執(zhí)行計(jì)算任務(wù)的組件，它是application運(yùn)行在worker上的一個(gè)進(jìn)程。這個(gè)進(jìn)程負(fù)責(zé)Task的運(yùn)行，它能夠?qū)?shù)據(jù)保存在內(nèi)存或磁盤存儲(chǔ)中，也能夠?qū)⒔Y(jié)果數(shù)據(jù)返回給Driver。

（4）Application

application是Spark API 編程的應(yīng)用程序，它包括實(shí)現(xiàn)Driver功能的代碼和在程序中各個(gè)executor上要執(zhí)行的代碼，一個(gè)application由多個(gè)job組成。其中應(yīng)用程序的入口為用戶所定義的main方法。

（5）Driver

驅(qū)動(dòng)器節(jié)點(diǎn)，它是一個(gè)運(yùn)行Application中main函數(shù)并創(chuàng)建SparkContext的進(jìn)程。application通過(guò)Driver 和Cluster Manager及executor進(jìn)行通訊。它可以運(yùn)行在application節(jié)點(diǎn)上，也可以由application提交給Cluster Manager，再由Cluster Manager安排worker進(jìn)行運(yùn)行。

Driver節(jié)點(diǎn)也負(fù)責(zé)提交Job，并將Job轉(zhuǎn)化為Task，在各個(gè)Executor進(jìn)程間協(xié)調(diào)Task的調(diào)度。

（6）sparkContext

sparkContext是整個(gè)spark應(yīng)用程序最關(guān)鍵的一個(gè)對(duì)象，是Spark所有功能的主要入口點(diǎn)。核心作用是初始化spark應(yīng)用程序所需要的組件，同時(shí)還負(fù)責(zé)向master程序進(jìn)行注冊(cè)等。

3. spark其它核心概念

（1）RDD

它是Spark中最重要的一個(gè)概念，是彈性分布式數(shù)據(jù)集，是一種容錯(cuò)的、可以被并行操作的元素集合，是Spark對(duì)所有數(shù)據(jù)處理的一種基本抽象?？梢酝ㄟ^(guò)一系列的算子對(duì)rdd進(jìn)行操作，主要分為Transformation和Action兩種操作。

• Transformation(轉(zhuǎn)換)：是對(duì)已有的RDD進(jìn)行換行生成新的RDD，對(duì)于轉(zhuǎn)換過(guò)程采用惰性計(jì)算機(jī)制，不會(huì)立即計(jì)算出結(jié)果。常用的方法有map，filter，flatmap等。

• Action(執(zhí)行)：對(duì)已有對(duì)RDD對(duì)數(shù)據(jù)執(zhí)行計(jì)算產(chǎn)生結(jié)果，并將結(jié)果返回Driver或者寫入到外部存儲(chǔ)中。常用到方法有reduce，collect，saveAsTextFile等。

（2）DAG

DAG是一個(gè)有向無(wú)環(huán)圖，在Spark中， 使用 DAG 來(lái)描述我們的計(jì)算邏輯。主要分為DAG Scheduler 和Task Scheduler。

圖片出自：https://blog.csdn.net/newchitu/article/details/92796302

（3）DAG Scheduler

DAG Scheduler 是面向stage的高層級(jí)的調(diào)度器，DAG Scheduler把DAG拆分為多個(gè)Task，每組Task都是一個(gè)stage，解析時(shí)是以shuffle為邊界進(jìn)行反向構(gòu)建的，每當(dāng)遇見一個(gè)shuffle，spark就會(huì)產(chǎn)生一個(gè)新的stage，接著以TaskSet的形式提交給底層的調(diào)度器（task scheduler），每個(gè)stage封裝成一個(gè)TaskSet。DAG Scheduler需要記錄RDD被存入磁盤物化等動(dòng)作，同時(shí)會(huì)需要Task尋找最優(yōu)等調(diào)度邏輯，以及監(jiān)控因shuffle跨節(jié)點(diǎn)輸出導(dǎo)致的失敗。

（4）Task Scheduler

Task Scheduler 負(fù)責(zé)每一個(gè)具體任務(wù)的執(zhí)行。它的主要職責(zé)包括

• 任務(wù)集的調(diào)度管理；

• 狀態(tài)結(jié)果跟蹤；

• 物理資源調(diào)度管理；

• 任務(wù)執(zhí)行；

• 獲取結(jié)果。

（5）Job

job是有多個(gè)stage構(gòu)建的并行的計(jì)算任務(wù)，job是由spark的action操作來(lái)觸發(fā)的，在spark中一個(gè)job包含多個(gè)RDD以及作用在RDD的各種操作算子。

（6）stage

DAG Scheduler會(huì)把DAG切割成多個(gè)相互依賴的Stage，劃分Stage的一個(gè)依據(jù)是RDD間的寬窄依賴。

在對(duì)Job中的所有操作劃分Stage時(shí)，一般會(huì)按照倒序進(jìn)行，即從Action開始，遇到窄依賴操作，則劃分到同一個(gè)執(zhí)行階段，遇到寬依賴操作，則劃分一個(gè)新的執(zhí)行階段，且新的階段為之前階段的parent，然后依次類推遞歸執(zhí)行。

child Stage需要等待所有的parent Stage執(zhí)行完之后才可以執(zhí)行，這時(shí)Stage之間根據(jù)依賴關(guān)系構(gòu)成了一個(gè)大粒度的DAG。在一個(gè)Stage內(nèi)，所有的操作以串行的Pipeline的方式，由一組Task完成計(jì)算。

（7）TaskSet Task

TaskSet 可以理解為一種任務(wù)，對(duì)應(yīng)一個(gè)stage，是Task組成的任務(wù)集。一個(gè)TaskSet中的所有Task沒有shuffle依賴可以并行計(jì)算。

Task是spark中最獨(dú)立的計(jì)算單元，由Driver Manager發(fā)送到executer執(zhí)行，通常情況一個(gè)task處理spark RDD一個(gè)partition。Task分為ShuffleMapTask和ResultTask兩種，位于最后一個(gè)Stage的Task為ResultTask，其他階段的屬于ShuffleMapTask。

四、spark作業(yè)運(yùn)行流程

1. spark作業(yè)運(yùn)行流程

spark應(yīng)用程序以進(jìn)程集合為單位在分布式集群上運(yùn)行，通過(guò)driver程序的main方法創(chuàng)建sparkContext的對(duì)象與集群進(jìn)行交互。具體運(yùn)行流程如下：

• sparkContext向cluster Manager申請(qǐng)CPU，內(nèi)存等計(jì)算資源。

• cluster Manager分配應(yīng)用程序執(zhí)行所需要的資源，在worker節(jié)點(diǎn)創(chuàng)建executor。

• sparkContext將程序代碼和task任務(wù)發(fā)送到executor上進(jìn)行執(zhí)行，代碼可以是編譯成的jar包或者python文件等。接著sparkContext會(huì)收集結(jié)果到Driver端。

2. spark RDD迭代過(guò)程

• sparkContext創(chuàng)建RDD對(duì)象，計(jì)算RDD間的依賴關(guān)系，并組成一個(gè)DAG有向無(wú)環(huán)圖。

• DAGScheduler將DAG劃分為多個(gè)stage，并將stage對(duì)應(yīng)的TaskSet提交到集群的管理中心，stage的劃分依據(jù)是RDD中的寬窄依賴，spark遇見寬依賴就會(huì)劃分為一個(gè)stage，每個(gè)stage中包含來(lái)一個(gè)或多個(gè)task任務(wù)，避免多個(gè)stage之間消息傳遞產(chǎn)生的系統(tǒng)開銷。

• taskScheduler 通過(guò)集群管理中心為每一個(gè)task申請(qǐng)資源并將task提交到worker的節(jié)點(diǎn)上進(jìn)行執(zhí)行。

• worker上的executor執(zhí)行具體的任務(wù)。

3. yarn資源管理器介紹

spark 程序一般是運(yùn)行在集群上的，spark on yarn是工作或生產(chǎn)上用的非常多的一種運(yùn)行模式。

沒有yarn模式前，每個(gè)分布式框架都要跑在一個(gè)集群上面，比如說(shuō)Hadoop要跑在一個(gè)集群上，Spark用集群的時(shí)候跑在standalone上。這樣的話整個(gè)集群的資源的利用率低，且管理起來(lái)比較麻煩。

yarn是分布式資源管理和任務(wù)管理管理，主要由ResourceManager，NodeManager和ApplicationMaster三個(gè)模塊組成。

ResourceManager 主要負(fù)責(zé)集群的資源管理，監(jiān)控和分配。對(duì)于所有的應(yīng)用它有絕對(duì)的控制權(quán)和資源管理權(quán)限。

NodeManager 負(fù)責(zé)節(jié)點(diǎn)的維護(hù)，執(zhí)行和監(jiān)控task運(yùn)行狀況。會(huì)通過(guò)心跳的方式向ResourceManager匯報(bào)自己的資源使用情況。

yarn資源管理器的每個(gè)節(jié)點(diǎn)都運(yùn)行著一個(gè)NodeManager，是ResourceManager的代理。如果主節(jié)點(diǎn)的ResourceManager宕機(jī)后，會(huì)連接ResourceManager的備用節(jié)點(diǎn)。

ApplicationMaster 負(fù)責(zé)具體應(yīng)用程序的調(diào)度和資源的協(xié)調(diào)，它會(huì)與ResourceManager協(xié)商進(jìn)行資源申請(qǐng)。ResourceManager以container容器的形式將資源分配給application進(jìn)行運(yùn)行。同時(shí)負(fù)責(zé)任務(wù)的啟停。

container 是資源的抽象，它封裝著每個(gè)節(jié)點(diǎn)上的資源信息(cpu,內(nèi)存，磁盤，網(wǎng)絡(luò)等)，yarn將任務(wù)分配到container上運(yùn)行，同時(shí)該任務(wù)只能使用container描述的資源，達(dá)到各個(gè)任務(wù)間資源的隔離。

4. spark程序在yarn上執(zhí)行流程

spark on yarn分為兩種模式y(tǒng)arn-client模式，和yarn—cluster模式，一般線上采用的是yarn-cluster模式。

（1）yarn-client模式

driver在客戶端本地執(zhí)行，這種模式可以使得spark application和客戶端進(jìn)行交互，因?yàn)閐river在客戶端可以通過(guò)webUI訪問(wèn)driver的狀態(tài)。同時(shí)Driver會(huì)與yarn集群中的Executor進(jìn)行大量的通信，會(huì)造成客戶機(jī)網(wǎng)卡流量的大量增加。

（2）yarn-cluster模式

Yarn-Cluster主要用于生產(chǎn)環(huán)境中，因?yàn)镈river運(yùn)行在Yarn集群中某一臺(tái)NodeManager中，每次提交任務(wù)的Driver所在的機(jī)器都是隨機(jī)的，不會(huì)產(chǎn)生某一臺(tái)機(jī)器網(wǎng)卡流量激增的現(xiàn)象，缺點(diǎn)是任務(wù)提交后不能看到日志。只能通過(guò)yarn查看日志。

下圖是yarn-cluster運(yùn)行模式：

client 向yarn提交應(yīng)用程序，包含ApplicationMaster程序、啟動(dòng)ApplicationMaster的命令、需要在Executor中運(yùn)行的程序等。

ApplicationMaster程序啟動(dòng)ApplicationMaster的命令、需要在Executor中運(yùn)行的程序等。

ApplicationMaster向ResourceManager注冊(cè)，這樣用戶可以直接通過(guò)ResourceManage查看應(yīng)用程序的運(yùn)行狀態(tài)。

ApplicationMaster申請(qǐng)到資源（也就是Container）后，便與對(duì)應(yīng)的NodeManager通信，啟動(dòng)Task。

Task向ApplicationMaster匯報(bào)運(yùn)行的狀態(tài)和進(jìn)度，以讓ApplicationMaster隨時(shí)掌握各個(gè)任務(wù)的運(yùn)行狀態(tài)，從而可以在任務(wù)失敗時(shí)重新啟動(dòng)任務(wù)。

應(yīng)用程序運(yùn)行完成后，ApplicationMaster向ResourceManager申請(qǐng)注銷并關(guān)閉自己。