ReceiverTracker怎么实现

ReceiverTracker怎么实现

• 来源:网络
• 更新日期:2022-01-13

摘要：本篇内容主要讲解“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怎么实现”有了更深的了解，不妨来实际操作一番吧！这里是箭头云网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！