本期內容：

　　1. Spark Streaming?Job架構與運行機制

　　2.?Spark Streaming?容錯架構與運行機制

?

　　事實上時間是不存在的，是由人的感官系統感覺時間的存在而已，是一種虛幻的存在，任何時候宇宙中的事情一直在發生著的。

　　Spark Streaming好比時間，一直遵循其運行機制和架構在不停的在運行，無論你寫多或者少的應用程序都跳不出這個范圍。

?

import org.apache.spark.SparkConf import org.apache.spark.streaming.{Seconds, StreamingContext}/*** 使用Scala開發集群運行的Spark 在線黑名單過濾程序* 背景描述：在廣告點擊計費系統中，我們在線過濾掉黑名單的點擊，進而保護廣告商的利益，只進行有效的廣告點擊計費* 或者在防刷評分（或者流量）系統，過濾掉無效的投票或者評分或者流量；* 實現技術：使用transform Api直接基于RDD編程，進行join操作
　**/ object OnlineForeachRDD2DB {def main(args: Array[String]){/*** 創建Spark的配置對象SparkConf，設置Spark程序的運行時的配置信息，* 例如說通過setMaster來設置程序要鏈接的Spark集群的Master的URL,如果設置* 為local，則代表Spark程序在本地運行，特別適合于機器配置條件非常差（例如* 只有1G的內存）的初學者 **/val conf = new SparkConf() //創建SparkConf對象conf.setAppName("OnlineForeachRDD") //設置應用程序的名稱，在程序運行的監控界面可以看到名稱//conf.setMaster("spark://Master:7077") //此時，程序在Spark集群conf.setMaster("local[6]")//設置batchDuration時間間隔來控制Job生成的頻率并且創建Spark Streaming執行的入口val ssc = new StreamingContext(conf, Seconds(5))val lines = ssc.socketTextStream("Master", 9999)val words = lines.flatMap(_.split(" "))val wordCounts = words.map(x => (x, 1)).reduceByKey(_ + _)wordCounts.foreachRDD { rdd =>rdd.foreachPartition { partitionOfRecords => {val connection = ConnectionPool.getConnection()partitionOfRecords.foreach(record => {val sql = "insert into streaming_itemcount(item,count) values('" + record._1 + "'," + record._2 + ")"val stmt = connection.createStatement();stmt.executeUpdate(sql);})ConnectionPool.returnConnection(connection) // return to the pool for future reuse }}}ssc.start()ssc.awaitTermination()} }

?

一.?? 通過案例透視Job執行過程的Spark Streaming機制解析，案例代碼如下：

　　通過運行以上代碼對Job運行機制進行解析：

　　　　1.? 首先通過StreamingContext調用start方法，其內部再啟動JobScheduler的Start方法，進行消息循環；

　　　　2.? 在JobScheduler的start內部會構造JobGenerator和ReceiverTacker；

　　　　3.??然后調用JobGenerator和ReceiverTacker的start方法執行以下操作：

　　　　01.? JobGenerator啟動后會不斷的根據batchDuration生成一個個的Job?；

　　　　02.? ReceiverTracker啟動后首先在Spark Cluster中啟動Receiver（其實是在Executor中先啟動ReceiverSupervisor）；

　　　　4.? 在Receiver收到數據后會通過ReceiverSupervisor存儲到Executor?；

　　　　5.??同時把數據的Metadata信息發送給Driver中的ReceiverTracker，在ReceiverTracker內部會通過ReceivedBlockTracker來管理接受到的元數據信息；

　　　　6.? 每個BatchInterval會產生一個具體的Job，其實這里的Job不是Spark Core中所指的Job，它只是基于DStreamGraph而生成的RDD的DAG而已；

　　　　7.? 要想運行Job需要提交給JobScheduler，在JobScheduler中通過線程池的方式找到一個單獨的線程來提交Job到集群運行，在線程中基于RDD的Action觸發作業的運行；

　　　　8.??由于流處理過程中作業不斷生成，為了提升效率，可以使用線程池。同時有可能設置了Job的FAIR公平調度的方式，也需要多線程的支持；

　　

?二.? 從容錯架構的角度透視Spark Streaming?運行機制：

　　Spark Streaming是基于DStream的容錯機制，DStream是隨著時間流逝不斷的產生RDD，也就是說DStream是在固定的時間上操作RDD，容錯會劃分到每一次所形成的RDD。

　　Spark Streaming的容錯包括 Executor?與 ?Driver兩方面的容錯機制?：

　　1.? Executor?容錯：?

　　　　01.? 數據接收：分布式方式、wal方式，先寫日志再保存數據到Executor?

　　　　02. 任務執行安全性 Job基于RDD容錯?：

　　2. Driver容錯?： checkpoint?。

　　基于RDD的特性，它的容錯機制主要就是兩種：

　　　　01.? 基于checkpoint；

　　　　　　在stage之間，是寬依賴，產生了shuffle操作，lineage鏈條過于復雜和冗長，這時候就需要做checkpoint。

　　　　02.? 基于lineage（血統）的容錯：

　　　　　　一般而言，spark選擇血統容錯，因為對于大規模的數據集，做檢查點的成本很高。

　　　　　　考慮到RDD的依賴關系，每個stage內部都是窄依賴，此時一般基于lineage容錯，方便高效。

　　總結： stage內部做lineage，stage之間做checkpoint。

?

轉載于:https://www.cnblogs.com/yinpin2011/p/5457197.html

總結

以上是生活随笔為你收集整理的通过案例对 spark streaming 透彻理解三板斧之三：spark streaming运行机制与架构的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。