[spark streaming]Receiver工作内幕源码分析

不念不忘少年蓝@ 2022-05-15 11:04 232阅读 0赞

**原创文章，转载请注明:***转载自 [听风居士][Link 1][博客][Link 1](**[http://www.cnblogs.com/zhouyf/][Link 1])*

Spark streaming 程序需要不断接收新数据，然后进行业务逻辑处理，而用于接受数据的就是Recever。显然Receiver的正常运行对应整个Spark Streaming应用程序至关重要，如果Receiver出现异常，后面的业务逻辑就无从谈起。Spark Streaming 是如何实现Receiver以保证其可靠性的，本文将结合Spark Streaming的Receiver源码实现详细解析Receiver的实现原理。

**一、Receiver 实现策略思考**

1、启动Receiver的时候,启动一个Job，这个Job里面有RDD的transformations操作和action的操作，这个Job只有一个partition.这个partition的特殊是里面只有一个成员，这个成员就是启动的Receiver。这样做的问题：

a)  如果有多个InputDStream，那就要启动多个Receiver，每个Receiver也就相当于分片partition，那我们启动Receiver的时候理想的情况下是在不同的机器上启动Receiver，但是Spark Core的角度来看就是应用程序，感觉不到Receiver的特殊性，所以就会按照正常的Job启动的方式来处理，极有可能在一个Executor上启动多个Receiver.这样的话就可能导致负载不均衡。

b)  有可能启动Receiver失败，只要集群存在Receiver就不应该失败。

c)  运行过程中，就默认的而言如果是一个partition的话，那启动的时候就是一个Task，但是此Task也很可能失败，因此以Task启动的Receiver也会挂掉。

2、由Spark Streaming 自己管理Receiver，负责Receiver的调度和容错和启动。这样做的好处：

a）由Spark Streaming 调度Receiver 可以充分考虑负责均衡，避免将多个Receiver调度到同一台机器上

b）Receiver 失败后可以自动重新启动，继续接受数据，从而使程序持续不断继续工作下去。

c）Receiver 重启不收Task重启次数的限制。

**二、Spark Streaming的Receiver实现原理**

** 2.1、和Receiver实现相关核心成员**

（1）ReceiverTracker

（2）ReceiverTrackerEndpoint

（3）ReceiverSuperVisor

（4）Receiver

ReceiverTracker在Driver端,ReceiverSuperVisor和Receiver在Executor端,架构图如下：

![592648-20160527074112334-1472621778.png][]

** ****2.2  Spark Streaming的Receiver实现源码解析**

首先SparkStream启动时候会启动JobScheduler,在JobSceduler的start方法中,会实例化ReceiverTracker,并调用ReceiverTracker的start方法启动ReceiverTracker。

![592648-20160527074112850-824691233.png][]

ReceiverTracker的start方法首先检查输入流是否为空，如果不为空会创建ReceiverTrackerEndpoint并注册给rpcEnv。然后调用launchReceivers方法启动Receiver。其中receiverInputStreams是注册到DStreamGraph中的ReceiverInputDStream。

![592648-20160527074113350-125631632.png][]

下面看一下launchReceiver方法：基于ReceiverInputDStream(是在Driver端)来获得具体的Receivers实例，然后再把他们分不到Worker节点上。一个ReceiverInputDStream只产生一个Receiver

![592648-20160527074113741-813559853.png][]

首先从ReceiverInputDStream中获取Receiver，然后调用runDummySparkJob启动一个虚拟任务，我们在后面再分析这个虚拟任务，先看一下后面的核心代码：

*endpoint*.send(*StartAllReceivers*(receivers))

此处的endpoint就是刚才实例化的ReceiverTrackerEndpoint对象的引用，可以看到此处给endpoint发送了StartAllReceivers消息。

下面看下一ReceiverTrackerEndpoint收到StartAllReceivers消息后的处理逻辑：

![592648-20160527074114147-1438112581.png][]

首先，根据一定的调度策略给传入receivers分配相应的executors，从这里可以看出，Receiver的调度并不是交给spark内核完成的，而是由Spark Streaming框架完成调度过程。这样做的目的就是为了避免Spark内核将Receiver当做普通的job而将多个Receiver调度到同一个节点上。

Spark Streaming的调度策略这里不做分析，接着看下面的代码，迭代所以的receiver ，对每个receiver调用 startReceiver方法在具体Executor上启动Receiver。

这个startReceiver比较复杂，我们一步步分析，先看最核心的一行代码：

![592648-20160527074114522-1013534400.png][]

可看到，Spark Streaming 为每个Receiver 启动了一个job，而不是由Action操作出发Job执行。

这里job的提交主要关注两个参数receiverRDD和startReceiverFunc。

receiverRDD的源码：

![592648-20160527074114913-281015364.png][]

可以看到调用了SparkContext的makeRDD方法创建了RDD，该RDD只有一条数据，就是receiver对象

下面看一看startReceiverFunc的源码

![592648-20160527074115350-1030549616.png][]

startReceiverFunc 在worker节点上启动receiver，首先创建了一个ReceiverSupervisiorImpl 对象 supervisor，然后调用supervisor的start方法在该节点上启动supervisor：

![592648-20160527074115803-68703720.png][]

ReceiverSupervisiorImpl 是继承自ReceiverSupervisor，ReceiverSupervisor中调用了startReceiver方法：

![592648-20160527074116147-283184802.png][]

首先调用onReceiverStart方法，将Receiver注册给receiverTracker：

![592648-20160527074116506-1424899959.png][]

如果注册成功，调用了Receiver的onStart方法在Executor启动Receiver不断接受数据，并将接收的数据交给BlockManager管理，至此Receiver启动完成。

回到ReceiverTracker的startReceiver方法，如果Receiver对应的job完成，无论返回成功或失败，只要ReceiverTracker还没有停止就会发送RestartReceiver消息给ReceiverTrakerEndpoint，重启Receiver。从这里可以看出Receiver不会像普通的spark core 程序一样受到重试次数的限制而导致作业失败

![592648-20160527074116944-923645569.png][]

最后，在看一下runDummyJob方法：

![592648-20160527074117381-796906042.png][]

该方法运行了一个简单的wordcount程序，运行该程序的目的是确保所有slaves节点都被注册了，让receiver尽量分配到不同的work上运行，看一下getExecutors的源码 :

![592648-20160527074117803-1495220110.png][]

**总结：**Driver端的ReceiverTracker管理所有Executor上的Receiver任务，他有一个ReceiverTrakerEndpoint 消息通讯体，这个消息通讯体在startReceiver方法中提交Receiver的job在具体Executor上运行，并接受Executor端发送过来的消息（比如注册Receiver）,在Executor端有一个ReceiverSupervisor专门管理Receiver，负责Receiver的注册启动与ReceiverTracker的信息交互。

![592648-20160527074118131-1332729847.png][]

[Link 1]: http://www.cnblogs.com/zhouyf/
[592648-20160527074112334-1472621778.png]: /images/20220515/b4224eaac998415a861e3820ba87a874.png
[592648-20160527074112850-824691233.png]: /images/20220515/22933db60c1f4c4abc790c89e5f8e56f.png
[592648-20160527074113350-125631632.png]: /images/20220515/e82a235e76364c0485f198afcd0974fb.png
[592648-20160527074113741-813559853.png]: /images/20220515/0cc86f848f094d2790c87e472f52b8f6.png
[592648-20160527074114147-1438112581.png]: /images/20220515/fc22551d53ee4fc7b05316aff3c5163e.png
[592648-20160527074114522-1013534400.png]: /images/20220515/6cc722a650a74474b13aba6d4adf6d6a.png
[592648-20160527074114913-281015364.png]: /images/20220515/a580b102ed0a4bf1ba479379cafb524b.png
[592648-20160527074115350-1030549616.png]: /images/20220515/84c811e827e24b93bcf17019da1ef938.png
[592648-20160527074115803-68703720.png]: /images/20220515/71d9fbfb4006487f9850f2e05b05973f.png
[592648-20160527074116147-283184802.png]: /images/20220515/8a5c47a6811743c6b19c341a4d776072.png
[592648-20160527074116506-1424899959.png]: /images/20220515/1f5921e3ba7c4d8d9a59648e8d624540.png
[592648-20160527074116944-923645569.png]: /images/20220515/bdfe46f49c4c413fa5cb47958ba7bf75.png
[592648-20160527074117381-796906042.png]: /images/20220515/a2fa0ff7a2884b148a0b3d39d083a52f.png
[592648-20160527074117803-1495220110.png]: /images/20220515/72d2fe536b32470cb21a9d865276c233.png
[592648-20160527074118131-1332729847.png]: /images/20220515/abf010e5950b460884f00e09b9313ed3.png