ReceiverTracker怎么實現(xiàn)

本篇內(nèi)容主要講解“ReceiverTracker怎么實現(xiàn)”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學(xué)習(xí)“ReceiverTracker怎么實現(xiàn)”吧!

ReceiverTacker類如下，從源碼注釋可以看出該類的作用。

    管理ReceiverInputDStreams的執(zhí)行，記錄Receiver發(fā)來的元數(shù)據(jù)信息。ReceiverTacker類構(gòu)造時必須傳入StreamingContext對象。

    ReceiverTacker類內(nèi)部有ReceiverTackerEndpoint這個消息通信體，用于和運行在Executor端的ReceiverSupervisorImpl進行通信，包括Receiver的注冊，重啟Receiver，清除之前的Block數(shù)據(jù)，更新限流值，添加Block元數(shù)據(jù)信息等消息。

    接下來以接收到來自Executor端的ReceiverSupervisorImpl發(fā)來添加元數(shù)據(jù)信息的AddBlock消息，進行講解具體的處理流程。

ReceivedBlockInfo類包含了StreamID，Block中記錄條數(shù)，元數(shù)據(jù)Metadata，接收Block的存儲結(jié)果（BlockID和記錄數(shù)量）

ReceiverBlockTracker類是addBlock方法的具體實現(xiàn)。

        1.調(diào)用ReceiverBlockTracker的writeToLog方法

        2.調(diào)用ReceiverBlockTracker的getReceivedBlockQueue方法，其中streamIdToUnallocatedBlockQueues為HashMap，Key為StreamID，Value為ReceivedBlockQueue。而ReceivedBlockQueue 的定義為private type ReceivedBlockQueue = mutable.Queue[ReceivedBlockInfo]

        ReceiverBlockTracker類，可以從源碼中看出，他會記錄所有接收到的Block信息，根據(jù)需要把Block分配給Batch。如果設(shè)置了checkpoint，開啟WAL，則會把所有的操作保存到預(yù)寫日志中，因此當Driver失敗后就可以從checkpoint和WAL中恢復(fù)ReceiverTracker的狀態(tài)。

        ReceiverBlockTracker類中重要的方法，allocateBlocksToBatch。private val timeToAllocatedBlocks = new mutable.HashMap[Time, AllocatedBlocks]存儲批處理時刻，分配到的Blocks數(shù)據(jù)。

        該方法是被ReceiverTracker調(diào)用的。

        而ReceiverTracker的allocateBlocksToBatch方法是被JobGenerator的generateJobs方法調(diào)用的。

ReceiverBlockTracker類中重要的方法，getBlocksOfBatch。

 該方法是被ReceiverTracker的getBlocksOfBatch調(diào)用。

ReceiverTracker的getBlocksOfBatch方法是被ReceiverInputDStream的compute方法調(diào)用的。

到此，相信大家對“ReceiverTracker怎么實現(xiàn)”有了更深的了解，不妨來實際操作一番吧！這里是億速云網(wǎng)站，更多相關(guān)內(nèi)容可以進入相關(guān)頻道進行查詢，關(guān)注我們，繼續(xù)學(xué)習(xí)！