spark发行版笔记11

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了spark发行版笔记11相关的知识,希望对你有一定的参考价值。

本期概览:

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,结束了。

技术分享

以上是关于spark发行版笔记11的主要内容,如果未能解决你的问题,请参考以下文章

spark发行版笔记9

spark发行版笔记10

spark发行版笔记13

Spark发行版笔记13:Spark Streaming源码解读之Driver容错安全性

Spark Streaming发行版笔记15:no receivers彻底思考

Spark Streaming发行版笔记16:数据清理内幕彻底解密