Job動態生成方法是什么

本篇內容介紹了“Job動態生成方法是什么”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

在spark stream程序中的一條關鍵的語句就是：ssc.start()

1，跟蹤進入StreamingContext的start 方法，有一句非常關鍵的語句scheduler.start()，是個JobScheduler(spark stream用來job調度的)

進行job調度的入口！

2，計入JobScheduler 的start方法。

在這個方法中幾個關鍵的點是：

eventLoop.start() 一個事件循環器，用于響應其它組件發來的事件(包括job的啟動，完成，以及錯誤報告)。

receiverTracker.start() 控制了整個receiver的生成，與數據的接受

jobGenerator.start() 真正開始進行job的生成
 

在這個方法中也維護了一個事件處理的循環器eventLoop，用于處理各種事件

其中最為關鍵的事件是GenerateJobs(time)，這個事件是進行生成job的事件！！

跟蹤計入generateJobs(time)

jobScheduler.receiverTracker.allocateBlocksToBatch(time) 為當前的bath分發收到的數據Blocks。

graph.generateJobs(time)：根據當前編寫的程序的output動作生成相應的job并封裝進入集合中。

最終通過

提交作業到executor

在回去看看jobGenerator.start()中的startFirstTime()

private def startFirstTime() {

  val startTime = new Time(timer.getStartTime())

  graph.start(startTime - graph.batchDuration)

  timer.start(startTime.milliseconds)

  logInfo("Started JobGenerator at " + startTime)

}

第一次啟動會啟動一個定時器，該定時器會根基duration bath 不斷的的給jobGenerator中的消息循環體！

在jobGenerator中的消息循環體就會不斷的去除消息進行處理

“Job動態生成方法是什么”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！