在Spark Streaming整个架构体系里面，数据接收其实最为关键的一个流程，在数据接收完之后，后面的数据处理上面就是复用Spark Core的数据处理架构(包括BlockManager来管理数据)，之后就是”水到渠成”的工作。所以作为Spark Streaming第一个章节，我们这里先从Receiver的启动流程介绍起走。
Receiver 架构
目前整个Spark Streaming架构里面，支持两种数据接收的方式: Receiver, Direct. 区别在[Spark-kafka-integration]里面有了详细介绍，后续会详细解读两者的区别。回到Receiver, 我们首要介绍下目前Receiver的架构。
Receiver.png
在我们自己实现Receiver里面主要关注几个点:
- onStart() 里面实现接受数据的逻辑，这里面只需要关心数据如何接收，无需关于SS里面的周期性等，而且必须是No-blocking的，所以这里一般会启动一个Thread，然后在里面不停地接受数据。
- Reliability保证，每个Receiver接受了数据之后，就需要存储到Spark的内存中区，这里就是调用Store方法，为了确保Reliability, 需要在Store成功之后进行ACK返回，甚至在这里需要进行WAL保证。
- 当写完成Receiver，会把这个Receiver封装到DStream里面，每个DStream都会有自己compute方法来触发从Receiver接受的数据进行转换转换为BlockRDD进而利用Spark Core的计算流程中
Receiver 启动流程
启动Receiver是一趟修炼的道路，会学习一种利用Spark分布式的环境启动自己的Process的方法。首先我们看下大致画的流程图如下:
Receiver启动流程.png
- Streaming的开始都是源于StreamingContext, 当初始化的时候，会产生JobScheduler(这货是驱动Streaming Job的关键)，在StreamingContext.start()的时候，整个JobScheduler就开始Start啦。
- ReceiverTracker, 他会负责整个StreamContexts里面的所有ReceiverInputDStream里面的Receivers的管理(略绕)。
- 那我们直接到StartAllReceivers()里面，这里面会先做schedule, 根据目前我们SS里面的Executors 和 所有需要启动的Receivers 统一考虑，来安排一下那些Receiver需要在哪个Executor上面进行启动。
val scheduledLocations = schedulingPolicy.scheduleReceivers(receivers, getExecutors)
- 最巧妙的地方来了，全部精华就在ReceiverTracker里面:
// Function to start the receiver on the worker node
val startReceiverFunc: Iterator[Receiver[_]] => Unit =
(iterator: Iterator[Receiver[_]]) => {
if (!iterator.hasNext) {
throw new SparkException(
"Could not start receiver as object not found.")
}
if (TaskContext.get().attemptNumber() == 0) {
val receiver = iterator.next()
assert(iterator.hasNext == false)
val supervisor = new ReceiverSupervisorImpl(
receiver, SparkEnv.get, serializableHadoopConf.value, checkpointDirOption)
supervisor.start()
supervisor.awaitTermination()
} else {
// It's restarted by TaskScheduler, but we want to reschedule it again. So exit it.
}
}
// Create the RDD using the scheduledLocations to run the receiver in a Spark job
val receiverRDD: RDD[Receiver[_]] =
if (scheduledLocations.isEmpty) {
ssc.sc.makeRDD(Seq(receiver), 1)
} else {
val preferredLocations = scheduledLocations.map(_.toString).distinct
ssc.sc.makeRDD(Seq(receiver -> preferredLocations))
}
receiverRDD.setName(s"Receiver $receiverId")
ssc.sparkContext.setJobDescription(s"Streaming job running receiver $receiverId")
ssc.sparkContext.setCallSite(Option(ssc.getStartSite()).getOrElse(Utils.getCallSite()))
val future = ssc.sparkContext.submitJob[Receiver[_], Unit, Unit](
receiverRDD, startReceiverFunc, Seq(0), (_, _) => Unit, ())
// We will keep restarting the receiver job until ReceiverTracker is stopped
future.onComplete {
case Success(_) =>
if (!shouldStartReceiver) {
onReceiverJobFinish(receiverId)
} else {
logInfo(s"Restarting Receiver $receiverId")
self.send(RestartReceiver(receiver))
}
case Failure(e) =>
if (!shouldStartReceiver) {
onReceiverJobFinish(receiverId)
} else {
logError("Receiver has been stopped. Try to restart it.", e)
logInfo(s"Restarting Receiver $receiverId")
self.send(RestartReceiver(receiver))
}
}(ThreadUtils.sameThread)
logInfo(s"Receiver ${receiver.streamId} started")
- 首先是startReceiverFunc，注意这里只是一个function, 它把传入的Receiver封装成ReceiverSupervisorImpl, 给每个Receiver创建监督人，监督人主要是负责启动Receiver，存储Receiver接受的数据 以及 同Driver保持通信
- 这时候创建出一个ReceiverRDD, 根据之前Receiver和Executor的安排, 这个RDD的内容就是 receiver -> 它安排的executor的host.
- 然后利用ssc.sparkContext.submitJob把这个RDD以及刚刚startReceiverFunc一起提交到集群上面去。这时候就利用Spark Core的分布式计算框架，让receiver 分散到自己prefer的host上面进行了启动。
这里就交了我们一招如何利用Spark分布式计算环境，启动我们自己进程的任务。关键点: 构造出自己数据对于的运行Executor, 然后把这批数据启动流程function构造出来，最后利用ssc.sparkContext.submitJob()
- 在ReceiverSupervisor, 用于处理Receiver收到的数据，并且利用BlockGenerator 来产生最终数据的Block.
总结
在Spark Streaming里面数据接收是关键，我们从上面看到我们的Receiver的启动流程，也学到如何利用Spark这个分布式平台进行提交自己的Job. 后面我们会看到Spark Streaming里面Job启动的流程。
