本期內(nèi)容：
1. Spark Streaming Job架構(gòu)與運(yùn)行機(jī)制
2. Spark Streaming 容錯架構(gòu)與運(yùn)行機(jī)制
事實上時間是不存在的，是由人的感官系統(tǒng)感覺時間的存在而已，是一種虛幻的存在，任何時候宇宙中的事情一直在發(fā)生著的。
Spark Streaming好比時間，一直遵循其運(yùn)行機(jī)制和架構(gòu)在不停的在運(yùn)行，無論你寫多或者少的應(yīng)用程序都跳不出這個范圍。
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* 使用Scala開發(fā)集群運(yùn)行的Spark 在線黑名單過濾程序
* 背景描述：在廣告點擊計費系統(tǒng)中，我們在線過濾掉黑名單的點擊，進(jìn)而保護(hù)廣告商的利益，只進(jìn)行有效的廣告點擊計費
* 或者在防刷評分（或者流量）系統(tǒng)，過濾掉無效的投票或者評分或者流量；
* 實現(xiàn)技術(shù)：使用transform Api直接基于RDD編程，進(jìn)行join操作
*
* 新浪微博：http://weibo.com/ilovepains/
*/
object OnlineForeachRDD2DB {
def main(args: Array[String]){
/**
* 創(chuàng)建Spark的配置對象SparkConf，設(shè)置Spark程序的運(yùn)行時的配置信息，
* 例如說通過setMaster來設(shè)置程序要鏈接的Spark集群的Master的URL,如果設(shè)置
* 為local，則代表Spark程序在本地運(yùn)行，特別適合于機(jī)器配置條件非常差（例如
* 只有1G的內(nèi)存）的初學(xué)者 *
*/
val conf = new SparkConf() //創(chuàng)建SparkConf對象
conf.setAppName("OnlineForeachRDD") //設(shè)置應(yīng)用程序的名稱，在程序運(yùn)行的監(jiān)控界面可以看到名稱
//conf.setMaster("spark://Master:7077") //此時，程序在Spark集群
conf.setMaster("local[6]")
//設(shè)置batchDuration時間間隔來控制Job生成的頻率并且創(chuàng)建Spark Streaming執(zhí)行的入口
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執(zhí)行過程的Spark Streaming機(jī)制解析，案例代碼如下：
通過運(yùn)行以上代碼對Job運(yùn)行機(jī)制進(jìn)行解析：
1. 首先通過StreamingContext調(diào)用start方法，其內(nèi)部再啟動JobScheduler的Start方法，進(jìn)行消息循環(huán)；
2. 在JobScheduler的start內(nèi)部會構(gòu)造JobGenerator和ReceiverTacker；
3. 然后調(diào)用JobGenerator和ReceiverTacker的start方法執(zhí)行以下操作：
01. JobGenerator啟動后會不斷的根據(jù)batchDuration生成一個個的Job ；
02. ReceiverTracker啟動后首先在Spark Cluster中啟動Receiver（其實是在Executor中先啟動ReceiverSupervisor）；
4. 在Receiver收到數(shù)據(jù)后會通過ReceiverSupervisor存儲到Executor ；
5. 同時把數(shù)據(jù)的Metadata信息發(fā)送給Driver中的ReceiverTracker，在ReceiverTracker內(nèi)部會通過ReceivedBlockTracker來管理接受到的元數(shù)據(jù)信息；
6. 每個BatchInterval會產(chǎn)生一個具體的Job，其實這里的Job不是Spark Core中所指的Job，它只是基于DStreamGraph而生成的RDD的DAG而已；
7. 要想運(yùn)行Job需要提交給JobScheduler，在JobScheduler中通過線程池的方式找到一個單獨的線程來提交Job到集群運(yùn)行，在線程中基于RDD的Action觸發(fā)作業(yè)的運(yùn)行；
8. 由于流處理過程中作業(yè)不斷生成，為了提升效率，可以使用線程池。同時有可能設(shè)置了Job的FAIR公平調(diào)度的方式，也需要多線程的支持；
** 二. 從容錯架構(gòu)的角度透視Spark Streaming 運(yùn)行機(jī)制：**
Spark Streaming是基于DStream的容錯機(jī)制，DStream是隨著時間流逝不斷的產(chǎn)生RDD，也就是說DStream是在固定的時間上操作RDD，容錯會劃分到每一次所形成的RDD。
Spark Streaming的容錯包括 Executor 與 Driver兩方面的容錯機(jī)制 ：
1. Executor 容錯：
01. 數(shù)據(jù)接收：分布式方式、wal方式，先寫日志再保存數(shù)據(jù)到Executor
02. 任務(wù)執(zhí)行安全性 Job基于RDD容錯 ：
2. Driver容錯 ： checkpoint 。
基于RDD的特性，它的容錯機(jī)制主要就是兩種：
01. 基于checkpoint；
在stage之間，是寬依賴，產(chǎn)生了shuffle操作，lineage鏈條過于復(fù)雜和冗長，這時候就需要做checkpoint。
02. 基于lineage（血統(tǒng)）的容錯：
一般而言，spark選擇血統(tǒng)容錯，因為對于大規(guī)模的數(shù)據(jù)集，做檢查點的成本很高。
考慮到RDD的依賴關(guān)系，每個stage內(nèi)部都是窄依賴，此時一般基于lineage容錯，方便高效。
總結(jié)： stage內(nèi)部做lineage，stage之間做checkpoint。
備注：
資料來源于：DT_大數(shù)據(jù)夢工廠
更多私密內(nèi)容，請關(guān)注微信公眾號：DT_Spark
新浪微博：http://www.weibo.com/ilovepains
如果您對大數(shù)據(jù)Spark感興趣，可以免費聽由王家林老師每天晚上20：00開設(shè)的Spark永久免費公開課，地址YY房間號：68917580
