一、Spark提交應用任務的四個階段： 總共提交的任務分為四個階段，提交+執行： 1、在分配完畢executor以後，解析代碼生成DAG有向無環圖； 2、將生成的DAG圖提交給DAGScheduler，這個組件在driver內，DAGScheduler負責切分階段，按照DAG圖中的shuffle運算元 ...
一、Spark提交應用任務的四個階段：
總共提交的任務分為四個階段，提交+執行：
1、在分配完畢executor以後，解析代碼生成DAG有向無環圖；
2、將生成的DAG圖提交給DAGScheduler，這個組件在driver內，DAGScheduler負責切分階段，按照DAG圖中的shuffle運算元進行stage階段的切分，切分完畢階段以後，按照每個階段分別生成對應task任務的集合，將每個階段所有的task任務放入到對應的set集合中，提交這個set集合，即一次性提交每個stage階段的所有任務（每個階段準備好就提交哪個階段）
3、將任務的集合提交給taskScheduler【driver端組件】,這個組件會將數據通過集群管理器提交給集群（executor），對任務進行監控，分配資源，負責提交，負責執行，負責故障重試，負責落後任務的重啟
4、真正提交到executor端，在executor中進行執行，保存執行過後的數據，或者存儲數據
二、Spark的運行流程詳解【重點】：
♎ spark-submit 提交命令的解析：
通過查看底層，我們可以瞭解到，使用spark-submit方法提交任務的時候的時候，實際上是運行的 Spark內的 SparkSubmit 類
提交任務的命令：
spark-submit --master xxx --class xxx --name xxx xxx.jar input output
我們使用命令提交任務的時候，設置的參數實際上就是給SparkSubmit 的參數；
A、查看SparkSubmit的源碼：spark-submit 提交任務時，實際上是運行sparksubmit中的main方法
spark-submit 提交任務時，實際上是運行sparksubmit中的main方法：
所以--master xxx --class com.bw.spark.wordcount --name xxx xxx.jar input output 這些東西都是main方法的參數
♈ main方法中會接收傳入的參數，將傳入的參數解析後，使用匹配模式，判斷輸入的參數並執行不同的操作；
1、submit 提交一個應用任務：
spark-submit --master spark://master:7077 --class com.bw.spark.wordcount --name xxx xxx.jar input output
A、可以看到在 submit 方法內解析出了四個參數：①
a) childArgs == 子類的參數列表，是提交的參數列表內的 輸入參數【input】 和 輸出路徑 【output】 ，數據用於要運行的jar包內的main方法
b) childClasspath == 子類的類的路徑，是 --class 的參數【com.bw.spark】 ，數據用於SparkSubmit 類中的main方法
c) sysProps == 系統屬性，即 --master 的參數，指定集群模式【spark://master:7077】，數據用於SparkSubmit 類中的main方法
d) childMainClass == --class 參數中類內的入口（main方法）【wordcount 】，數據用於SparkSubmit 類中的main方法
由SparkSubmit類中的main方法管理並啟動自定義的類中的main方法
B、通過 runMain 方法使用解析出來的參數繼續向下執行②
C、通過反射機制，將mainClass的字元串轉換成一個主類
D、根據主類找到這個類中的main方法
E、通過反射機制執行main方法，將傳遞近來的參數放置到main方法中執行
總結：其實任務的提交就是運行main方法，解析代碼解析main方法，解析到此進入等待狀態
2、開始初始化driver端的東西，初始化上下文 ①sparkContext ②DAGScheduler ③TaskScheduler【屬於預提交的階段】
A、 SparkContext中需要初始化的組件：
B、 在提交任務的時候就要初始化的重要組件：
- DAGScheduler 【劃分DAG並按階段提交】
- TaskScheduler 【任務的向集群提交，監控執行】
- SchedulerBackEnd 【提交任務的通信組件】
1--2、根據部署的集群模式不一樣。創建不同的DAGScheduler和TaskScheduler
3、根據部署模式不一樣創建的SchedulerBackEnd也不一樣，根據資源分配不同的核數：
本地模式時，根據設置的線程數創建不同的SchedulerBackEnd；
集群模式時，不需要設置，線程數由集群管理；
3、組件的創建實例完畢以後，開始解析代碼：
driver初始化完成以後開始解析代碼，（executors已經啟動），記錄textFile從什麼位置開始讀取數據，記錄每個運算元生成rdd的數量，分區個數，邏輯，各個rdd之間的血緣關係，只有遇見真正的action運算元才開始執行，會生成DAG有向無環圖，rdd就是點，運算元就是線；
A、開始將DAG有向無環圖提交給DAGScheduler進行階段的切分：
從saveASTextFile開始進入，找到最後一步，可以發現是將任務提交給DAGScheduler，進行任務階段的切分：
B、到DAGScheduler中進行任務的切分階段，將每一個準備好的階段提交給TaskScheduler
Ⅰ、在DAGScheduler中的doOnReceive方法接收傳進DAGScheduler的任務，進行任務的處理
Ⅱ、DAGScheduler中負責將任務進行拆分，按照shuffle運算元進行拆分不同的stage
❶ 找到最後一個RDD，創建一個resultStage：
❷ 根據最後一個resultStage，找到這個result的父stage，如果父stage為空，那麼表明已經到了最開始的stage，直接進行提交
如果父stage不為空，那麼繼續調用自身以遞歸的形式進行倒序的向前排查，直到找到初始stage為止；
❸ Stage的劃分：通過最後一個rdd向前推，如果這個RDD是寬依賴就將stage+1，如果是窄依賴就將當前stage階段中的rdd+1；當每個階段都推衍完畢以後，將每個階段中的所有的task組成一個taskSet。
❹將每個階段中的所有的task組成一個taskSet集合：
每個階段匹配一下，如果是shuffleMapStage就組裝一個集合,這個集合中裝入的都是shuffleMapTask；如果是resultStage那麼這個stage中裝入的就是resultTask；
C、將任務集合提交給TaskScheduler,TaskScheduler進行任務的提交到集群中，然後執行操作，負責監控，申請資源，故障重試
TaskScheduler是一個介面：
TaskScheduler的其中一個實現類【TaskSchedulerImpl】：
實現類中存在各項組件方法，實現對任務進行各項初始化與管理，在配置好任務之後提交給Executor 執行；
ctrl+alt+b找到TaskScheduler介面的實現類【TaskSchedulerImpl】：
在taskSchedulerImpl中通過submitTasks方法將任務提交SchedulerBackEnd組件進行提交任務：
//提交任務的一個組件：
override def submitTasks(taskSet: TaskSet) {
//取出集合內所有的Task任務
val tasks = taskSet.tasks
logInfo("Adding task set " + taskSet.id + " with " + tasks.length + " tasks")
this.synchronized {
//創建一個Stage管理器，對不同的stage階段的task任務集合進行管理
val manager = createTaskSetManager(taskSet, maxTaskFailures)
//得到當前處理的stage是第幾個階段的
val stage = taskSet.stageId
val stageTaskSets =
taskSetsByStageIdAndAttempt.getOrElseUpdate(stage, new HashMap[Int, TaskSetManager])
stageTaskSets(taskSet.stageAttemptId) = manager
val conflictingTaskSet = stageTaskSets.exists { case (_, ts) =>
ts.taskSet != taskSet && !ts.isZombie
}
if (conflictingTaskSet) {
throw new IllegalStateException(s"more than one active taskSet for stage $stage:" +
s" ${stageTaskSets.toSeq.map{_._2.taskSet.id}.mkString(",")}")
}
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
if (!isLocal && !hasReceivedTask) {
starvationTimer.scheduleAtFixedRate(new TimerTask() {
override def run() {
if (!hasLaunchedTask) {
logWarning("Initial job has not accepted any resources; " +
"check your cluster UI to ensure that workers are registered " +
"and have sufficient resources")
} else {
this.cancel()
}
}
}, STARVATION_TIMEOUT_MS, STARVATION_TIMEOUT_MS)
}
hasReceivedTask = true
}
//將所有的task取出之後，使用 SchedulerBackEnd 通信方式將任務提交給executor執行
backend.reviveOffers()
}
D、SchedulerBackEnd 通信組件：接收TaskScheduler 發送的任務，將任務轉發給 executor
SchedulerBackEnd 是一個介面，這個介面存在兩個實現類，一個是本地通信的實現類，一個是負責集群通信的實現類
♈ 集群模式的通信方式：
1、接收TaskScheduler 發送的任務
2、遍歷將接收到的TaskScheduler 任務進行序列化以用於傳輸
3、通過rpc協議進行任務傳輸，到executor端
4、運行任務：【executor開始執行任務】
A、 executor的本質就是線程池
B、 Task類包裝為多線程類：
executor 執行器就是一個線程池，但是這個線程池中能夠執行的只有多線程的類，而task任務不是多線程的，所以用一個taskRunner多線程的類對task進行包裝後，Task就成為了多線程的，可以放入到線程池中運行！！！！
C、Task在executor中的運行流程：
在運行任務的時候調用taskRunner中的run方法，先進行任務的反序列化，計算時間，以及分配資源，然後交給執行器進行執行，將執行完畢的任務從Taskset中去除
D、執行器【runTask方法】：任務真正運行的地方
task 分為 shuffleMapTask 和 resultTask 兩種，都在task中調用runTask方法執行：
5、任務管理：
driver端是所有應用的老大，他會管理每一個executor中的任務執行；監聽，數據管理，任務重啟。。。。 TaskScheduler
所有的組件，以及代碼，變數等數據都在driver端，只有運行的時候會被傳送至executor端，因此在driver中運行的程式+變數，都需要被實例化
三、spark任務的四大調度
1、application
spark-submit spark-shell提交的任務就是一個應用，會生成一個application
2、job
遇見一個action運算元就會生成一個job
3、stage
遇見shuffle就會切分stage, stage 數量 = shuffle運算元數+1
4、task
運行任務的最小單位，一個stage中最後一個rdd的分區數量就是這個stage中task任務的個數
四、幾個重要的數值：
- 讀取外部文件的時候，rdd的分區數量是，這個被讀取的文件存在多少個block快就有多少個分區；但是當文件只有一個分區時，產生兩個分區；
- 每個stage中task的個數取決於最後一個rdd的分區數量
- 寫入到hdfs中的文件個數（saveAsTextFile）,是存儲的rdd的分區數量
- 一個job每個能夠運行多少個task任務？這個job區段內每個stage中的最後一個rdd的分區的總和
- 同時並行能夠運行多少task任務？集群中總核數，一個線程對應一個Task任務，如果任務數量比總的核數多，則等待執行
- 帶有shuffle的運算元切分stage，產生的依賴是寬依賴；判斷是否是寬依賴的最簡單依據就是看運算元是否會產生shuffle；
- 不帶shuffle的運算元，都在一個stage內，產生依賴是窄依賴
- 在整個應用任務流程中，action行動類運算元會產生job任務使懶載入開始執行；shuffle運算元切割stage，產生不同的pipeline管道形式的stage提交階段；兩者不是對等關係，job 的階段 ≠ stage的階段；
