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发行版笔记13:Spark Streaming源码解读之Driver容错安全性