背景：測試環(huán)境運(yùn)行一個SparkStreaming任務(wù)，yarn-cluster模式，duration為5分鐘一個批次，每個批次平均2000w條records，并行度為60

資源配置為：

${SPARK_HOME}/bin/spark-submit --name ${jobname} --driver-cores 3 --driver-memory 6g --num-executors 24 --executor-memory 6g --executor-cores 2

問題：
觀察了前幾個批次的任務(wù)，運(yùn)行正常，效率也能趕得上批次，平均每個批次的task都在4分鐘以內(nèi)處理完了，結(jié)果數(shù)據(jù)也沒問題，心想應(yīng)該沒問題了，就下班了。結(jié)果第二天上班后再觀察，發(fā)現(xiàn)數(shù)據(jù)延遲了幾個小時，任務(wù)卡在了某個批次，再看這個批次的詳細(xì)情況，發(fā)現(xiàn)除了運(yùn)行在某臺服務(wù)器上的task還在running，其他的task都已經(jīng)完成了，而這臺服務(wù)器上的task已經(jīng)running了幾個小時了...

嘗試解決問題：
首先登陸這臺服務(wù)器查看yarn的container日志，并沒有發(fā)現(xiàn)ERROR，再看了看這臺服務(wù)器的資源，cpu和內(nèi)存使用情況，也是正常的！頓時有點摸不著頭腦，沒有報錯，任務(wù)就卡在那兒了。

重啟大法：
重啟大法好，秉承著重啟解決一切問題的思想，我把SparkStreaming任務(wù)重啟了，萬一是偶然現(xiàn)象呢。
結(jié)果沒過兩個小時就被打臉了，同樣的問題又出現(xiàn)了。

查看日志：
沒有ERROR日志，那就看看INFO日志，任務(wù)在哪個環(huán)節(jié)卡住的，結(jié)果發(fā)現(xiàn)了以下日志內(nèi)容

INFO executor.Executor: Finished task 51.0 in stage 148.0 (TID 8931). 903 bytes result sent to driver
INFO storage.BlockManager: Dropping broadcast blocks older than 1541581868061
INFO util.MetadataCleaner: Ran metadata cleaner for BROADCAST_VARS
INFO storage.BlockManager: Dropping non broadcast blocks older than 1541581868066
INFO util.MetadataCleaner: Ran metadata cleaner for BLOCK_MANAGER
INFO storage.BlockManager: Dropping broadcast blocks older than 1541582048061
INFO util.MetadataCleaner: Ran metadata cleaner for BROADCAST_VARS
INFO storage.BlockManager: Dropping non broadcast blocks older than 1541582048066
INFO util.MetadataCleaner: Ran metadata cleaner for BLOCK_MANAGER
INFO storage.BlockManager: Dropping broadcast blocks older than 1541582228061
...

在卡死的服務(wù)器的yarn stderr日志中，卡死批次的時間點找到以上日志內(nèi)容，每三分鐘出現(xiàn)一次，正常的日志內(nèi)容應(yīng)該是如下所示：

INFO executor.Executor: Finished task 28.0 in stage 145.0 (TID 8728). 1852 bytes result sent to driver
INFO storage.BlockManager: Dropping broadcast blocks older than 1541581328061
INFO storage.BlockManager: Dropped block broadcast_133_piece0
INFO storage.BlockManager: Dropping non broadcast blocks older than 1541581328066
INFO storage.BlockManager: Dropped block broadcast_133
INFO util.MetadataCleaner: Ran metadata cleaner for BLOCK_MANAGER
INFO util.MetadataCleaner: Ran metadata cleaner for BROADCAST_VARS
INFO output.FileOutputCommitter: Saved output of task 'attempt_201811071730_0218_r_000037_0' to hdfs://xxxxxxx/xxxxxxx/log/_temporary/0/task_201811071730_0218_r_000037
INFO executor.Executor: Finished task 37.0 in stage 145.0 (TID 8737). 1852 bytes result sent to driver
INFO output.FileOutputCommitter: Saved output of task 'attempt_201811071730_0218_r_000058_0' to hdfs://xxxxxxx/xxxxxxx/log/_temporary/0/task_201811071730_0218_r_000058
INFO executor.Executor: Finished task 58.0 in stage 145.0 (TID 8758). 1852 bytes result sent to driver
INFO storage.BlockManager: Removing RDD 213

MetadataCleaner用于定期清理persist的RDD緩存和stage task中產(chǎn)生的元數(shù)據(jù)，實質(zhì)上是一個TimerTask實現(xiàn)的定時器，其中BLOCK_MANAGER和BROADCAST_VARS屬于MetadataCleanerType枚舉類的其中兩個元數(shù)據(jù)類別，代表blockmanager中非broadcast的元數(shù)據(jù)部分和broadcast的元數(shù)據(jù)部分。

但是搞不懂為什么會一直重復(fù)的在清理，導(dǎo)致task卡死，查閱了網(wǎng)上各種文檔也沒有類似的問題解決方法。

于是在想有沒有辦法能夠跳過這個task或者讓這個task重啟，最后想到了Spark的推測執(zhí)行。
Spark推測執(zhí)行就是適用于個別task比其他的task慢的情況，當(dāng)某些個task特別慢的時候（滿足條件），Spark就重啟一個task處理同樣的一份數(shù)據(jù)，誰先處理好就用誰的數(shù)據(jù)，把另外一個task殺掉，正好可以解決我的問題。

配置推測執(zhí)行：
sparkConf.set("spark.speculation", "true")
sparkConf.set("spark.speculation.interval", "300s")
sparkConf.set("spark.speculation.quantile","0.9")

spark.speculation設(shè)置為true表示打開推測執(zhí)行功能
spark.speculation.interval表示檢測周期，spark會開啟一個線程來檢測是否需要推測執(zhí)行
spark.speculation.quantile表示閾值，設(shè)為0.9表示該批次所有的task有90%執(zhí)行完成即對剩余的task執(zhí)行推測執(zhí)行
spark.speculation.multiplier默認(rèn)值為1.5，表示慢的task執(zhí)行時間比完成的task平均時間多耗費1.5倍開啟推測執(zhí)行

這里把檢測周期設(shè)為5分鐘是為了防止資源浪費，過了批次時間之后再檢測是否有卡死的task，畢竟task卡死是偶爾出現(xiàn)，當(dāng)卡死情況出現(xiàn)時，是個別幾個task出現(xiàn)問題，不會超過60個task總數(shù)的10%，所以把閾值設(shè)為0.9，這樣設(shè)置不會在正常批次開啟推測執(zhí)行，從而節(jié)約了資源。

總結(jié)：
通過巧妙的設(shè)置推測執(zhí)行，這個問題暫時被我繞開了，等有時間一定會研究MetadataCleaner的源碼，嘗試正面解決問題，立個flag。