ReceiverTracker怎么实现

本篇内容主要讲解“ReceiverTracker怎么实现”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“ReceiverTracker怎么实现”吧!

创新互联专注于坡头网站建设服务及定制,我们拥有丰富的企业做网站经验。 热诚为您提供坡头营销型网站建设,坡头网站制作、坡头网页设计、坡头网站官网定制、微信小程序开发服务,打造坡头网络公司原创品牌,更为您提供坡头网站排名全网营销落地服务。

ReceiverTacker类如下,从源码注释可以看出该类的作用。

管理ReceiverInputDStreams的执行,记录Receiver发来的元数据信息。ReceiverTacker类构造时必须传入StreamingContext对象。

ReceiverTacker类内部有ReceiverTackerEndpoint这个消息通信体,用于和运行在Executor端的ReceiverSupervisorImpl进行通信,包括Receiver的注册,重启Receiver,清除之前的Block数据,更新限流值,添加Block元数据信息等消息。

接下来以接收到来自Executor端的ReceiverSupervisorImpl发来添加元数据信息的AddBlock消息,进行讲解具体的处理流程。

ReceivedBlockInfo类包含了StreamID,Block中记录条数,元数据Metadata,接收Block的存储结果(BlockID和记录数量)

ReceiverBlockTracker类是addBlock方法的具体实现。

1.调用ReceiverBlockTracker的writeToLog方法

2.调用ReceiverBlockTracker的getReceivedBlockQueue方法,其中streamIdToUnallocatedBlockQueues为HashMap,Key为StreamID,Value为ReceivedBlockQueue。而ReceivedBlockQueue 的定义为private type ReceivedBlockQueue = mutable.Queue[ReceivedBlockInfo]

ReceiverBlockTracker类,可以从源码中看出,他会记录所有接收到的Block信息,根据需要把Block分配给Batch。如果设置了checkpoint,开启WAL,则会把所有的操作保存到预写日志中,因此当Driver失败后就可以从checkpoint和WAL中恢复ReceiverTracker的状态。

ReceiverBlockTracker类中重要的方法,allocateBlocksToBatch。private val timeToAllocatedBlocks = new mutable.HashMap[Time, AllocatedBlocks]存储批处理时刻,分配到的Blocks数据。

该方法是被ReceiverTracker调用的。

而ReceiverTracker的allocateBlocksToBatch方法是被JobGenerator的generateJobs方法调用的。

ReceiverBlockTracker类中重要的方法,getBlocksOfBatch。

该方法是被ReceiverTracker的getBlocksOfBatch调用。

ReceiverTracker的getBlocksOfBatch方法是被ReceiverInputDStream的compute方法调用的。

到此,相信大家对“ReceiverTracker怎么实现”有了更深的了解,不妨来实际操作一番吧!这里是箭头云网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!


标题名称:ReceiverTracker怎么实现
文章网址:http://pwwzsj.com/article/cjhoei.html