大資料基礎：Spark作業原理及基礎概念

導語 | Apache Spark 是專為大規模資料處理而設計的快速通用計算引擎，在資料挖掘和機器學習領域有著廣泛的應用，現在也已形成一個高速發展、應用廣泛的生態系統，本文將為大家詳細介紹 Spark 的核心技術原理，希望與大家一同交流，文章作者：熊峰，騰訊大資料研發工程師，

一、Spark 介紹及生態

Spark是UC Berkeley AMP Lab開源的通用分布式并行計算框架，目前已成為Apache軟體基金會的頂級開源專案，至于為什么我們要學習Spark，可以總結為下面三點：

1. Spark相對于hadoop的優勢

（1）高性能

Spark具有hadoop MR所有的優點，hadoop MR每次計算的中間結果都會存盤到HDFS的磁盤上，而Spark的中間結果可以保存在記憶體，在記憶體中進行資料處理，

（2）高容錯

• 基于“血統”(Lineage)的資料恢復：spark引入了彈性分布式資料集RDD的抽象，它是分布在一組節點中的只讀的資料的集合，這些集合是彈性的且是相互依賴的，如果資料集中的一部分的資料發生丟失可以根據“血統”關系進行重建，

• CheckPoint容錯：RDD計算時可以通過checkpoint進行容錯，checkpoint有兩種檢測方式：通過冗余資料和日志記錄更新操作，在RDD中的doCheckPoint方法相當于通過冗余資料來快取資料，而“血統”是通過粗粒度的記錄更新操作來實作容錯的，CheckPoint容錯是對血統檢測進行的容錯輔助，避免“血統”(Lineage)過長造成的容錯成本過高，

（3）spark的通用性

spark 是一個通用的大資料計算框架，相對于hadoop它提供了更豐富的使用場景，

spark相對于hadoop map reduce兩種操作還提供了更為豐富的操作，分為action(collect,reduce,save…)和transformations（map,union,join,filter…）,同時在各節點的通信模型中相對于hadoop的shuffle操作還有磁區，控制中間結果存盤，物化視圖等，

2. spark 生態介紹

Spark支持多種編程語言，包括Java、Python、R和Scala，在計算資源調度層支持local模式，standalone模式，yarn模式以及k8s等，

同時spark有多組件的支持應用場景，在spark core的基礎上提供了spark Streaming，spark SQL，spark Mllib，spark R，GraphX等組件，

spark Streaming用于實時流計算，spark SQL旨在將熟悉的SQL資料庫查詢與更復雜的基于演算法的分析相結合，GraphX用于圖計算，spark Mllib用于機器學習，spark R用于對R語言的資料計算，

spark 支持多種的存盤介質，在存盤層spark支持從hdfs，hive，aws等讀入和寫出資料，也支持從hbase，es等大資料庫中讀入和寫出資料，同時也支持從mysql，pg等關系型資料庫中讀入寫出資料，在實時流計算在可以從flume，kafka等多種資料源獲取資料并執行流式計算，

在資料格式上spark也支持的非常豐富，比如常見的txt，json，csv等格式，同時也支持parquet，orc，avro等格式，這幾種格式在資料壓縮和海量資料查詢上優勢也較為明顯，

二、spark 原理及特點

1. spark core

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

（1）spark 基礎配置

sparkContext是spark應用程式的入口，spark應用程式的提交和執行離不開sparkContext，它隱藏了網路通信，分布式部署，訊息通信，存盤體系，計算存盤等，開發人員只需要通過sparkContext等api進行開發即可，

sparkRpc 基于netty實作，分為異步和同步兩種方式，事件總線主要用于sparkContext組件間的交換，它屬于監聽者模式，采用異步呼叫，度量系統主要用于系統的運行監控，

（2）spark 存盤系統

它用于管理spark運行中依賴的資料存盤方式和存盤位置，spark的存盤系統優先考慮在各節點以記憶體的方式存盤資料，記憶體不足時將資料寫入磁盤中，這也是spark計算性能高的重要原因，

我們可以靈活的控制資料存盤在記憶體還是磁盤中，同時可以通過遠程網路呼叫將結果輸出到遠程存盤中，比如hdfs，hbase等，

（3）spark 調度系統

spark 調度系統主要由DAGScheduler和TaskScheduler組成，

DAGScheduler 主要是把一個Job根據RDD間的依賴關系，劃分為多個Stage，對于劃分后的每個Stage都抽象為一個或多個Task組成的任務集，并交給TaskScheduler來進行進一步的任務調度，而TaskScheduler 負責對每個具體的Task進行調度，

具體調度演算法有FIFO，FAIR：

• FIFO調度：先進先出，這是Spark默認的調度模式，

• FAIR調度：支持將作業分組到池中，并為每個池設定不同的調度權重，任務可以按照權重來決定執行順序，

2. spark sql

spark sql提供了基于sql的資料處理方法，使得分布式的資料集處理變的更加簡單，這也是spark 廣泛使用的重要原因，

目前大資料相關計算引擎一個重要的評價指標就是：是否支持sql，這樣才會降低使用者的門檻，spark sql提供了兩種抽象的資料集合DataFrame和DataSet，

DataFrame 是spark Sql 對結構化資料的抽象，可以簡單的理解為spark中的表，相比較于RDD多了資料的表結構資訊(schema).DataFrame = Data + schema

RDD是分布式物件集合，DataFrame是分布式Row的集合，提供了比RDD更豐富的算子，同時提升了資料的執行效率，

DataSet 是資料的分布式集合 ，它具有RDD強型別的優點 和Spark SQL優化后執行的優點，DataSet可以由jvm物件構建，然后使用map，filter，flatmap等操作函式操作，

3. spark streaming

這個模塊主要是對流資料的處理，支持流資料的可伸縮和容錯處理，可以與Flume和Kafka等已建立的資料源集成，Spark Streaming的實作，也使用RDD抽象的概念，使得在為流資料撰寫應用程式時更為方便，

4. spark特點

（1）spark 計算速度快

spark將每個任務構建成DAG進行計算，內部的計算程序通過彈性式分布式資料集RDD在記憶體在進行計算，相比于hadoop的mapreduce效率提升了100倍，

（2）易于使用

spark 提供了大量的算子，開發只需呼叫相關api進行實作無法關注底層的實作原理，

通用的大資料解決方案

相較于以前離線任務采用mapreduce實作，實時任務采用storm實作，目前這些都可以通過spark來實作，降低來開發的成本，同時spark 通過spark SQL降低了用戶的學習使用門檻，還提供了機器學習，圖計算引擎等，

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

學習使用中可以采用local 模型進行任務的除錯，在正式環境中又提供了standalone，yarn等模式，方便用戶選擇合適的資源管理模式進行適配，

（4）社區支持

spark 生態圈豐富，迭代更新快，成為大資料領域必備的計算引擎，

三、spark 運行模式及集群角色

1. spark運行模式

2. spark集群角色

下圖是spark的集群角色圖，主要有集群管理節點cluster manager，作業節點worker，執行器executor，驅動器driver和應用程式application 五部分組成，下面詳細說明每部分的特點，

（1）Cluster Manager

集群管理器，它存在于Master行程中，主要用來對應用程式申請的資源進行管理，根據其部署模式的不同，可以分為local，standalone，yarn，mesos等模式，

（2）worker

worker是spark的作業節點，用于執行任務的提交，主要作業職責有下面四點：

• worker節點通過注冊機向cluster manager匯報自身的cpu，記憶體等資訊，

• worker 節點在spark master作用下創建并啟用executor，executor是真正的計算單元，

• spark master將任務Task分配給worker節點上的executor并執行運用，

• worker節點同步資源資訊和executor狀態資訊給cluster manager，

在yarn 模式下運行worker節點一般指的是NodeManager節點，standalone模式下運行一般指的是slave節點，

（3）executor

executor 是真正執行計算任務的組件，它是application運行在worker上的一個行程，這個行程負責Task的運行，它能夠將資料保存在記憶體或磁盤存盤中，也能夠將結果資料回傳給Driver，

（4）Application

application是Spark API 編程的應用程式，它包括實作Driver功能的代碼和在程式中各個executor上要執行的代碼，一個application由多個job組成，其中應用程式的入口為用戶所定義的main方法，

（5）Driver

驅動器節點，它是一個運行Application中main函式并創建SparkContext的行程，application通過Driver 和Cluster Manager及executor進行通訊，它可以運行在application節點上，也可以由application提交給Cluster Manager，再由Cluster Manager安排worker進行運行，

Driver節點也負責提交Job，并將Job轉化為Task，在各個Executor行程間協調Task的調度，

（6）sparkContext

sparkContext是整個spark應用程式最關鍵的一個物件，是Spark所有功能的主要入口點，核心作用是初始化spark應用程式所需要的組件，同時還負責向master程式進行注冊等，

3. spark其它核心概念

（1）RDD

它是Spark中最重要的一個概念，是彈性分布式資料集，是一種容錯的、可以被并行操作的元素集合，是Spark對所有資料處理的一種基本抽象，可以通過一系列的算子對rdd進行操作，主要分為Transformation和Action兩種操作，

• ?????Transformation(轉換)：是對已有的RDD進行換行生成新的RDD，對于轉換程序采用惰性計算機制，不會立即計算出結果，常用的方法有map，filter，flatmap等，

• Action(執行)：對已有對RDD對資料執行計算產生結果，并將結果回傳Driver或者寫入到外部存盤中，常用到方法有reduce，collect，saveAsTextFile等，

（2）DAG

DAG是一個有向無環圖，在Spark中， 使用 DAG 來描述我們的計算邏輯，主要分為DAG Scheduler 和Task Scheduler，

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

（3）DAG Scheduler

DAG Scheduler 是面向stage的高層級的調度器，DAG Scheduler把DAG拆分為多個Task，每組Task都是一個stage，決議時是以shuffle為邊界進行反向構建的，每當遇見一個shuffle，spark就會產生一個新的stage，接著以TaskSet的形式提交給底層的調度器（task scheduler），每個stage封裝成一個TaskSet，DAG Scheduler需要記錄RDD被存入磁盤物化等動作，同時會需要Task尋找最優等調度邏輯，以及監控因shuffle跨節點輸出導致的失敗，

（4）Task Scheduler

Task Scheduler 負責每一個具體任務的執行，它的主要職責包括

• 任務集的調度管理；

• 狀態結果跟蹤；

• 物理資源調度管理；

• 任務執行；

• 獲取結果，

（5）Job

job是有多個stage構建的并行的計算任務，job是由spark的action操作來觸發的，在spark中一個job包含多個RDD以及作用在RDD的各種操作算子，

（6）stage

DAG Scheduler會把DAG切割成多個相互依賴的Stage，劃分Stage的一個依據是RDD間的寬窄依賴，

在對Job中的所有操作劃分Stage時，一般會按照倒序進行，即從Action開始，遇到窄依賴操作，則劃分到同一個執行階段，遇到寬依賴操作，則劃分一個新的執行階段，且新的階段為之前階段的parent，然后依次類推遞回執行，

child Stage需要等待所有的parent Stage執行完之后才可以執行，這時Stage之間根據依賴關系構成了一個大粒度的DAG，在一個Stage內，所有的操作以串行的Pipeline的方式，由一組Task完成計算，

（7）TaskSet Task

TaskSet 可以理解為一種任務，對應一個stage，是Task組成的任務集，一個TaskSet中的所有Task沒有shuffle依賴可以并行計算，

Task是spark中最獨立的計算單元，由Driver Manager發送到executer執行，通常情況一個task處理spark RDD一個partition，Task分為ShuffleMapTask和ResultTask兩種，位于最后一個Stage的Task為ResultTask，其他階段的屬于ShuffleMapTask，

四、spark作業運行流程

1. spark作業運行流程

spark應用程式以行程集合為單位在分布式集群上運行，通過driver程式的main方法創建sparkContext的物件與集群進行互動，具體運行流程如下：

• sparkContext向cluster Manager申請CPU，記憶體等計算資源，

• cluster Manager分配應用程式執行所需要的資源，在worker節點創建executor，

• sparkContext將程式代碼和task任務發送到executor上進行執行，代碼可以是編譯成的jar包或者python檔案等，接著sparkContext會收集結果到Driver端，
  

2. spark RDD迭代程序

• sparkContext創建RDD物件，計算RDD間的依賴關系，并組成一個DAG有向無環圖，

• DAGScheduler將DAG劃分為多個stage，并將stage對應的TaskSet提交到集群的管理中心，stage的劃分依據是RDD中的寬窄依賴，spark遇見寬依賴就會劃分為一個stage，每個stage中包含來一個或多個task任務，避免多個stage之間訊息傳遞產生的系統開銷，

• taskScheduler 通過集群管理中心為每一個task申請資源并將task提交到worker的節點上進行執行，

• worker上的executor執行具體的任務，
  

3. yarn資源管理器介紹

spark 程式一般是運行在集群上的，spark on yarn是作業或生產上用的非常多的一種運行模式，

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

yarn是分布式資源管理和任務管理管理，主要由ResourceManager，NodeManager和ApplicationMaster三個模塊組成，

ResourceManager 主要負責集群的資源管理，監控和分配，對于所有的應用它有絕對的控制權和資源管理權限，

NodeManager 負責節點的維護，執行和監控task運行狀況，會通過心跳的方式向ResourceManager匯報自己的資源使用情況，

yarn資源管理器的每個節點都運行著一個NodeManager，是ResourceManager的代理，如果主節點的ResourceManager宕機后，會連接ResourceManager的備用節點，

ApplicationMaster 負責具體應用程式的調度和資源的協調，它會與ResourceManager協商進行資源申請，ResourceManager以container容器的形式將資源分配給application進行運行，同時負責任務的啟停，

container 是資源的抽象，它封裝著每個節點上的資源資訊(cpu,記憶體，磁盤，網路等)，yarn將任務分配到container上運行，同時該任務只能使用container描述的資源，達到各個任務間資源的隔離，

4. spark程式在yarn上執行流程

spark on yarn分為兩種模式yarn-client模式，和yarn—cluster模式，一般線上采用的是yarn-cluster模式，

（1）yarn-client模式

driver在客戶端本地執行，這種模式可以使得spark application和客戶端進行互動，因為driver在客戶端可以通過webUI訪問driver的狀態，同時Driver會與yarn集群中的Executor進行大量的通信，會造成客戶機網卡流量的大量增加，

（2）yarn-cluster模式

Yarn-Cluster主要用于生產環境中，因為Driver運行在Yarn集群中某一臺NodeManager中，每次提交任務的Driver所在的機器都是隨機的，不會產生某一臺機器網卡流量激增的現象，缺點是任務提交后不能看到日志，只能通過yarn查看日志，

下圖是yarn-cluster運行模式：

client 向yarn提交應用程式，包含ApplicationMaster程式、啟動ApplicationMaster的命令、需要在Executor中運行的程式等，

ApplicationMaster程式啟動ApplicationMaster的命令、需要在Executor中運行的程式等，

ApplicationMaster向ResourceManager注冊，這樣用戶可以直接通過ResourceManage查看應用程式的運行狀態，

ApplicationMaster申請到資源（也就是Container）后，便與對應的NodeManager通信，啟動Task，

Task向ApplicationMaster匯報運行的狀態和進度，以讓ApplicationMaster隨時掌握各個任務的運行狀態，從而可以在任務失敗時重新啟動任務，

應用程式運行完成后，ApplicationMaster向ResourceManager申請注銷并關閉自己，

參考文獻：

[1] Spark on Yarn架構原理:

https://blog.csdn.net/lijingjingchn/article/details/85012470

[1] Spark on Yarn詳解:

https://www.cnblogs.com/bigdata1024/p/12116621.html

[1] Spark任務提交方式和執行流程:

https://www.cnblogs.com/frankdeng/p/9301485.html

[1] Spark容錯機制:

https://www.cnblogs.com/cynchanpin/p/7163160.html

[1] spark 之 scheduler:

https://mp.weixin.qq.com/s/9g5e5WlmXUyQDXiU6PTGZA?token=1292183487&lang=zh_CN

[1] Spark作業原理:

https://blog.csdn.net/qq_16681169/article/details/82432841

[1] Spark RDD:

https://www.cnblogs.com/zlslch/p/5942204.html

[1] Spark基本概念快速入門:

https://www.leonlu.cc/profession/17-spark-terminology/

[1] spark中dag的介紹:

https://blog.csdn.net/newchitu/article/details/92796302

[2] spark:

https://spark.apache.org/docs/3.0.0-preview/index.html

文章推薦

百億級實時計算系統性能優化–—Elasticsearch篇