項目3Spark的設計與運行原理學習目標（1）（2）（3）了解Spark架構中各個組件的作用熟悉Spark的運行流程掌握Spark的工作機制學習目標任務1Spark運行架構3.1.1Spark架構設計Spark運行架構如圖3-1所示，包括應用程序的控制進程DriverProgram、資源管理器ClusterManager、工作節(jié)點Worker和執(zhí)行進程Executor。DriverProgram進行任務控制，ClusterManager負責資源分配，Worker用于運行多個Executor，Executor負責具體的計算和數(shù)據(jù)存儲，每個Executor又由多個具體的Task（任務）組成。圖3-1Spark運行架構3.1.1Spark架構設計關于Spark架構的組件介紹如下。（1）Application用戶編寫的Spark應用程序。由一個DriverProgram代碼和分布在集群上的一個或多個Executor代碼構成。（2）DriverProgram運行Application的main函數(shù)并創(chuàng)建SparkContext的控制進程，負責任務控制。（3）SparkContext應用上下文對象。SparkContext會準備Spark應用程序的運行環(huán)境，向ClusterManager申請資源，進行任務的分配和監(jiān)控等。3.1.1Spark架構設計關于Spark架構的組件介紹如下。（4）ClusterManger資源管理器。在集群上獲取資源的外部服務，用于進行資源調(diào)度和監(jiān)控資源的使用情況，主要有3種：Standalone、Mesos和YARN。（5）Worker工作節(jié)點。在集群中運行應用程序代碼的節(jié)點，一個Worker可以運行一個或多個Executor進程。（6）Executor執(zhí)行進程。運行在Worker節(jié)點上的應用程序進程，負責運行Task和存儲數(shù)據(jù)，它可以將數(shù)據(jù)存儲在內(nèi)存或磁盤上。（7）Task：任務分配給Executor運行的工作單元，一個Executor可以運行一個或多個Task。3.1.2

Spark運行流程在學習Spark運行流程之前，先來認識幾個重要的概念。一個由多個Task組成的并行計算，執(zhí)行SparkAction算子時會觸發(fā)Job。（1）Job：作業(yè)看這里！每個Job被劃分成了多個相互獨立的Stage，每個Stage由一組Task組成。類似于MapReduce中的Map和Reduce。在Spark中，一個Application由一個含有入口函數(shù)main的DriverProgram和一個或多個Job構成，每個Job包含了多個Stage，每個Stage由多個Task組成。（2）Stage：階段RDD：彈性分布式數(shù)據(jù)集（resilientdistributeddataset）的英文縮寫，一個RDD就是一個分布式對象集合，它是Spark的基本計算單元。在進行數(shù)據(jù)存儲時，首先將數(shù)據(jù)放入內(nèi)存，當內(nèi)存放不下時會將一部分數(shù)據(jù)放到磁盤。（3）RDD3.1.2

Spark運行流程在學習Spark運行流程之前，先來認識幾個重要的概念。有向無環(huán)圖，Spark中使用DAG對RDD的關系進行建模。（4）DAG看這里！SparkContext將Application拆分成很多的Job，每個Job根據(jù)RDD之間的依賴關系構建成DAG，然后提交給DAGScheduler。DAGScheduler將DAG劃分為多個Stage，每個Stage都是一個TaskSet（任務集）。（5）DAGScheduler3.1.2

Spark運行流程在學習Spark運行流程之前，先來認識幾個重要的概念。將TaskSet分發(fā)給Worker運行。（6）TaskScheduler看這里！作用在DriverProgram中，負責分配任務。（7）SchedulerBackend作用在Executor中，管理所有任務的運行，接收到SchedulerBackend分配的任務后，將任務派發(fā)給線程池中的空閑線程去執(zhí)行。ExecutorBackend和Executor是一一對應的。（8）ExecutorBackend3.1.2

Spark運行流程Spark的基本運行流程如圖3-2所示，基本運行流程如下。圖3-2Spark的基本運行流程3.1.2

Spark運行流程（1）初始化Spark應用程序提交后，首先啟動DriverProgram運行Application的main函數(shù)并創(chuàng)建SparkContext。SparkContext初始化時會初始化DAGScheduler和TaskScheduler兩個核心組件。（2）SparkContext連接到ClusterManager，向ClusterManger注冊并申請資源，如申請內(nèi)存和磁盤等資源。ClusterManager目前主要有Spark原生的資源管理器Standalone、Mesos或YARN。（3）Worker節(jié)點會定期向ClusterManager發(fā)送心跳信息，這些心跳信息中包含Executor的狀態(tài)。ClusterManager根據(jù)Worker節(jié)點發(fā)送的心跳信息和SparkContext申請的資源情況來決定使用哪些Worker節(jié)點，并將資源分配給這些Worker，啟動Executor，Executor初始化線程池。3.1.2

Spark運行流程（4）ExecutorBackend向SparkContext注冊，SparkContext獲取了它自己的Executor進程集。（5）SparkContext接收到用戶代碼后生成具體任務集的過程如下。②劃分Stage①構建DAG③分發(fā)TaskSet3.1.2

Spark運行流程（6）ExecutorBackend接收到SchedulerBackend分配的任務后，將任務派發(fā)給線程池中的空閑線程去執(zhí)行，Executor以多線程的方式并行執(zhí)行任務，并將運行結果報告給SparkContext。（7）Task執(zhí)行結束后，通過BlockManager將結果存儲到內(nèi)存或磁盤中。（8）注銷。所有Task完成后，SparkContext向ClusterManager申請注銷，釋放所有資源。任務2Spark工作機制3.2.1

Spark內(nèi)存管理1.內(nèi)存劃分Spark的一個重要特點是：它是基于內(nèi)存計算的，因而更快。Executor是一個JVM進程，它的內(nèi)存管理建立在JVM的內(nèi)存管理之上。Spark采用堆內(nèi)內(nèi)存（onheapmemory）和堆外內(nèi)存（offheapmemory）的規(guī)劃機制，為了充分利用內(nèi)存，Spark對堆內(nèi)內(nèi)存進行了詳細的規(guī)劃式管理，同時在Worker的系統(tǒng)內(nèi)存中開辟空間作為堆外內(nèi)存來進一步優(yōu)化內(nèi)存的使用。3.2.1

Spark內(nèi)存管理1.內(nèi)存劃分堆內(nèi)內(nèi)存和堆外內(nèi)存示意圖如圖3-3所示。圖3-3Spark堆內(nèi)內(nèi)存和堆外內(nèi)存示意圖3.2.1

Spark內(nèi)存管理1.內(nèi)存劃分堆內(nèi)內(nèi)存包括以下4塊，默認情況下，Spark僅使用堆內(nèi)內(nèi)存。（1）ExecutionMemory01020304（2）StorageMemory（3）UserMemory（4）ReservedMemory3.2.1

Spark內(nèi)存管理2.內(nèi)存空間分配Spark1.6之前采用的是靜態(tài)內(nèi)存管理機制，即Execution內(nèi)存、Storage內(nèi)存和其他內(nèi)存的分配是靜態(tài)的，用戶在Spark應用啟動之前進行配置，在Spark應用運行期間，這些內(nèi)存都是固定的。這就導致對內(nèi)存的調(diào)優(yōu)難以進行，Execution內(nèi)存或Storage內(nèi)存任何一方內(nèi)存不足時，即使另一方有很大空閑空間也無法申請使用，內(nèi)存資源利用率不高。3.2.1

Spark內(nèi)存管理2.內(nèi)存空間分配堆內(nèi)內(nèi)存的統(tǒng)一管理示意圖如圖3-4所示。圖3-4堆內(nèi)內(nèi)存的統(tǒng)一管理示意圖3.2.1

Spark內(nèi)存管理2.內(nèi)存空間分配SystemMemory（系統(tǒng)內(nèi)存）就是通過參數(shù)executormemory或spark.executor.memory來設置堆內(nèi)內(nèi)存的。ReservedMemory（預留內(nèi)存）默認為300MB，用于防止實際內(nèi)存超出預設值而導UnifiedMemory（統(tǒng)一內(nèi)存）為Execution內(nèi)存和Storage內(nèi)存共用的內(nèi)存空間，約占UsableMemory的60%，具體占用的比例由參數(shù)spark.memory.fraction進行配置。3.2.1

Spark內(nèi)存管理2.內(nèi)存空間分配堆外內(nèi)存的統(tǒng)一管理示意圖如圖3-5所示。圖3-5堆外內(nèi)存的統(tǒng)一管理示意圖Storage內(nèi)存和Execution空間大小由參數(shù)spark.memory.storageFraction來設定各自占用的比例，默認為各占off-heap內(nèi)存的50%。堆外內(nèi)存的動態(tài)占用機制與堆內(nèi)內(nèi)存一致。3.2.2

Spark容錯機制

RDD之間的依賴關系有兩種：窄依賴（narrowdependency）和寬依賴（widedependency），如圖3-6所示。圖3-6RDD的依賴關系1.Lineage機制3.2.2

Spark容錯機制1.Lineage機制（1）窄依賴：一個父RDD的每個分區(qū)最多被一個子RDD的一個分區(qū)所使用，具體分為兩種：一個子RDD分區(qū)只依賴于一個父RDD分區(qū)，一個子RDD分區(qū)依賴于多個父RDD分區(qū)。（2）寬依賴：一個父RDD至少有一個分區(qū)被一個子RDD的多個分區(qū)所使用。3.2.2

Spark容錯機制2.Cache機制對于迭代型算法和交互式應用，可能需要對同一塊數(shù)據(jù)進行多次重復訪問，若每次都進行重新運算，則時間開銷太大。Spark采用了Cache機制來保證需要訪問重復數(shù)據(jù)的應用可以運行得更快。在計算RDD分區(qū)時判斷這個分區(qū)要不要進行緩存，如果設置了需要進行緩存，那么先將分區(qū)計算出來，然后通過BlockManager緩存到內(nèi)存中。在下一次使用該RDD分區(qū)時，Task會先通過BlockManager去內(nèi)存中查看是否有這個RDD的緩存，若有則直接讀取使用而不再重新進行計算。Cache機制適用于那些會被重復使用的但是又不太大的RDD，而對于運算時間很長或運算量很大的RDD和計算鏈過長或依賴其他RDD很多的RDD，則需要使用Checkpoint機制。3.2.2

Spark容錯機制3.Checkpoint機制RDD窄依賴的重新計算代價比較小，但是對于RDD寬依賴來說，重新計算的代價是非常大的。如圖3-7所示，左圖是窄依賴，右圖是寬依賴，如果b1分區(qū)丟失，對于左圖來說，只需要重新計算a1便可得到b1，而對于右圖來說，需要將a1、a2、a3和a4都重新計算，這就產(chǎn)生了冗余計算，因為計算a1、a2、a3和a4相當于把b2、b3和b4也重新計算了。圖3-7窄依賴和寬依賴的重新計算3.2.3

Spark監(jiān)控管理Spark提供了3種方式來對應用進行監(jiān)控：010203WebUI指標第三方監(jiān)控工具3.2.3

Spark監(jiān)控管理例如，使用sparkshell提交應用，用于統(tǒng)計文件“source.md”中包含“Spark”關鍵字的行數(shù)，代碼如下。valtextFile=spark.read.textFile("file:///hadoop/source.md")textFile.counttextFile.first()vallinesWithSpark=textFile.filter(line=>line.contains("Spark"))linesWithSpark.cache()//設置緩存linesWithSpark.count()linesWithSpark.count()程序運行后，在瀏覽器中訪問“http://master:4040”就可以查看程序的監(jiān)控信息。3.2.3

Spark監(jiān)控管理如圖3-8所示，可以看到有4個Job。圖3-8（a）展示了事件發(fā)生的時間線，從中可以直觀地看到每個步驟的時間消耗情況，如果某個步驟的時間消耗異常，就可以有針對性地對這個步驟進行調(diào)試。每個Job的具體提交時間、運行時長、Stage和Task的個數(shù)及完成情況都可以在圖3-8（b）中看到。圖中用方框標示的部分為通過WebUI可以看到的選項。圖3-8Job監(jiān)控3.2.3

Spark監(jiān)控管理圖3-9所示為Stage的監(jiān)控信息，共有7個Stage，每個Stage的提交時間、運行時長、Task的個數(shù)與完成情況、Task輸入/輸出文件大小及ShuffleWrite大小都清晰地顯示出來。圖3-9Stage監(jiān)控3.2.3

Spark監(jiān)控管理單擊Description可以看到Stage的詳細信息，圖3-10展示了Stage0的DAG，圖3-11展示了Stage0的Task和Executor的詳細信息。圖3-10Stage0的DAG圖3-11Stage0的Task和Executor的詳細信息3.2.3

Spark監(jiān)控管理圖3-12展示的是Storage監(jiān)控中某個RDD的緩存信息，該RDD占用561.9KB的堆內(nèi)內(nèi)存，具有1個分區(qū)。圖3-12某個RDD的緩存信息3.2.3

Spark監(jiān)控管理圖3-13展示了Executor的運行情況。圖3-13Executor的運行情況小結Spark的運行架構包括應用程序的控制進程DriverProgram、資源管理器ClusterManager、工作節(jié)點Worker和執(zhí)行進程Executor。SparkContext接收到用戶代碼后會執(zhí)行以下操作：構建DAG、劃分Stage、分發(fā)TaskSet、Executor以多線程方式并行執(zhí)行任務、注銷。Spark是基于內(nèi)存計算的，采用堆內(nèi)內(nèi)存和堆外內(nèi)存的規(guī)劃機制。Spark1.6之后引入了統(tǒng)一內(nèi)存管理機制，Execution內(nèi)存和Storage內(nèi)存共享同一塊空間，互相可以動態(tài)調(diào)整。Spark采用的RDD具有高效的容錯性，主要使用了Lineage、Cache和Checkpoint三種方式。Spark提供了3種方式來對應用進行監(jiān)控：WebUI、指標和第三方監(jiān)控工具，其中WebUI監(jiān)控是默認開啟的，也是最常用的方式。小結習題1.填空題（1）Spark的運行架構包括應用程序的

、

、和

。（2）Spark的基本計算單元是

。（3）Spark中使用

對RDD的關系進行建模。（4）Spark的一個重要特點是基于

計算的，因而更快。（5）Spark采用

和

的規(guī)劃機制。（6）Spark1.6之后引入了統(tǒng)一內(nèi)存管理機制，

和

共享同一塊空間，互相可以動態(tài)調(diào)整。習題2.選擇題（1）在Spark運行架構中，以下（