在前面的章節(jié)Client的加載中，Spark的DriverRunner已開始執(zhí)行用戶任務(wù)類（比如：org.apache.spark.examples.SparkPi），下面我們開始針對于用戶任務(wù)類（或者任務(wù)代碼）進行分析

 

一、整體預(yù)覽

          基于上篇圖做了擴展，增加任務(wù)執(zhí)行的相關(guān)交互

     iOS培訓(xùn),Swift培訓(xùn),蘋果開發(fā)培訓(xùn),移動開發(fā)培訓(xùn)
iOS培訓(xùn),Swift培訓(xùn),蘋果開發(fā)培訓(xùn),移動開發(fā)培訓(xùn)

  • Code：指的用戶編寫的代碼

  • RDD：彈性分布式數(shù)據(jù)集，用戶編碼根據(jù)SparkContext與RDD的api能夠很好的將Code轉(zhuǎn)化為RDD數(shù)據(jù)結(jié)構(gòu)（下文將做轉(zhuǎn)化細節(jié)介紹）

  • DAGScheduler：有向無環(huán)圖調(diào)度器，將RDD封裝為JobSubmitted對象存入EventLoop（實現(xiàn)類DAGSchedulerEventProcessLoop）隊列中

  • EventLoop： 定時掃描未處理JobSubmitted對象，將JobSubmitted對象提交給DAGScheduler

  • DAGScheduler：針對于JobSubmitted進行處理，最終將RDD轉(zhuǎn)化為執(zhí)行TaskSet,并將TaskSet提交至TaskScheduler

  • TaskScheduler： 根據(jù)TaskSet創(chuàng)建TaskSetManager對象存入SchedulableBuilder的數(shù)據(jù)池（Pool）中，并調(diào)用DriverEndpoint喚起消費（ReviveOffers）操作

  • DriverEndpoint：接受ReviveOffers指令后將TaskSet中的Tasks根據(jù)相關(guān)規(guī)則均勻分配給Executor

  • Executor：啟動一個TaskRunner執(zhí)行一個Task

 

二、Code轉(zhuǎn)化為初始RDDs

          我們的用戶代碼通過調(diào)用Spark的Api（比如：SparkSession.builder.appName("Spark Pi").getOrCreate()），該Api會創(chuàng)建Spark的上下文（SparkContext）,當我們調(diào)用transform類方法 （如：parallelize(),map()）都會創(chuàng)建（或者裝飾已有的） Spark數(shù)據(jù)結(jié)構(gòu)（RDD）, 如果是action類操作（如：reduce()），那么將最后封裝的RDD作為一次Job提交，存入待調(diào)度隊列中（DAGSchedulerEventProcessLoop ）待后續(xù)異步處理。

          如果多次調(diào)用action類操作，那么封裝的多個RDD作為多個Job提交。

     流程如下：

     iOS培訓(xùn),Swift培訓(xùn),蘋果開發(fā)培訓(xùn),移動開發(fā)培訓(xùn)iOS培訓(xùn),Swift培訓(xùn),蘋果開發(fā)培訓(xùn),移動開發(fā)培訓(xùn)

  • ExecuteEnv（執(zhí)行環(huán)境 ）


    • 這里可以是通過spark-submit提交的MainClass，也可以是spark-shell腳本

    • MainClass : 代碼中必定會創(chuàng)建或者獲取一個SparkContext

    • spark-shell：默認會創(chuàng)建一個SparkContext

  • RDD（彈性分布式數(shù)據(jù)集）


    • create：可以直接創(chuàng)建（如：sc.parallelize(1 until n, slices) ）,也可以在其他地方讀?。ㄈ纾簊c.textFile("README.md")）等

      網(wǎng)友評論