本篇内容主要讲解“Driver容错安全性是什么”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“Driver容错安全性是什么”吧!
成都创新互联公司专注于企业营销型网站、网站重做改版、巴里坤哈萨克网站定制设计、自适应品牌网站建设、HTML5建站、商城网站建设、集团公司官网建设、成都外贸网站制作、高端网站制作、响应式网页设计等建站业务,价格优惠性价比高,为巴里坤哈萨克等各大城市提供网站开发制作服务。
从数据层面,ReceivedBlockTracker为整个Spark Streaming应用程序记录元数据信息。
从调度层面,DStreamGraph和JobGenerator是Spark Streaming调度的核心,记录当前调度到哪一进度,和业务有关。
ReceivedBlockTracker在接收到元数据信息后调用addBlock方法,先写入磁盘中,然后在写入内存中。


根据batchTime分配属于当前BatchDuration要处理的数据到timToAllocatedBlocks数据结构中。

Time类的是一个case class,记录时间,重载了操作符,隐式转换,值得借鉴。
case classTime(private valmillis: Long) { defmilliseconds: Long = millis def< (that: Time): Boolean = (this.millis < that.millis) def<= (that: Time): Boolean = (this.millis <= that.millis) def> (that: Time): Boolean = (this.millis > that.millis) def>= (that: Time): Boolean = (this.millis >= that.millis) def+ (that: Duration): Time = newTime(millis + that.milliseconds) def- (that: Time): Duration = newDuration(millis - that.millis) def- (that: Duration): Time = newTime(millis - that.milliseconds) // Java-friendlier versions of the above. defless(that: Time): Boolean = this< that deflessEq(that: Time): Boolean = this<= that defgreater(that: Time): Boolean = this> that defgreaterEq(that: Time): Boolean = this>= that defplus(that: Duration): Time = this+ that defminus(that: Time): Duration = this- that defminus(that: Duration): Time = this- that deffloor(that: Duration): Time = { valt = that.milliseconds newTime((this.millis / t) * t) } deffloor(that: Duration, zeroTime: Time): Time = { valt = that.milliseconds newTime(((this.millis - zeroTime.milliseconds) / t) * t + zeroTime.milliseconds) } defisMultipleOf(that: Duration): Boolean = (this.millis % that.milliseconds == 0) defmin(that: Time): Time = if(this< that) this elsethat defmax(that: Time): Time = if(this> that) this elsethat defuntil(that: Time, interval: Duration): Seq[Time] = { (this.milliseconds) until (that.milliseconds) by (interval.milliseconds) map (newTime(_)) } defto(that: Time, interval: Duration): Seq[Time] = { (this.milliseconds) to (that.milliseconds) by (interval.milliseconds) map (newTime(_)) } override deftoString: String = (millis.toString + " ms") } objectTime { implicit valordering = Ordering.by((time: Time) => time.millis) } |
跟踪Time对象,ReceiverTracker的allocateBlocksToBatch方法中的入参batchTime是被JobGenerator的generateJobs方法调用的。

JobGenerator的generateJobs方法是被定时器发送GenerateJobs消息调用的。



GenerateJobs中的时间参数就是nextTime,而nextTime+=period,这个period就是ssc.graph.batchDuration.milliseconds。

nextTime的初始值是在start方法中传入的startTime赋值的,即RecurringTimer的getStartTime方法的返回值,是当前时间period的(整数倍+1)。


Period这个值是我们调用new StreamingContext来构造StreamingContext时传入的Duration值。


ReceivedBlockTracker会清除过期的元数据信息,从HashMap中移除,也是先写入磁盘,然后在写入内存。

元数据的生成,消费和销毁都有WAL,所以失败时就可以从日志中恢复。从源码分析中得出只有设置了checkpoint目录,才进行WAL机制。

对传入的checkpoint目录来创建日志目录进行WAL。

这里是在checkpoint目录下创建文件夹名为receivedBlockMetadata的文件夹来保存WAL记录的数据。


把当前的DStream和JobGenerator的状态进行checkpoint,该方法是在generateJobs方法最后通过发送DoCheckpoint消息,来调用的。



到此,相信大家对“Driver容错安全性是什么”有了更深的了解,不妨来实际操作一番吧!这里是创新互联网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!
分享标题:Driver容错安全性是什么
分享路径:
http://cqcxhl.cn/article/geegdc.html