spark任務運行過程的源碼分析

spark任務運行的源碼分析

在整個spark任務的編寫、提交、執行分三個部分：
① 編寫程序和提交任務到集群中
②sparkContext的初始化
③觸發action算子中的runJob方法，執行任務

(1)編程程序并提交到集群：

①編程spark程序的代碼
②打成jar包到集群中運行
③使用spark-submit命令提交任務
在提交任務時，需要指定 --class 程序的入口（有main方法的類），
1） spark-submit --class xxx
2） ${SPARK_HOME}/bin/spark-class org.apache.spark.deploy.SparkSubmit $@
3） org.apache.spark.launcher.Main
submit(appArgs, uninitLog)
doRunMain()
runMain(childArgs, childClasspath, sparkConf, childMainClass, args.verbose)
childMainClass：…./.WordCount （自己編寫的代碼的主類）
mainClass = Utils.classForName(childMainClass)
val app: SparkApplication = if() {} else {new JavaMainApplication(mainClass)}
app.start(childArgs.toArray, sparkConf) // 通過反射調用mainClass執行
// 到此為止，相當于調用了我們的自己編寫的任務類的main方法執行了。！！！
val mainMethod = klass.getMethod("main", new ArrayString.getClass)
mainMethod.invoke(null, args)
④開始執行自己編寫的代碼

(2)初始化sparkContext：

當自己編寫的程序運行到：new SparkContext()時，就開始了精妙而細致的sparkContext的初始化。
sparkContext的相關介紹：sparkContext是用戶通往spark集群的唯一入口，可以用來在spark集群中創建RDD、累加器和廣播變量。sparkContext也是整個spark應用程序的一個至關重要的對象，是整個應用程序運行調度的核心（不是資源調度的核心）。在初始化sparkContext時，同時的會初始化DAGScheduler、TaskScheduler和SchedulerBackend，這些至關重要的對象。
sparkContext的構建過程：

1)Driver端執行的代碼：

 初始化 TaskScheduler
 初始化 SchedulerBackend
 初始化 DAGScheduler

2)worker和master端執行的代碼：

  driver向master注冊申請資源。
  Worker負責啟動executor。

(3)觸發action算子中的runJob方法：

spark任務運行總結：

• 將編寫的程序打成jar包
• 調用spark-submit提交任務到集群上運行
• 運行sparkSubmit 的main方法，在這個方法中通過反射的方式創建我們編寫的主類的實例對象，然后調用該對象的main方法，開始執行我們編寫的代碼
• 當代碼運行到new SparkContext對象的的時候，就開始了復雜和精致的sparkContext對象的初始化
• 在初始化SparkContext對象的時候，會創建兩個特別重要的對象，分別是：DAGScheduler 和 TaskScheduler，其中【DAGScheduler 的作用】將RDD的依賴切成一個一個的stage，然后stage作為taskSet提交給Taskscheduler。
• 在構建TaskScheduler的同時，會創建兩個非常重要的對象，分別是 DriverActor 和 ClientActor，DriverActor負責接收executor的反向注冊，將任務提交給executor運行，clientActor是負責向master注冊并提交任務
• 當clientActor啟動時，會將用戶提交的任務相關的參數分裝到applicationDescription對象中去，然后提交給master進行任務注冊
• 當master接收到clientActor提交的任務請求時，會將請求的參數進行分析，并封裝成application，然后將其持久化，然后將其加入到任務隊列waitingApps中。
• 當輪到我們提交任務的時候，就開始執行schedule()，進行任務資源的調度
• worker接收到master發送來的launchExecutor 時，會將其解壓并封裝到ExecutorRunner中，然后調用這個對象的start方法，啟動executor
• executor啟動后會向driver反向注冊
• driver會發送注冊成功信息，給executor
• executor接收到driver actor注冊成功信息后，就會創建一個線程池，用于執行driveractor發送過來的任務
• 當屬于這個任務的所有的 Executor 啟動并反向注冊成功后，就意味著運行這個任務的 環境已經準備好了，driver 會結束 SparkContext 對象的初始化，也就意味著 new SparkContext 這句代碼運行完成
• 當sparkContext初始化完成之后，就會繼續運行我們的代碼，直到運行到action算子時，也就意味著觸發了一個job的提交
• driver 會將這個 job 提交給 DAGScheduler
• DAGScheduler將接收到的job，從最后一個算子開始推導，將DAG根據依賴關系劃分成為一個個stage，然后將stage封裝成一個taskSet，并將taskSet中的task提交給taskScheduler
• taskScheduler接收到DAGScheduler發送過來的task，會拿到一個序列化器，對task進行序列化，然后將序列化好的task封裝到launchTask中，然后將launchTask發送給指定的executor中運行
• executor接收到了DriverActor 發送過來的launchTask 時，會拿到一個反序列化器，對launchTask 進行反序列化，封裝到一個TaskRunner 中，然后從executor這個線程池中獲取一個線程，將反序列化好的任務中的算子作用在RDD對應的分區上。
• 最終當所有的task任務完成之后，整個application執行完成，關閉sparkContext對象。