RELATEED CONSULTING
相关咨询
选择下列产品马上在线沟通
服务时间:8:30-17:00
你可能遇到了下面的问题
关闭右侧工具栏

新闻中心

这里有您想知道的互联网营销解决方案
Job动态生成方法是什么

本篇内容介绍了“Job动态生成方法是什么”的有关知识,在实际案例的操作过程中,不少人都会遇到这样的困境,接下来就让小编带领大家学习一下如何处理这些情况吧!希望大家仔细阅读,能够学有所成!

成都创新互联公司长期为上千家客户提供的网站建设服务,团队从业经验10年,关注不同地域、不同群体,并针对不同对象提供差异化的产品和服务;打造开放共赢平台,与合作伙伴共同营造健康的互联网生态环境。为临潼企业提供专业的成都做网站、网站设计,临潼网站改版等技术服务。拥有十年丰富建站经验和众多成功案例,为您定制开发。

在spark stream程序中的一条关键的语句就是:ssc.start()

1,跟踪进入StreamingContext的start 方法,有一句非常关键的语句scheduler.start(),是个JobScheduler(spark stream用来job调度的)

进行job调度的入口!

2,计入JobScheduler 的start方法。

Job动态生成方法是什么

在这个方法中几个关键的点是:

eventLoop.start() 一个事件循环器,用于响应其它组件发来的事件(包括job的启动,完成,以及错误报告)。

receiverTracker.start() 控制了整个receiver的生成,与数据的接受

jobGenerator.start() 真正开始进行job的生成
 

在这个方法中也维护了一个事件处理的循环器eventLoop,用于处理各种事件

其中最为关键的事件是GenerateJobs(time),这个事件是进行生成job的事件!!

跟踪计入generateJobs(time)

Job动态生成方法是什么

jobScheduler.receiverTracker.allocateBlocksToBatch(time) 为当前的bath分发收到的数据Blocks。

graph.generateJobs(time):根据当前编写的程序的output动作生成相应的job并封装进入集合中。

Job动态生成方法是什么

最终通过

提交作业到executor

Job动态生成方法是什么

在回去看看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动态生成方法是什么

“Job动态生成方法是什么”的内容就介绍到这里了,感谢大家的阅读。如果想了解更多行业相关的知识可以关注创新互联网站,小编将为大家输出更多高质量的实用文章!


文章名称:Job动态生成方法是什么
本文路径:http://lswzjz.com/article/pesjid.html