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