Spark Streaming怎么使用

蜗牛 互联网技术资讯 2021-12-16 216 0

这篇文章主要介绍“Spark Streaming怎么使用”,在日常操作中,相信很多人在Spark Streaming怎么使用问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”Spark Streaming怎么使用”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!

DStream是逻辑级别的,而RDD是物理级别的。DStream是随着时间的流动内部将集合封装RDD。对DStream的操作,转过来对其内部的RDD操作。

Spark Streaming怎么使用  spark 第1张

纵轴为空间维度:代表的是RDD的依赖关系构成的具体的处理逻辑的步骤,是用DStream来表示的。

横轴为时间维度:按照特定的时间间隔不断地生成job对象,并在集群上运行。

随着时间的推移,基于DStream Graph 不断生成RDD Graph ,也即DAG的方式生成job,并通过Job Scheduler的线程池的方式提交给spark cluster不断的执行。

由上可知,RDD    与  DStream的关系如下

RDD是物理级别的,而 DStream 是逻辑级别的

DStream是RDD的封装类,是RDD进一步的抽象

DStream 是RDD的模板。DStream要依赖RDD进行具体的数据计算

注意:纵轴维度需要RDD,DAG的生成模板,需要TimeLine的job控制器

横轴维度(时间维度)包含batch interval,窗口长度,窗口滑动时间等。

3,Spark Streaming源码解析

StreamingContext方法中调用JobScheduler的start方法

Spark Streaming怎么使用  spark 第2张

JobGenerator的start方法中,调用startFirstTime方法,来开启定时生成Job的定时器

Spark Streaming怎么使用  spark 第3张

startFirstTime方法,首先调用DStreamGraph的start方法,然后再调用RecurringTimer的start方法。

Spark Streaming怎么使用  spark 第4张

timer对象为一个定时器,根据batchInterval时间间隔定期向EventLoop发送GenerateJobs的消息。

Spark Streaming怎么使用  spark 第5张

接收到GenerateJobs消息后,会回调generateJobs方法。

Spark Streaming怎么使用  spark 第6张

generateJobs方法再调用DStreamGraph的generateJobs方法生成Job

Spark Streaming怎么使用  spark 第7张

DStreamGraph的generateJobs方法

Spark Streaming怎么使用  spark 第8张

DStreamGraph的实例化是在StreamingContext中的

Spark Streaming怎么使用  spark 第9张

DStreamGraph类中保存了输入流和输出流信息

Spark Streaming怎么使用  spark 第10张

Spark Streaming怎么使用  spark 第11张

Spark Streaming怎么使用  spark 第12张

回到JobGenerator的start方法中receiverTracker.start()

Spark Streaming怎么使用  spark 第13张

Spark Streaming怎么使用  spark 第14张

其中ReceiverTrackerEndpoint对象为一个消息循环体

Spark Streaming怎么使用  spark 第15张

launchReceivers方法中发送StartAllReceivers消息

Spark Streaming怎么使用  spark 第16张

接收到StartAllReceivers消息后,进行如下处理

Spark Streaming怎么使用  spark 第17张

Spark Streaming怎么使用  spark 第18张

StartReceiverFunc方法如下,实例化Receiver监控者,开启并等待退出

Spark Streaming怎么使用  spark 第19张

supervisor的start方法中调用startReceiver方法

Spark Streaming怎么使用  spark 第20张

Spark Streaming怎么使用  spark 第21张

我们以socketTextStream为例,其启动的是SocketReceiver,内部开启一个线程,来接收数据。

Spark Streaming怎么使用  spark 第22张

Spark Streaming怎么使用  spark 第23张

内部调用supervisor的pushSingle方法,将数据聚集后存放在内存中

Spark Streaming怎么使用  spark 第24张

supervisor的pushSingle方法如下,将数据放入到defaultBlockGenerator中,defaultBlockGenerator为BlockGenerator,保存Socket接收到的数据

Spark Streaming怎么使用  spark 第25张

Spark Streaming怎么使用  spark 第26张

BlockGenerator对象中有一个定时器,来更新当前的Buffer

Spark Streaming怎么使用  spark 第27张

Spark Streaming怎么使用  spark 第28张

BlockGenerator对象中有一个线程,来从阻塞队列中取出数据

Spark Streaming怎么使用  spark 第29张

Spark Streaming怎么使用  spark 第30张

Spark Streaming怎么使用  spark 第31张

调用ReceiverSupervisorImpl类中的继承BlockGeneratorListener的匿名类中的onPushBlock方法。

Spark Streaming怎么使用  spark 第32张

Spark Streaming怎么使用  spark 第33张

Spark Streaming怎么使用  spark 第34张

receivedBlockHandler对象如下

Spark Streaming怎么使用  spark 第35张

这里我们讲解BlockManagerBasedBlockHandler的方式

Spark Streaming怎么使用  spark 第36张

trackerEndpoint如下

Spark Streaming怎么使用  spark 第37张

Spark Streaming怎么使用  spark 第38张

其实是发送给ReceiverTrackerEndpoint类,

Spark Streaming怎么使用  spark 第39张

Spark Streaming怎么使用  spark 第40张

Spark Streaming怎么使用  spark 第41张

Spark Streaming怎么使用  spark 第42张

InputInfoTracker类的reportInfo方法只是对数据进行记录统计

Spark Streaming怎么使用  spark 第43张

Spark Streaming怎么使用  spark 第44张

Spark Streaming怎么使用  spark 第45张

其generateJob方法是被DStreamGraph调用

Spark Streaming怎么使用  spark 第46张

DStreamGraph的generateJobs方法是被JobGenerator类的generateJobs方法调用。

Spark Streaming怎么使用  spark 第47张

JobGenerator类中有一个定时器,batchInterval发送GenerateJobs消息

Spark Streaming怎么使用  spark 第48张

总结:

1,当调用StreamingContext的start方法时,启动了JobScheduler

2,当JobScheduler启动后会先后启动ReceiverTracker和JobGenerator

3,ReceiverTracker启动后会创建ReceiverTrackerEndpoint这个消息循环体,来接收运行在Executor上的Receiver发送过来的消息

4,ReceiverTracker在启动时会给自己发送StartAllReceivers消息,自己接收到消息后,向Spark提交startReceiverFunc的Job

5,startReceiverFunc方法中在Executor上启动Receiver,并实例化ReceiverSupervisorImpl对象,来监控Receiver的运行

6,ReceiverSupervisorImpl对象会调用Receiver的onStart方法,我们以SocketReceiver为例,启动一个线程,连接Server,读取网络数据先调用ReceiverSupervisorImpl的pushSingle方法,

保存在BlockGenerator对象中,该对象内部有个定时器,放到阻塞队列blocksForPushing,等待内部线程取出数据放到BlockManager中,并发AddBlock消息给ReceiverTrackerEndpoint。

ReceiverTrackerEndpoint为ReceiverTracker的内部类,在接收到addBlock消息后将streamId对应的数据阻塞队列streamIdToUnallocatedBlockQueues中

7,JobGenerator启动后会启动以batchInterval时间间隔发送GenerateJobs消息的定时器

8,接收到GenerateJobs消息会先后触发ReceiverTracker的allocateBlocksToBatch方法和DStreamGraph的generateJobs方法

9,ReceiverTracker的allocateBlocksToBatch方法会调用getReceivedBlockQueue方法从阻塞队列streamIdToUnallocatedBlockQueues中根据streamId获取数据

10,DStreamGraph的generateJobs方法,继而调用变量名为outputStreams的DStream集合的generateJob方法

11,继而调用DStream的getOrCompute来调用具体的DStream的compute方法,我们以ReceiverInputDStream为例,compute方法是从ReceiverTracker中获取数据

到此,关于“Spark Streaming怎么使用”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注蜗牛博客网站,小编会继续努力为大家带来更多实用的文章!

免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:niceseo99@gmail.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。

评论