淺談Spark內(nèi)部運(yùn)行機(jī)制

1. Spark中最重要的機(jī)制有那些？

   1.RDD，2.Spark調(diào)度機(jī)制，3Shuffle過(guò)程

2. 什么是RDD?

   可以這么說(shuō)，你懂了RDD，基本上就可以對(duì)Hadoop和Spark的一半給吃透了，那么到底是RDD

   RDD(彈性分布式數(shù)據(jù)集）首先體現(xiàn)數(shù)據(jù)集，RDD是對(duì)原始數(shù)據(jù)的封裝，該種數(shù)據(jù)結(jié)構(gòu)內(nèi)部可以對(duì)數(shù)據(jù)進(jìn)行邏輯分區(qū)，其次分布式體現(xiàn)是并行計(jì)算以及需要解決容錯(cuò)問(wèn)題，也就是根據(jù)依賴(lài)，找到第一層RDD，最后根據(jù)RDD編號(hào)與分區(qū)編號(hào)，可以唯一確定該分區(qū)對(duì)應(yīng)的塊編號(hào)，就能從存儲(chǔ)介質(zhì)中提取出分區(qū)對(duì)應(yīng)的數(shù)據(jù)。在就是彈性，RDD在可以不改變內(nèi)部存儲(chǔ)數(shù)據(jù)記錄的前提下，去調(diào)整并行計(jì)算單元的劃分結(jié)構(gòu)（這個(gè)可能就是Stage）

3. 基本概念

   （1）應(yīng)用程序：用戶(hù)構(gòu)建的Spark應(yīng)用程序，包含驅(qū)動(dòng)程序（一個(gè)Driver功能的代碼）和在集群的多個(gè)工作結(jié)點(diǎn)上運(yùn)行的Executor代碼。

   （2）驅(qū)動(dòng)程序：包含main入口函數(shù)并在main函數(shù)內(nèi)實(shí)例化SparkContext對(duì)象的應(yīng)用程序稱(chēng)為驅(qū)動(dòng)應(yīng)用程序。不說(shuō)了，直接上代碼如下：

   Var logFile="YOUR_SPARK_HOME/README.md"http://本地文件目錄
   val conf=new SparkConf().setAppName("Simple Application");//給Application命名
   val sc=new SparkContext(conf);

   （3）Master（ClusterManager）管理者整個(gè)集群，目前Spark主要支持三種類(lèi)型：Standlone模式，Mesos模式，Yarn模式。

   （4）Worker節(jié)點(diǎn)：運(yùn)行Worker守護(hù)進(jìn)程的集群結(jié)點(diǎn)。

    

   （5）任務(wù)執(zhí)行器（Executor）:一個(gè)Worker節(jié)點(diǎn)上可能有多個(gè)Executor, 每個(gè)Executor都擁有固定的核心數(shù)量和堆棧大小。

   （6）作業(yè)(job)::包含多個(gè)Task(任務(wù))組成的并行計(jì)算（并排的那些分區(qū)）），往往由Spark的action觸發(fā)產(chǎn)生。在Spark中通過(guò)runJob方法向Spark集群中提交Job

   （7）階段（Stage）：每個(gè)job會(huì)因?yàn)镽DD之間的依賴(lài)關(guān)系被拆分成多個(gè)Task集合，其名稱(chēng)稱(chēng)為Stage，每一個(gè)Task集合，也可以叫TaskSet（任務(wù)集）

   補(bǔ)充：

   每個(gè)Application中可能有多個(gè)job，相互獨(dú)立。

   每個(gè)Worker可以起一個(gè)或多個(gè)Executor。
   每個(gè)Executor由若干core組成，每個(gè)Executor的每個(gè)core一次只能執(zhí)行一個(gè)Task。
   每個(gè)Task執(zhí)行的結(jié)果就是生成了目標(biāo)RDD的一個(gè)partiton。

4. 依賴(lài)于并行計(jì)算如何理解？

   4.1分區(qū)是并行計(jì)算的基本單位：一個(gè)原始數(shù)據(jù)分成了10個(gè)分區(qū)，那么就可以同時(shí)并行這個(gè)10分區(qū)，是不是可以這樣去理解？不一定，如果都是窄依賴(lài)，沒(méi)有問(wèn)題，但其中會(huì)涉及到寬依賴(lài)，這其中就會(huì)產(chǎn)生分區(qū)與分區(qū)之間的數(shù)據(jù)進(jìn)行交叉，反正不像同時(shí)完這10個(gè)分區(qū)數(shù)據(jù)這么快。

   4.2每個(gè)分區(qū)內(nèi)數(shù)據(jù)的計(jì)算當(dāng)成一個(gè)并行任務(wù)，每個(gè)并行任務(wù)包含一個(gè)計(jì)算鏈，每一個(gè)CPU核心就去執(zhí)行這些計(jì)算連。直接，簡(jiǎn)單，干脆，不玩虛的，上代碼理解計(jì)算鏈：

   rdd.map(line=>line.length).filter().等等之類(lèi)的。 

   如果這些計(jì)算鏈之間都是獨(dú)立的，而且互不影響，那么我們可以并行計(jì)算。我們可以將這些鏈條之間的關(guān)系定義為窄依賴(lài)（一對(duì)一依賴(lài)和范圍依賴(lài)）

5. RDD為什么要?jiǎng)澐諷tage，怎么劃分stage？

   如果子RDD一個(gè)分區(qū)內(nèi)的數(shù)據(jù)依賴(lài)于多個(gè)父RDD中分區(qū)的數(shù)據(jù)，這個(gè)叫做寬依賴(lài)，或者叫做Shuffle依賴(lài)，那么如果有多個(gè)子RDD，每個(gè)子RDD都依賴(lài)多個(gè)父RDD中分區(qū)的數(shù)據(jù)，我們是不是要想辦法把RDD數(shù)據(jù)保存起來(lái)，提供給這些子分區(qū)計(jì)算使用，否則是不是每個(gè)分區(qū)都要重新計(jì)算多個(gè)父RDD數(shù)據(jù)，也在這個(gè)地方開(kāi)始劃分Stage的原因。凡是遇到寬依賴(lài)，就劃分stage。

6. Spark如何管理資源？

    

7. Spark集群管理器分為三種，Standlone模式，Mesos模式，Yarn模式。這是重點(diǎn)，但又不是很重要，所以這地方不是十分了解，也沒(méi)有多大損失。

8. Spark內(nèi)部如何調(diào)度？

   DAGScheduler是面向Stage的任務(wù)調(diào)度器，負(fù)責(zé)接收Spark應(yīng)用提交的Job，根據(jù)RDD的依賴(lài)關(guān)系劃分Stage，并提交Stage給TaskScheduler

   TaskScheduler是面向Task的任務(wù)調(diào)度器，它接受DAGScheduler提交過(guò)來(lái)的TaskSets,然后把一個(gè)個(gè)Task提交到Work結(jié)點(diǎn)運(yùn)行，每個(gè)Executor運(yùn)行什么Task也是在此處分配的。

   最重要的就是這張圖了：

9. 

    (1)任何的Spark應(yīng)用程序都包含Driver和Executor代碼。Spark應(yīng)用程序首先在Driver初始化SparkContext。因?yàn)镾parkContext是Spark應(yīng)用程序通往集群的唯一途徑。在SparkContext里面包含了兩個(gè)調(diào)度器，一個(gè)是DAGScheduler和TaskScheduler，在創(chuàng)建SparkContext對(duì)象的同時(shí)也自動(dòng)創(chuàng)建了這兩個(gè)類(lèi)。

        （2）SparkContext初始化完成后，首先根據(jù)Spark的相關(guān)配置，想Cluster Master申請(qǐng)所需要的資源，然后在各個(gè)Worker結(jié)點(diǎn)初始化相應(yīng)的Executor。Executor初始化完成后，Driver將通過(guò)對(duì)Spark應(yīng)用程序中的RDD代碼進(jìn)行解析，生成相應(yīng)的RDD graph(RDD圖），該圖描述了RDD的相關(guān)信息及彼此之間的依賴(lài)關(guān)系。即是圖中第一個(gè)部分，這些RDD Objects

        （3）RDD圖構(gòu)建完畢后，Driver將提交給DAGScheduler進(jìn)行解析。DAGScheduler在解析RDD圖的過(guò)程中，當(dāng)遇到Action算子后將進(jìn)行逆向解析，根據(jù)RDD之間的依賴(lài)關(guān)系，以及是否存在Shuffle，將RDD圖解析成一系列具有先后依賴(lài)關(guān)系的Stage。Stage以shuffle進(jìn)行劃分，即如果兩個(gè)RDD之間存在依賴(lài)關(guān)系，DAGScheduler將會(huì)在這RDD之間拆分為兩個(gè)Stage進(jìn)行執(zhí)行，且只有前一個(gè)Stage執(zhí)行完畢之后，才執(zhí)行后一個(gè)Stage。

        （4）DAGScheduler將劃分的一系列的Stage（TaskSet），按照Stage的先后順序依次提交給底層的調(diào)度器TaskScheduler執(zhí)行。

        （5）TaskScheduler接收到DAGScheduler的stage任務(wù)后，將會(huì)在集群環(huán)境中構(gòu)建一個(gè)TaskSetManager實(shí)例來(lái)管理Stage（TaskSet）的生命周期。

            (6)TaskSetManager將會(huì)把相關(guān)的計(jì)算代碼，數(shù)據(jù)資源文件等發(fā)送到相應(yīng)的Executor上，并在相應(yīng)的Executor上啟動(dòng)線(xiàn)程池執(zhí)行。

          （7）在Task執(zhí)行的過(guò)程中，可能有部分應(yīng)用程序涉及到I/0的輸入輸出，在每個(gè)Executor由相應(yīng)的BlockManager進(jìn)行管理，相關(guān)BlockManager的信息將會(huì)與Driver中的Blocktracker進(jìn)行交互和同步。

            （8）在TaskThreads執(zhí)行的過(guò)程中，如果存在運(yùn)行錯(cuò)誤，或其他影響的問(wèn)題導(dǎo)致失敗，TaskSetManager將會(huì)默認(rèn)嘗試3次，嘗試均失敗后將上報(bào)TaskScheduler，TaskScheduler如果解決不了，在上報(bào)DAGScheduler，DAGScheduler將根據(jù)各個(gè)Worker結(jié)點(diǎn)的運(yùn)行情況重新提交到別Executor中執(zhí)行。

        （9）TaksThread執(zhí)行完畢后，將把執(zhí)行的結(jié)果反饋給TaskSetManager，TaskSetManager反饋給TaskScheduler，TaskScheduler在上報(bào)DAGScheduler，DAGScheduler將根據(jù)是否還存在待執(zhí)行的的Stage，將繼續(xù)循環(huán)迭代提交給TaskScheduler去執(zhí)行。

        （10）待所有的Stage都執(zhí)行完畢后，將會(huì)最終達(dá)到應(yīng)用程序的目標(biāo)，或者輸出到文件，或者在屏幕顯示等，Driver的本次運(yùn)行過(guò)程結(jié)束，等待用戶(hù)的其他指令或者關(guān)閉。

        （11）在用戶(hù)顯示關(guān)閉SparkContext,整個(gè)運(yùn)行過(guò)程結(jié)束，相關(guān)的資源或被釋放，或被回收。

Spark這種運(yùn)行形式有利于不同Application之間的資源調(diào)度，同時(shí)也就意味著不同的Application無(wú)法做到相互通信和信息交互。

            Driver負(fù)責(zé)所有任務(wù)調(diào)度，所以他應(yīng)該盡可能地靠近Worker結(jié)點(diǎn)，能在同一個(gè)網(wǎng)絡(luò)中最后了。

10.Shuffle是怎么個(gè)過(guò)程？

只有當(dāng)Shuffle依賴(lài)中父RDD所有分區(qū)的數(shù)據(jù)被計(jì)算和存儲(chǔ)完畢后，子RDD才會(huì)開(kāi)始拉取需要的分區(qū)數(shù)據(jù)。這里將整個(gè)數(shù)據(jù)傳輸?shù)倪^(guò)程稱(chēng)為Spark的Shuffle過(guò)程。在Shuffle過(guò)程中，把一個(gè)分區(qū)數(shù)據(jù)計(jì)算完畢到數(shù)據(jù)被寫(xiě)入到磁盤(pán)的過(guò)程，稱(chēng)為Shuffle寫(xiě)過(guò)程。對(duì)應(yīng)的，在子RDD某個(gè)分區(qū)計(jì)算的過(guò)程中，把所需的數(shù)據(jù)從父RDD拉取過(guò)來(lái)的過(guò)程，稱(chēng)為Shuffle讀過(guò)程。

不論是Spark還是Hadoop，在對(duì)待shuffle的過(guò)程中有著諸多類(lèi)似，一些概念可以直接套用，例如shuffle過(guò)程中，提供數(shù)據(jù)的一端稱(chēng)作map端， map端生成的任務(wù)稱(chēng)為mapper.對(duì)應(yīng)的，接受數(shù)據(jù)的一端稱(chēng)作reduce端，reduce端每個(gè)拉取數(shù)據(jù)的任務(wù)稱(chēng)為reducer。Shuffle過(guò)程的本質(zhì)是將map端獲得的數(shù)據(jù)使用分區(qū)器進(jìn)行劃分，并將數(shù)據(jù)發(fā)送給對(duì)應(yīng)的reducer的過(guò)程。