解密SparkStreaming运行机制和架构进阶之Job和容错
2016-08-25 18:57
489 查看
本博文主要包括以下内容:
解密Spark Streaming Job架构和运行机制
解密Spark Streaming容错架构和运行机制
一、解密SparkStreaming Job架构和运行机制:
理解SparkStreaming的Job的整个架构和运行机制对于精通SparkStreaming是至关重要的。我们知道对于一般的Spark应用程序来说,是RDD的action操作触发了Job的运行。那对于SparkStreaming来说,Job是怎么样运行的呢?我们在编写SparkStreaming程序的时候,设置了BatchDuration,Job每隔BatchDuration时间会自动触发,这个功能肯定是SparkStreaming框架提供了一个定时器,时间一到就将编写的程序提交给Spark,并以Spark job的方式运行。
这里面涉及到两个Job的概念:
每个BatchInterval会产生一个具体的Job,其实这里的Job不是Spark Core中所指的Job,它只是基于DStreamGraph(代表RDD的依赖关系具体构成)而生成的RDD的DAG而已,从Java角度讲,相当于Runnable接口实例,此时要想运行Job需要提交给JobScheduler,在JobScheduler中通过线程池的方式找到一个单独的线程来提交Job到集群运行(其实是在线程中基于RDD的Action触发真正的作业的运行),为什么使用线程池呢?
作业不断生成,所以为了提升效率,我们需要线程池;这和在Executor中通过线程池执行Task有异曲同工之妙;
有可能设置了Job的FAIR公平调度的方式,这个时候也需要多线程的支持;
上面提交的Spark Job本身。单从这个时刻来看,此次的Job和Spark core中的Job没有任何的区别。
下面我们看看job运行的过程:
1.首先实例化SparkConf,设置运行期参数。
2.实例化StreamingContext,设置batchDuration时间间隔来控制Job生成的频率并且创建Spark Streaming执行的入口。
3,StreamingContext.scala的第183行
4,JobScheduler.scala的第50行
5,StreamingContext调用start方法。
6,在StreamingContext.start()内部启动JobScheduler的Start方法。
StreamingContext.scala第610行:
在JobScheduler.start()内部实例化EventLoop,并执行EventLoop.start()进行消息循环。
在JobScheduler.start()内部构造ReceiverTacker,并且调用JobGenerator和ReceiverTracker的start方法:
7,JobGenerator启动后会不断的根据batchDuration生成一个个的Job
8、ReceiverTracker启动后首先在Spark Cluster中启动Receiver(其实是在Executor中先启动ReceiverSupervisor),在Receiver收到数据后会通过ReceiverSupervisor存储到Executor并且把数据的Metadata信息发送给Driver中的ReceiverTracker,在ReceiverTracker内部会通过ReceivedBlockTracker来管理接受到的元数据信息。
二、解密SparkStreaming容错架构和运行机制:
1、运行过程总结如下:
在StreamingContext调用start方法的内部其实是会启动JobScheduler的Start方法,进行消息循环,在JobScheduler
的start内部会构造JobGenerator和ReceiverTacker,并且调用JobGenerator和ReceiverTacker的start方法:
JobGenerator启动后会不断的根据batchDuration生成一个个的Job
ReceiverTracker启动后首先在Spark Cluster中启动Receiver(其实是在Executor中先启动ReceiverSupervisor),在Receiver收到
数据后会通过ReceiverSupervisor存储到Executor并且把数据的Metadata信息发送给Driver中的ReceiverTracker,在ReceiverTracker
内部会通过ReceivedBlockTracker来管理接受到的元数据信息
2、Spark Streaming容错机制:
我们知道DStream与RDD的关系就是随着时间流逝不断的产生RDD,对DStream的操作就是在固定时间上操作RDD。所以从某种意义上而言,Spark Streaming的基于DStream的容错机制,实际上就是划分到每一次形成的RDD的容错机制,这也是Spark Streaming的高明之处。
Spark Streaming的容错要考虑两个方面:
Driver运行失败时的恢复
使用Checkpoint,记录Driver运行时的状态,失败后可以读取Checkpoint并恢复Driver状态。
具体的每次Job运行失败时的恢复
要考虑到Receiver的失败恢复,也要考虑到RDD计算失败的恢复。Receiver可以采用写wal日志的方式。RDD的容错是spark core天生提供的,基于RDD的特性,它的容错机制主要就是两种:
01. 基于checkpoint;
在stage之间,是宽依赖,产生了shuffle操作,lineage链条过于复杂和冗长,这时候就需要做checkpoint。
02. 基于lineage(血统)的容错:
一般而言,spark选择血统容错,因为对于大规模的数据集,做检查点的成本很高。考虑到RDD的依赖关系,每个stage内部都是窄依赖,此时一般基于lineage容错,方便高效。
总结: stage内部做lineage,stage之间做checkpoint。
博文内容源自DT大数据梦工厂Spark课程的总结笔记。相关课程内容视频可以参考:
百度网盘链接:http://pan.baidu.com/s/1slvODe1(如果链接失效或需要后续的更多资源,请联系QQ460507491或者微信号:DT1219477246 获取上述资料)。
解密Spark Streaming Job架构和运行机制
解密Spark Streaming容错架构和运行机制
一、解密SparkStreaming Job架构和运行机制:
理解SparkStreaming的Job的整个架构和运行机制对于精通SparkStreaming是至关重要的。我们知道对于一般的Spark应用程序来说,是RDD的action操作触发了Job的运行。那对于SparkStreaming来说,Job是怎么样运行的呢?我们在编写SparkStreaming程序的时候,设置了BatchDuration,Job每隔BatchDuration时间会自动触发,这个功能肯定是SparkStreaming框架提供了一个定时器,时间一到就将编写的程序提交给Spark,并以Spark job的方式运行。
这里面涉及到两个Job的概念:
每个BatchInterval会产生一个具体的Job,其实这里的Job不是Spark Core中所指的Job,它只是基于DStreamGraph(代表RDD的依赖关系具体构成)而生成的RDD的DAG而已,从Java角度讲,相当于Runnable接口实例,此时要想运行Job需要提交给JobScheduler,在JobScheduler中通过线程池的方式找到一个单独的线程来提交Job到集群运行(其实是在线程中基于RDD的Action触发真正的作业的运行),为什么使用线程池呢?
作业不断生成,所以为了提升效率,我们需要线程池;这和在Executor中通过线程池执行Task有异曲同工之妙;
有可能设置了Job的FAIR公平调度的方式,这个时候也需要多线程的支持;
上面提交的Spark Job本身。单从这个时刻来看,此次的Job和Spark core中的Job没有任何的区别。
下面我们看看job运行的过程:
1.首先实例化SparkConf,设置运行期参数。
val conf = new SparkConf().setAppName("--")
2.实例化StreamingContext,设置batchDuration时间间隔来控制Job生成的频率并且创建Spark Streaming执行的入口。
val ssc = new StreamingContext(conf,Seconds(20))
3,StreamingContext.scala的第183行
private[streaming] val scheduler = new JobScheduler(this)
4,JobScheduler.scala的第50行
private val jobGenerator = new JobGenerator(this)
5,StreamingContext调用start方法。
def start(): Unit = synchronized { state match { case INITIALIZED => startSite.set(DStream.getCreationSite()) StreamingContext.ACTIVATION_LOCK.synchronized { StreamingContext.assertNoOtherContextIsActive() try { validate() // Start the streaming scheduler in a new thread, so that thread local properties // like call sites and job groups can be reset without affecting those of the // current thread. ThreadUtils.runInNewThread("streaming-start") { sparkContext.setCallSite(startSite.get) sparkContext.clearJobGroup() sparkContext.setLocalProperty(SparkContext.SPARK_JOB_INTERRUPT_ON_CANCEL, "false") scheduler.start() } state = StreamingContextState.ACTIVE } catch { case NonFatal(e) => logError("Error starting the context, marking it as stopped", e) scheduler.stop(false) state = StreamingContextState.STOPPED throw e } StreamingContext.setActiveContext(this) } shutdownHookRef = ShutdownHookManager.addShutdownHook( StreamingContext.SHUTDOWN_HOOK_PRIORITY)(stopOnShutdown) // Registering Streaming Metrics at the start of the StreamingContext assert(env.metricsSystem != null) env.metricsSystem.registerSource(streamingSource) uiTab.foreach(_.attach()) logInfo("StreamingContext started") case ACTIVE => logWarning("StreamingContext has already been started") case STOPPED => throw new IllegalStateException("StreamingContext has already been stopped") } }
6,在StreamingContext.start()内部启动JobScheduler的Start方法。
StreamingContext.scala第610行:
scheduler.start()
在JobScheduler.start()内部实例化EventLoop,并执行EventLoop.start()进行消息循环。
在JobScheduler.start()内部构造ReceiverTacker,并且调用JobGenerator和ReceiverTracker的start方法:
def start(): Unit = synchronized { if (eventLoop != null) return // scheduler has already been started logDebug("Starting JobScheduler") eventLoop = new EventLoop[JobSchedulerEvent]("JobScheduler") { override protected def onReceive(event: JobSchedulerEvent): Unit = processEvent(event) override protected def onError(e: Throwable): Unit = reportError("Error in job scheduler", e) } eventLoop.start() // attach rate controllers of input streams to receive batch completion updates for { inputDStream <- ssc.graph.getInputStreams rateController <- inputDStream.rateController } ssc.addStreamingListener(rateController) listenerBus.start(ssc.sparkContext) receiverTracker = new ReceiverTracker(ssc) inputInfoTracker = new InputInfoTracker(ssc) receiverTracker.start() jobGenerator.start() logInfo("Started JobScheduler") }
7,JobGenerator启动后会不断的根据batchDuration生成一个个的Job
/** Generate jobs and perform checkpoint for the given `time`. */ private def generateJobs(time: Time) { // Set the SparkEnv in this thread, so that job generation code can access the environment // Example: BlockRDDs are created in this thread, and it needs to access BlockManager // Update: This is probably redundant after threadlocal stuff in SparkEnv has been removed. SparkEnv.set(ssc.env) Try { jobScheduler.receiverTracker.allocateBlocksToBatch(time) // allocate received blocks to batch graph.generateJobs(time) // generate jobs using allocated block } match { case Success(jobs) => val streamIdToInputInfos = jobScheduler.inputInfoTracker.getInfo(time) jobScheduler.submitJobSet(JobSet(time, jobs, streamIdToInputInfos)) case Failure(e) => jobScheduler.reportError("Error generating jobs for time " + time, e) } eventLoop.post(DoCheckpoint(time, clearCheckpointDataLater = false)) }
8、ReceiverTracker启动后首先在Spark Cluster中启动Receiver(其实是在Executor中先启动ReceiverSupervisor),在Receiver收到数据后会通过ReceiverSupervisor存储到Executor并且把数据的Metadata信息发送给Driver中的ReceiverTracker,在ReceiverTracker内部会通过ReceivedBlockTracker来管理接受到的元数据信息。
/** Start the endpoint and receiver execution thread. */ def start(): Unit = synchronized { if (isTrackerStarted) { throw new SparkException("ReceiverTracker already started") } if (!receiverInputStreams.isEmpty) { endpoint = ssc.env.rpcEnv.setupEndpoint( "ReceiverTracker", new ReceiverTrackerEndpoint(ssc.env.rpcEnv)) if (!skipReceiverLaunch) launchReceivers() logInfo("ReceiverTracker started") trackerState = Started } }
二、解密SparkStreaming容错架构和运行机制:
1、运行过程总结如下:
在StreamingContext调用start方法的内部其实是会启动JobScheduler的Start方法,进行消息循环,在JobScheduler
的start内部会构造JobGenerator和ReceiverTacker,并且调用JobGenerator和ReceiverTacker的start方法:
JobGenerator启动后会不断的根据batchDuration生成一个个的Job
ReceiverTracker启动后首先在Spark Cluster中启动Receiver(其实是在Executor中先启动ReceiverSupervisor),在Receiver收到
数据后会通过ReceiverSupervisor存储到Executor并且把数据的Metadata信息发送给Driver中的ReceiverTracker,在ReceiverTracker
内部会通过ReceivedBlockTracker来管理接受到的元数据信息
2、Spark Streaming容错机制:
我们知道DStream与RDD的关系就是随着时间流逝不断的产生RDD,对DStream的操作就是在固定时间上操作RDD。所以从某种意义上而言,Spark Streaming的基于DStream的容错机制,实际上就是划分到每一次形成的RDD的容错机制,这也是Spark Streaming的高明之处。
Spark Streaming的容错要考虑两个方面:
Driver运行失败时的恢复
使用Checkpoint,记录Driver运行时的状态,失败后可以读取Checkpoint并恢复Driver状态。
具体的每次Job运行失败时的恢复
要考虑到Receiver的失败恢复,也要考虑到RDD计算失败的恢复。Receiver可以采用写wal日志的方式。RDD的容错是spark core天生提供的,基于RDD的特性,它的容错机制主要就是两种:
01. 基于checkpoint;
在stage之间,是宽依赖,产生了shuffle操作,lineage链条过于复杂和冗长,这时候就需要做checkpoint。
02. 基于lineage(血统)的容错:
一般而言,spark选择血统容错,因为对于大规模的数据集,做检查点的成本很高。考虑到RDD的依赖关系,每个stage内部都是窄依赖,此时一般基于lineage容错,方便高效。
总结: stage内部做lineage,stage之间做checkpoint。
博文内容源自DT大数据梦工厂Spark课程的总结笔记。相关课程内容视频可以参考:
百度网盘链接:http://pan.baidu.com/s/1slvODe1(如果链接失效或需要后续的更多资源,请联系QQ460507491或者微信号:DT1219477246 获取上述资料)。
相关文章推荐
- HBase的Write Ahead Log (WAL) —— 整体架构、线程模型
- HBase的Write Ahead Log (WAL) —— 整体架构、线程模型
- MVC架构设计——EF-Code First
- 逃离x86架构-----CPU体系结构CISC与RISC之争
- [置顶] 关于网站多语言机制的实现的原型demo
- 中国这10家慕课网站,您需要知道!
- [置顶] 抛弃框架的基于纯 Java Servlet 与 JSP的网站系统开发
- 关于高可用的系统
- [置顶] 网站unicode字符乱码问题一扫空解决方案 -- 带demo 简单的小说应用
- Magento与大型网站技术架构
- [置顶] 网站浏览统计功能的简单实现-- vb2005xu自己动手系列(2)
- 大学生应该常去的网站
- 构建高并发网站-网站架构(架构优化)
- 部级网站的bug: 12306 几亿的东西如此狼狈
- 12306网站的体验太棒了!和去售票窗口一样,让你体验排队的乐趣
- 关于 瀑布流布局网站 的一点记录
- 标记一个好的应用了VimWiki的网站
- 今天访问不了国内网站了,咋回事呢?
- 新公司网站[数字星空-团]上线 -- 兴奋莫名
- 公司网站的样例代码基本