Spark Streaming源码解读之流数据不断接收详解.doc

Spark Streaming源码解读之流数据不断接收详解 特别说明： 在上一遍文章中有详细的叙述Receiver启动的过程，如果不清楚的朋友，请您查看上一篇博客，这里我们就基于上篇的结论，继续往下说。 博文的目标是： Spark Streaming在接收数据的全生命周期贯通 组织思路如下： a) 接收数据的架构模式的设计 b) 然后再具体源码分析 接收数据的架构模式的设计 1. 当有Spark Streaming有application的时候Spark Streaming会持续不断的接收数据。 2. 一般Receiver和Driver不在一个进程中的，所以接收到数据之后要不断的汇报给Driver。 3. Spark Streaming要接收数据肯定要使用消息循环器，循环器不断的接收到数据之后，然后将数据存储起来，再将存储完的数据汇报给Driver。 4. Spark Streaming数据接收的过程也是MVC的架构，M是model也就是Receiver. C是Control也就是存储级别的ReceiverSupervisor。V是界面。 5. ReceiverSupervisor是控制器，Receiver的启动是靠ReceiverTracker启动的，Receiver接收到数据之后是靠ReceiverSupervisor存储数据的。然后Driver就获得元数据也就是界面，通过界面来操作底层的数据，这个元数据就相当于指针。 Spark Streaming接收数据流程如下： 具体源码分析 1. ReceiverTracker通过发送Job的方式，并且每个Job只有一个Task，并且Task中只通过一个ReceiverSupervisor启动一个Receiver. 2. 下图就是Receiver启动的流程图，现在就从ReceiverTracker的start开始今天的旅程。 3. Start方法中创建Endpoint实例 /** Start the endpoint and receiver execution thread. */ def start(): Unit = synchronized { if (isTrackerStarted) { throw new SparkException(ReceiverTracker already started) } if (!receiverInputStreams.isEmpty) { endpoint = ssc.env.rpcEnv.setupEndpoint( ReceiverTracker, new ReceiverTrackerEndpoint(ssc.env.rpcEnv)) if (!skipReceiverLaunch) launchReceivers() logInfo(ReceiverTracker started) trackerState = Started } } 4. LaunchReceivers源码如下： /** * Get the receivers from the ReceiverInputDStreams, distributes them to the * worker nodes as a parallel collection, and runs them. */ private def launchReceivers(): Unit = { val receivers = receiverInputStreams.map(nis = { val rcvr = nis.getReceiver() rcvr.setReceiverId(nis.id) rcvr }) runDummySparkJob() logInfo(Starting + receivers.length + receivers) //此时的endpoint就是前面实例化的ReceiverTrackerEndpoint endpoint.send(StartAllReceivers(receivers)) } 5. 从图上可以知道，send发送消息之后，ReceiverTrackerEndpoint的receive就接收到了消息。 override def receive: PartialFunction[Any, Unit] = { // Local messages case StartAllReceivers(receivers) = val scheduledLocations = sched