spark发行版笔记13

Posted 2020-07-09

本期概览：

ReceiverTracker架构设计

消息循环系统

ReceiverTracker具体的实现

Spark Streaming作为Spark Core基础 架构之上的一个应用程序，其中的ReceiverTracker接收到数据之后，具体该怎么进行数据处理呢？

为了弄清楚这个问题，首先，我们打开源码

找到ReceiverSupervisorImpl这个类

从源码中可以看出，写数据是通过ReceivedBlockHandler的对象receivedBlockHandler写的。写的过程中有俩种方式，一种是基于WAL方式进行容错写。一种是直接写（相对不安全）。如下图所示

 

然后存储数据完成后并报告给Driver,以便Driver对元数据进行存储，如下所示

用于汇报给Driver的消息类、如下图所示

上图谈到了Record,要注意到，一般专业的描述处理的数据的大小的时候，应该用多少条记录来描述更科学，一般说数据规模达到多少多少百亿条记录，而不是说数据规模达到多少PB的数据规模，这样不是很科学，因为记录可能有很多字段，比如说，1PB的数据，5个字段，和5PB的数据1个字段是差不多的。所以1PB的数据规模未必比5PB的数据规模体现出一个大数据引擎的数据处理能力。也比如说，有些数据是视频或者音频。更不适合说多少个PB来描述规模大小。

上图说明ReceiverSupervisorImpl中有ReceiverTracker的通信体，能进行与ReceiverTracker的通信

并且ReceiverSupervisorImpl将数据的元数据信息汇报给ReceiverTracker

于是，我们进入ReceiverTracker这个类，这个类是整个流处理数据管理的中心。

 

ReceiverTracker中有endpoint通信体，这个通信体接收来自ReceiverSuperVisorImpl的元数据的数据汇报。

 

 

接下来，我们再进入ReceiverTracker本身，从整体上认识ReceiverTracker。

记录Receiver的三种状态，分别为非活跃状态，正在执行调度任务状态，活跃状态

密封关键字，说明所有的子类都密封在这里，方便管理

/**

 * This message will trigger ReceiverTrackerEndpoint to restart a Spark job for the receiver.

 */

这个消息用来告知为receiver启动一个job， ReceiverTracker有很多这样的case class用于通信。

private[streaming] case class RestartReceiver(receiver: Receiver[_])

  extends ReceiverTrackerLocalMessage

再比如此类相同的消息

/**

 * This message will trigger ReceiverTrackerEndpoint to send stop signals to all registered

 * receivers.

 */

private[streaming] case object StopAllReceivers extends ReceiverTrackerLocalMessage

注意：param skipReceiverLaunch Do not launch the receiver. This is useful for testing.，如下图

简单的来说，ReceiverTracker可以简单的说包括Receiver的数据的启动接收，管理，回收三个过程。

 

事先来个预告，我们将把Streaming流处理的所有的代码一行行的过滤，讲整个streaming通过一滴水看世界。

 

所有的输入流都会交给grapx对象，因为该对象会将所有的待调度的数据统一调度。

内部还有一个成员叫做ReceiverBlockTracker

ListenerBus非常的重要，后续我们会重点分析ListenerBus的源代码，它在监控层面起着重要的作用。

在这里，可以看出ReceiverTracker的状态有如下的4种状态，分别为

初始化，开始，正在停止中，停止了。

接收到ReceiverSuperVisorImpl远程发送过来的消息之后进行处理的过程在此。

这也是今天的重点之一。

先写日志后再进行下一步操作，这里是出于容错的原因考虑的。

注意：这里如果指定了checkpoint目录的话，才会使得isWriteAheadLogEnabled为true.

ReceivedBlockTrackerLogEvent其实就是元数据信息。

用一个HashMap结构将Stream 与 BlockQueue中的Block一一对应，可谓是真的巧妙到了极点。

 

再回到我们的消息通信层面。

回复对方，告知对方，addBlock成功。并且保存有数据的元数据信息。

ReceivedBlockTracker类的主要的任务在于将Block分配给没有分配Block的Stream batch。

这是具体分配Block给batch的代码。

这里说明具体的分配是以batch time为单位分配的.

再次看看消息通信体。

这里说启动所有的Receiver.

启动所有的receiver

这样，整个数据接收的环节就打通了。

最后做点补充：

该阶段是CleanupOldBlocks阶段，此时将发送消息给ReceiverSuperVisorImpl，从而让它执行cleanUpOldBlocks方法。

/** Update a receiver‘s maximum ingestion rate */

最后stopAllReceivers，结束了。

?