Spark Streaming原理剖析
1.初始化與集群上分布接收器 圖8-12所示為Spark Streaming執(zhí)行模型從中可看到數(shù)據(jù)接收及組件間的通信。
初始化的過程主要可以概括為以下兩點(diǎn)。
1)調(diào)度器的初始化。
2)將輸入流的接收器轉(zhuǎn)化為RDD在集群打散，然后啟動(dòng)接收器集合中的每個(gè)接收器。
下面通過具體的代碼更深入地理解這個(gè)過程。
(1)NetworkWordCount示例 本例以NetworkWordCount作為研究Spark Streaming的入口程序。
- object NetworkWordCount {
- def main(args: Array[String]) {
- if (args.length < 2) {
- System.err.println("Usage: NetworkWordCount <hostname> <port>"))
- System.exit(1)
- }
- StreamingExamples.setStreamingLogLevels()
- val sparkConf = new SparkConf().setAppName("NetworkWordCount")
- /*創(chuàng)建StreamingContext對(duì)象，形成整個(gè)程序的上下文*/
- val ssc = new StreamingContext(sparkConf, Seconds(1))
- /*通過socketTextStream接收源源不斷地socket文本流*/
- val lines = ssc.socketTextStream(args(0), args(1).toInt, StorageLevel.MEMORY_AND_DISK_SER)
- val words = lines.flatMap(_.split(" "))
- val wordCounts = words.map(x => (x， 1)).reduceByKey(_ + _)
- wordCounts.print()
- ssc.start()
- ssc.awaitTermination()
- }
- }
(2)進(jìn)入scoketTextStream
- def socketTextStream(hostname:String,port:Int,storageLevel:StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2):ReceiverInputDStream[String] = {
- /*內(nèi)部實(shí)際調(diào)用的socketStream方法 */
- socketStream[String](hostname, port, SocketReceiver.bytesToLines, storageLevel)
- }
- /*進(jìn)入socketStream方法 */
- def socketStream[T： ClassTag](hostname:String, port:Int, converter： (InputStream) => Iterator[T], storageLevel： StorageLevel )： ReceiverInputDStream[T] = {
- /*此處初始化SocketInputDStream對(duì)象 */
- new SocketInputDStream[T](this， hostname， port， converter， storageLevel)
- }
(3)初始化SocketInputDStream 在之前的Spark Streaming介紹中，讀者已經(jīng)了解到整個(gè)Spark Streaming的調(diào)度靈魂就是DStream的DAG，可以將這個(gè)DStream DAG類比Spark中的RDD DAG，而DStream類比RDD，DStream可以理解為包含各個(gè)時(shí)間段的一個(gè)RDD集合。SocketInputDStream就是一個(gè)DStream。
- private[streaming] class SocketInputDStream[T： ClassTag](
- @transient ssc_ : StreamingContext,host:String,port:Int, bytesToObjects:InputStream => Iterator[T],storageLevel:StorageLevel)extends ReceiverInputDStream[T](ssc_) {
- def getReceiver(): Receiver[T] = {
- new SocketReceiver(host,port,bytesToObjects,storageLevel)
- }
- }
(4)觸發(fā)StreamingContext中的Start()方法上面的步驟基本完成了Spark Streaming的初始化工作。類似于Spark機(jī)制，Spark Streaming也是延遲(Lazy)觸發(fā)的，只有調(diào)用了start()方法，才真正地執(zhí)行了。
- private[streaming] val scheduler = new JobScheduler(this)
- /*StreamingContext中維持著一個(gè)調(diào)度器*/
- def start(): Unit = synchronized {
- ……
- /*啟動(dòng)調(diào)度器*/
- scheduler.start()
- ……
- }
(5)JobScheduler.start()啟動(dòng)調(diào)度器在start方法中初始化了很多重要的組件。
- def start(): Unit = synchronized {
- ……
- /*初始化事件處理Actor,當(dāng)有消息傳遞給Actor時(shí),調(diào)用processEvent進(jìn)行事件處理*/
- eventActor = ssc.env.actorSystem.actorOf(Props(new Actor {
- def receive = {
- case event: JobSchedulerEvent => processEvent(event)
- }
- }), "JobScheduler")
- /*啟動(dòng)監(jiān)聽總線*/
- listenerBus.start()
- receiverTracker = new ReceiverTracker(ssc)
- /*啟動(dòng)接收器的監(jiān)聽器receiverTracker*/
- receiverTracker.start()
- /*啟動(dòng)job生成器*/
- jobGenerator.start()
- ……
- }
(6)ReceiverTracker類
- /*進(jìn)入ReceiverTracker查看*/
- private[streaming] class ReceiverTracker(ssc: StreamingContext) extends Logging {
- val receiverInputStreams = ssc.graph.getReceiverInputStreams()
- def start() = synchronized {
- ……
- val receiverExecutor = new ReceiverLauncher()
- ……
- if (！receiverInputStreams.isEmpty) {
- /*初始化ReceiverTrackerActor */
- actor = ssc.env.actorSystem.actorOf(Props(new ReceiverTrackerActor), "ReceiverTracker")
- /*啟動(dòng)ReceiverLauncher()實(shí)例,(7)中進(jìn)行介紹*/
- receiverExecutor.start()
- ……
- }
- }
- /*讀者可以先參考ReceiverTrackerActor的代碼查看實(shí)現(xiàn)注冊(cè)Receiver和注冊(cè)Block元數(shù)據(jù)信息的功能。 */
- private class ReceiverTrackerActor extends Actor {
- def receive = {
- /*接收注冊(cè)receiver的消息,每個(gè)receiver就是一個(gè)輸入流接收器,Receiver分布在Worker節(jié)點(diǎn),一個(gè)Receiver接收一個(gè)輸入流,一個(gè)Spark Streaming集群可以有多個(gè)輸入流 */
- case RegisterReceiver(streamId, typ, host, receiverActor) => registerReceiver(streamId, typ, host, receiverActor, sender)
- sender ! true case AddBlock(receivedBlockInfo) => addBlocks(receivedBlockInfo)
- ……
- }
- }
(7)receivelauncher類，在集群上分布式啟動(dòng)接收器
- class ReceiverLauncher {
- ……
- @transient val thread = new Thread() {
- override def run() {
- ……
- /*啟動(dòng)ReceiverTrackerActor已經(jīng)注冊(cè)的Receiver*/
- startReceivers()
- ……
- }
- }
- }
下面進(jìn)入startReceivers方法，方法中將Receiver集合轉(zhuǎn)變?yōu)镽DD，從而在集群上打散，分布式分布。如圖8-13所示，一個(gè)集群可以分布式地在不同的Worker節(jié)點(diǎn)接收輸入數(shù)據(jù)流。
- private def startReceivers() {
- /*獲取之前配置的接收器 */
- val receivers = receiverInputStreams.map(nis => {
- val rcvr = nis.getReceiver()
- rcvr.setReceiverId(nis.id)
- cvr
- })
- ……
- /* 創(chuàng)建并行的在不同Worker節(jié)點(diǎn)分布的receiver集合 */
- val tempRDD = if (hasLocationPreferences) {
- val receiversWithPreferences = receivers.map(r => (r, Seq(r.preferredLocation.get)))
- ssc.sc.makeRDD[Receiver[_]](receiversWithPreferences)
- } else {
- /*在這里創(chuàng)造RDD相當(dāng)于進(jìn)入SparkContext.makeRDD,此經(jīng)典之處在于將receivers集合作為一個(gè)RDD [Receiver]進(jìn)行分區(qū)。即使只有一個(gè)輸入流,按照分布式分區(qū)方式,也是將輸入分布在Worker端,而不在Master*/
- ssc.sc.makeRDD(receivers, receivers.size)
- /*調(diào)用Sparkcontext中的makeRDD方法,本質(zhì)是調(diào)用將數(shù)據(jù)分布式化的方法parallelize*/
- /* def makeRDD[T: ClassTag](seq: Seq[T], numSlices: Int = defaultParallelism): //RDD[T] = { parallelize(seq, numSlices) */
- /*在RDD[Receiver[_]]每個(gè)分區(qū)的每個(gè)Receiver 上都同時(shí)啟動(dòng),這樣其實(shí)Spark Streaming可以構(gòu)建大量的分布式輸入流 */
- val startReceiver = (iterator: Iterator[Receiver[_]]) => {
- if (!iterator.hasNext) {
- throw new SparkException( "Could not start receiver as object not found.")
- }
- val receiver = iterator.next()
- /*此處的supervisorImpl是一個(gè)監(jiān)督者的角色,在下面的內(nèi)容中將會(huì)剖析這個(gè)對(duì)象的作用 */
- val executor = new ReceiverSupervisorImpl(receiver, SparkEnv.get)
- executor.start()
- executor.awaitTermination()
- }
- /*將receivers的集合打散,然后啟動(dòng)它們 */
- ……
- ssc.sparkContext.runJob(tempRDD, startReceiver)
- ……
- }
2.數(shù)據(jù)接收與轉(zhuǎn)化
在“1.初始化與集群上分布接收器”中介紹了，receiver集合轉(zhuǎn)換為RDD在集群上分布式地接收數(shù)據(jù)流。那么每個(gè)receiver是怎樣接收并處理數(shù)據(jù)流的呢?Spark Streaming數(shù)據(jù)接收與轉(zhuǎn)化的示意圖如圖8-14所示。圖8-14的主要流程如下。
1)數(shù)據(jù)緩沖：在Receiver的receive函數(shù)中接收流數(shù)據(jù)，將接收到的數(shù)據(jù)源源不斷地放入BlockGenerator.currentBuffer。
2)緩沖數(shù)據(jù)轉(zhuǎn)化為數(shù)據(jù)塊：在BlockGenerator中有一個(gè)定時(shí)器(recurring timer)，將當(dāng)前緩沖區(qū)中的數(shù)據(jù)以用戶定義的時(shí)間間隔封裝為一個(gè)數(shù)據(jù)塊Block，放入BlockGenerator的blocksForPush隊(duì)列中。
3)數(shù)據(jù)塊轉(zhuǎn)化為Spark數(shù)據(jù)塊：在BlockGenerator中有一個(gè)BlockPushingThread線程，不斷地將blocksForPush隊(duì)列中的塊傳遞給Blockmanager，讓BlockManager將數(shù)據(jù)存儲(chǔ)為塊，讀者可以在本書的Spark IO章節(jié)了解Spark的底層存儲(chǔ)機(jī)制。BlockManager負(fù)責(zé)Spark中的塊管理。
4)元數(shù)據(jù)存儲(chǔ)：在pushArrayBuffer方法中還會(huì)將已經(jīng)由BlockManager存儲(chǔ)的元數(shù)據(jù)信息(如Block的ID號(hào))傳遞給ReceiverTracker，ReceiverTracker將存儲(chǔ)的blockId放到對(duì)應(yīng)StreamId的隊(duì)列中。 上面過程中涉及最多的類就是BlockGenerator，在數(shù)據(jù)轉(zhuǎn)化的過程中，其扮演著不可或缺的角色。
- private[streaming] class BlockGenerator（ listener： BlockGeneratorListener， receiverId： Int， conf： SparkConf ） extends Logging
感興趣的讀者可以參照?qǐng)D8-14中的類和方法更加具體地了解機(jī)制。由于篇幅所限，這個(gè)數(shù)據(jù)生成過程的代碼不再具體剖析。
【本文為51CTO專欄作者“王森豐”的原創(chuàng)稿件，轉(zhuǎn)載請(qǐng)注明出處】
