首页 > 资讯 > 云计算 > 正文

Spark定制版：009~SparkStreaming源码解读之Receiver在Driver的精妙实现全生命周期彻底研究和思考

16-05-30 来源：[db:作者]

收藏我要投稿

本讲内容：

a. Receiver启动的方式设想

b. Receiver启动源码彻底分析

注：本讲内容基于Spark 1.6.1版本(在2016年5月来说是Spark最新版本)讲解。

上节回顾

上一讲中，我们给大家具体分析了RDD的物理生成和逻辑生成过程，彻底明白DStream和RDD之间的关系，及其内部其他有关类的具体依赖等信息：

a. DStream是RDD的模板，其内部generatedRDDs 保存了每个BatchDuration时间生成的RDD对象实例。DStream的依赖构成了RDD依赖关系，即从后往前计算时，只要对最后一个DStream计算即可。

b. JobGenerator每隔BatchDuration调用DStreamGraph的generateJobs方法，调用了ForEachDStream的generateJob方法，其内部先调用父DStream的getOrCompute方法来获取RDD，然后在进行计算，从后往前推，第一个DStream是ReceiverInputDStream，其comput方法中从receiverTracker中获取对应时间段的metadata信息，然后生成BlockRDD对象，并放入到generatedRDDs中

开讲

由上几节课中我们知道：

a. Spark Streaming的应用程序在处理数据时，会在开始的阶段做好接收数据的准备

b. Spark Streaming的应用程序代码定义DStream时，会定义一个或多个InputDStream;而每个InputDStream则分别对应有一个Receiver

结合源码的具体类和方法绘制Receiver启动全生命周期主流程图：

我们就从本讲的内容开始，为大家解析：

那么Receiver启动的方式设想到底是什么呢?

Receiver启动的设计问题分析:

a. Spark Streaming通过Receiver持续不断的从外部数据源接收数据，并把数据汇报给Driver端，由此每个Batch Durations就可以根据汇报的数据生成不同的Job

b. 即有可能在同一个Executor之中启动多个Receiver，这种情况之下导致负载不均匀

c. 由于Executor运行本身的故障，task 有可能启动失败，整个job启动就失败，即receiver启动失败

d. Receiver属于Spark Streaming应用程序启动阶段，它又是如何设计，来达到Receiver始终会被启动或被启动

e. Receivers和InputDStreams又是如何一一对应的，默认情况下一般只有一个Receiver吗?

来吧，走进源码一起看个究竟!!
[page]

Receiver启动源码彻底分析：

如何启动Receiver?

a. 从Spark Core的角度来看，Receiver的启动Spark Core并不知道， Receiver是通过Job的方式启动的，运行在Executor之上的，由task运行

b. 一般情况下，只有一个Receiver，但是可以创建不同的数据来源的InputDStream

c. 启动Receiver的时候，实际上一个receiver就是一个partition，并由一个Job启动，这个Job里面有RDD的transformations操作和action的操作，随着定时器触发，不断的产生有数据接收，每个时间段中产生的接收数据实其上就是一个partition

如此，又回到了最初Receiver启动的方式设想中的问题：

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

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

c. 从运行过程中看，一个Reveiver就是一个partition的话，启动的由一个Task，如果Task启动失败，以Task启动的Receiver也会失败

由此，可以得出，对于Receiver失败的话，后果是非常严重的，那么在Spark Streaming如何防止这些事的呢?

Spark Streaming源码分析，在Spark Streaming之中就指定如下信息：

a. Spark使用一个Job启动一个Receiver.最大程度的保证了负载均衡

b. Spark Streaming已经指定每个Receiver运行在那些Executor上，在Receiver运行之前就指定了运行的地方

c. 如果Receiver启动失败，此时并不是Job失败，在内部会重新启动Receiver

进入到StreamingContext源码，开启解密之旅吧!

在StreamingContext的start方法被调用的时候，JobScheduler的start方法会被调用

scheduler.start()：//启动子线程，一方面为了本地初始化工作，另外一方面是不要阻塞主线程

而在JobScheduler的start方法中ReceiverTracker的start方法被调用，Receiver就启动了

ReceiverTracker的start方法启动RPC消息通信体，为啥呢?因为receiverTracker会监控整个集群中的Receiver，Receiver转过来要向ReceiverTrackerEndpoint汇报自己的状态，接收的数据，包括生命周期等信息

基于ReceiverInputDStream(是在Driver端)来获得具体的Receivers实例，然后再把他们分布到Worker节点上。一个ReceiverInputDStream只对应一个Receiver

其中runDummySparkJob()为了确保所有节点活着，而且避免所有的receivers集中在一个节点上

再回去看ReceiverTracker.launchReceivers()中的getReceiver()

ReceiverInputDStream的getReceiver()方法返回Receiver对象。该方法实际上要在ReceiverInputDStream的子类实现。

相应的，ReceiverInputDStream的子类中必须要实现这个getReceiver()方法。ReceiverInputDStream的子类还必须定义自己对应的Receiver子类，因为这个Receiver子类会在getReceiver()方法中用来创建这个Receiver子类的对象。
[page]

因此，我们需要查看以下ReceiverInputDStream的继承关系

根据继承关系，这里看一下ReceiverInputDStream的子类SocketInputDStream中的getReceiver方法

SocketInputDStream中还定义了相应的Receiver子类SocketReceiver。SocketReceiver类中还必须定义onStart方法

onStart方法会启动后台线程，调用receive方法

启动socket开始接收数据

再回到ReceiverTracker.launchReceivers()中，看最后的代码 endpoint.send(StartAllReceivers(receivers))。这个代码给ReceiverTrackerEndpoint对象发送了StartAllReceivers消息，ReceiverTrackerEndpoint对象接收后所做的处理在ReceiverTrackerEndpoint.receive中。