10.Spark Streaming源码分析:Receiver数据接收全过程详解
2016-05-26 08:55
483 查看
原创文章,转载请注明:转载自 听风居士博客(http://blog.csdn.net/zhouzx2010)
在上一篇中介绍了Receiver的整体架构和设计原理,本篇内容主要介绍Receiver在Executor中数据接收和存储过程
一、Receiver启动过程回顾
如图,从ReceiverTracker的start方法开始,调用launchReceivers()方法,给endpoint发送消息,endpoint.send(StartAllReceivers(receivers)),endpoint就是ReceiverTrackerEndpoint,ReceiverTrackerEndpoint调用startReceiver方法在Executor中启动ReceiverSupervisor。详情请阅读链接中的文章http://blog.csdn.net/zhouzx2010
二、Receiver数据接收全过程总览
Receiver数据接收核心成员:
1.ReceiverSupervisor
2.BlockGenerator
3.Receiver
4.ReceiverTracker
5.BlockManager
首先ReceiverSupervisor启动,继而启动BlockGenerator和Receiver,其中BlockGenerator首先启动一个定时器定时将接收的数据封装成Block,然后启动一个线程不断将Block数据推送给BlockManager。而Receiver启动后不断接收数据,并不断将接受的数据添加到BlockGenerator中,这样BlockGenerator就不断将Receiver接收的数据推送到BlockManager,整个数据接收过程如下图:
三、Receiver数据接收过程详解
3.1 ReceiverSupervisor 启动
Spark Streaming 在集群启动Receiver的时候会先在Executor中启动ReceiverSupervisor来管理Receiver,ReceiverSupervisor的启动是通过调用他的start方法完成的,源码如下:
ReceiverSupervisor的start方法首先调用了onStart 方法,该方法启动了BlockGenerator,必须在启动Receiver之前调用该方法来启动BlockGenerator,因为必须BlockGenerator先启动,以保证接收到的数据能够被存储起来。该方法在ReceiverSupervisor的子类ReceiverSupervisorImpl有实现:
可以看到,在onStart方法中启动了所有的已经注册的BlockGenerator,其中registeredBlockGenerators是ReceiverSupervisorImpl的成员变量,在ReceiverSupervisorImpl实例化是被创建:
registeredBlockGenerators中的Generartor是什么时候产生,并加入到registeredBlockGenerators这个队列里的呢?ReceiverSupervisorImpl的createBlockGenerator方法对registeredBlockGenerators进行了元素添加:
从源码可以看到每一个receiver创建一个BlockGenerator,因为streamId与receiver一一对应。createBlockGenerator首先还将已经停止的BlockGenerator从registeredBlockGenerators队列清除。createBlockGenerator
方法又是在什么时候被调用的呢,答案是在ReceiverSupervisorImpl实例化时候被调用,请看源码:
现在registeredBlockGenerators中有了BlockGenerator,接下来分析BlockGenerator的启动过程
3.2 BlockGenerator 启动
BlockGenerator的start方法启动一个定时器blockIntervalTimer,用来定时生成调用updateCurrentBuffer方法:
时间间隔默认是200毫秒:
下面看一下updateCurrentBuffer的源码:
currentBuffer 是一个ArrayBuffer 存储了数据记录,updateCurrentBuffer方法将间隔时间段的数据记录封装成Block,然后将新的Block 放入等待推送的block队列blockForPushing:
回到BlockGenerator的start方法的另一个线程blockPushingThread
可以看到blockPushingThread的run方法调用了keepPushingBlocks方法将blockForPushing队列中的Block推送给BlockManager:
可以看到最多等待10毫秒从blockForPushing中取去Block,然后掉pushBlock方法将block推送给BlockManager,下面看一下pushBlock的源码
这里的listener 是 BlockGeneratorListener 对象,在ReceiverSupervisorImpl中被实例化:
我们关注其中的onPushBlock方法:调用了pushArrayBuffer方法:
这里调用了一个很重要的方法
pushAndReportBlock
我们看一下源码:
pushAndReportBlock方法用来存储接收的的数据Block并将元数据汇报给ReceiverTracker。
Block存储交给receivedBlockHandler处理,receivedBlockHandlery有两种实现方式:
看一下BlockManagerBasedBlockHandler的storeBlock方法:
BlockManagerBasedBlockHandler的storeBlock方法根据ReceivedBlock的不同类型分别处理,最终都是交给了BlockManger管理。
至此BlockGenerator 分析完毕,下面分析 Receiver 的启动过程
3.3 Receiver 启动
首先回到ReceiverSupervisorImpl的startReceiver方法:
首先调用了onReceiverStart方法,将Receiver注册给ReceiverTracker,重点看代码rever.onStart(),调用了receiver的onStart方法,此处我们以SocketReceiver为例进行讲解:
可以看到SocketReceiver的onStart()方法先给创建了一个Socket对象赋值给了socket,然后启动了一个后台进行“Socket Receiver” ,在后台进程的run方法中调用了receive()方法进行数据接收:
SocketReceiver的receive()方法不断从socket接收数据,然后调用store方法进行存储。
此处再次回到ReceiverSupervisor,调用了其中的pushSingle函数:
最终调用了BlockGenerator的addData方法:
可以看到BlockGenerator的addData方法将数据添加到currentBuffer中,然后blockIntervalTimer会每个200毫秒从currentBuffer取出数据封装成Block,并将Block添加到blockForPushing队列,blockPushingThread会将blockForPushing队列中的block推送给BlockManager。至此Receiver运行流程分析完毕。
四、总结:receiver数据接收流程图
总体过程:ReceiverSupervisor启动时,启动了BlockGenerator和Receiver,其中BlockGenerator首先启动一个定时器定时将接收的数据封装成Block,然后启动一个线程不断将Block数据推送给BlockManager。而Receiver启动后不断接收数据,不断将接受的数据添加到BlockGenerator的currentBuffer中,BlockGenerator不断从currentBuffer取数据,这样整个数据接收过程不断循环。
原创文章,转载请注明:转载自 听风居士博客(http://blog.csdn.net/zhouzx2010)
在上一篇中介绍了Receiver的整体架构和设计原理,本篇内容主要介绍Receiver在Executor中数据接收和存储过程
一、Receiver启动过程回顾
如图,从ReceiverTracker的start方法开始,调用launchReceivers()方法,给endpoint发送消息,endpoint.send(StartAllReceivers(receivers)),endpoint就是ReceiverTrackerEndpoint,ReceiverTrackerEndpoint调用startReceiver方法在Executor中启动ReceiverSupervisor。详情请阅读链接中的文章http://blog.csdn.net/zhouzx2010
二、Receiver数据接收全过程总览
Receiver数据接收核心成员:
1.ReceiverSupervisor
2.BlockGenerator
3.Receiver
4.ReceiverTracker
5.BlockManager
首先ReceiverSupervisor启动,继而启动BlockGenerator和Receiver,其中BlockGenerator首先启动一个定时器定时将接收的数据封装成Block,然后启动一个线程不断将Block数据推送给BlockManager。而Receiver启动后不断接收数据,并不断将接受的数据添加到BlockGenerator中,这样BlockGenerator就不断将Receiver接收的数据推送到BlockManager,整个数据接收过程如下图:
三、Receiver数据接收过程详解
3.1 ReceiverSupervisor 启动
Spark Streaming 在集群启动Receiver的时候会先在Executor中启动ReceiverSupervisor来管理Receiver,ReceiverSupervisor的启动是通过调用他的start方法完成的,源码如下:
ReceiverSupervisor的start方法首先调用了onStart 方法,该方法启动了BlockGenerator,必须在启动Receiver之前调用该方法来启动BlockGenerator,因为必须BlockGenerator先启动,以保证接收到的数据能够被存储起来。该方法在ReceiverSupervisor的子类ReceiverSupervisorImpl有实现:
可以看到,在onStart方法中启动了所有的已经注册的BlockGenerator,其中registeredBlockGenerators是ReceiverSupervisorImpl的成员变量,在ReceiverSupervisorImpl实例化是被创建:
registeredBlockGenerators中的Generartor是什么时候产生,并加入到registeredBlockGenerators这个队列里的呢?ReceiverSupervisorImpl的createBlockGenerator方法对registeredBlockGenerators进行了元素添加:
从源码可以看到每一个receiver创建一个BlockGenerator,因为streamId与receiver一一对应。createBlockGenerator首先还将已经停止的BlockGenerator从registeredBlockGenerators队列清除。createBlockGenerator
方法又是在什么时候被调用的呢,答案是在ReceiverSupervisorImpl实例化时候被调用,请看源码:
现在registeredBlockGenerators中有了BlockGenerator,接下来分析BlockGenerator的启动过程
3.2 BlockGenerator 启动
BlockGenerator的start方法启动一个定时器blockIntervalTimer,用来定时生成调用updateCurrentBuffer方法:
时间间隔默认是200毫秒:
下面看一下updateCurrentBuffer的源码:
currentBuffer 是一个ArrayBuffer 存储了数据记录,updateCurrentBuffer方法将间隔时间段的数据记录封装成Block,然后将新的Block 放入等待推送的block队列blockForPushing:
回到BlockGenerator的start方法的另一个线程blockPushingThread
可以看到blockPushingThread的run方法调用了keepPushingBlocks方法将blockForPushing队列中的Block推送给BlockManager:
可以看到最多等待10毫秒从blockForPushing中取去Block,然后掉pushBlock方法将block推送给BlockManager,下面看一下pushBlock的源码
这里的listener 是 BlockGeneratorListener 对象,在ReceiverSupervisorImpl中被实例化:
我们关注其中的onPushBlock方法:调用了pushArrayBuffer方法:
这里调用了一个很重要的方法
pushAndReportBlock
我们看一下源码:
pushAndReportBlock方法用来存储接收的的数据Block并将元数据汇报给ReceiverTracker。
Block存储交给receivedBlockHandler处理,receivedBlockHandlery有两种实现方式:
看一下BlockManagerBasedBlockHandler的storeBlock方法:
BlockManagerBasedBlockHandler的storeBlock方法根据ReceivedBlock的不同类型分别处理,最终都是交给了BlockManger管理。
至此BlockGenerator 分析完毕,下面分析 Receiver 的启动过程
3.3 Receiver 启动
首先回到ReceiverSupervisorImpl的startReceiver方法:
首先调用了onReceiverStart方法,将Receiver注册给ReceiverTracker,重点看代码rever.onStart(),调用了receiver的onStart方法,此处我们以SocketReceiver为例进行讲解:
可以看到SocketReceiver的onStart()方法先给创建了一个Socket对象赋值给了socket,然后启动了一个后台进行“Socket Receiver” ,在后台进程的run方法中调用了receive()方法进行数据接收:
SocketReceiver的receive()方法不断从socket接收数据,然后调用store方法进行存储。
此处再次回到ReceiverSupervisor,调用了其中的pushSingle函数:
最终调用了BlockGenerator的addData方法:
可以看到BlockGenerator的addData方法将数据添加到currentBuffer中,然后blockIntervalTimer会每个200毫秒从currentBuffer取出数据封装成Block,并将Block添加到blockForPushing队列,blockPushingThread会将blockForPushing队列中的block推送给BlockManager。至此Receiver运行流程分析完毕。
四、总结:receiver数据接收流程图
总体过程:ReceiverSupervisor启动时,启动了BlockGenerator和Receiver,其中BlockGenerator首先启动一个定时器定时将接收的数据封装成Block,然后启动一个线程不断将Block数据推送给BlockManager。而Receiver启动后不断接收数据,不断将接受的数据添加到BlockGenerator的currentBuffer中,BlockGenerator不断从currentBuffer取数据,这样整个数据接收过程不断循环。
原创文章,转载请注明:转载自 听风居士博客(http://blog.csdn.net/zhouzx2010)
相关文章推荐
- PHP新的垃圾回收机制:Zend GC详解
- 《上古天真论》第五讲文字版
- Python Dict用法
- s5pv210 datasheet_system_audio_etc
- 服务器磁盘raid0,raid1,raid10,raid5,raid50,raid6,raid60 分析与案例
- Python 目录操作
- LeetCode 337. House Robber III
- 软件工程用图
- linuxcon-europe 2015 linux大会
- c#遍历并判断实体或类的成员属性
- tomcat7+jdk的keytool生成证书 配置https
- 第11周项目1-点-圆-圆柱类族的设计 (3)
- 解析配置文件ConfigParser模块
- 第十三周项目2.1—动物这样叫
- zzuli 1877: 蛤玮打扫教室
- 移动端canvas.arc与css的filter
- Unity热门插件推荐
- 深复制体验
- 孙钟秀-《 操作系统教程 》(第4版)注释(稿)
- 《上古天真论》第四讲文字版