前言

說人話：其實(shí)就是講Spark Streaming 的好處與坑。好處主要從一些大的方面講，坑則是從實(shí)際場(chǎng)景中遇到的一些小細(xì)節(jié)描述。

玫瑰篇

玫瑰篇主要是說Spark Streaming的優(yōu)勢(shì)點(diǎn)。

玫瑰之代碼復(fù)用

這主要得益于Spark的設(shè)計(jì)，以及平臺(tái)的全面性。你寫的流處理的代碼可以很方便的適用于Spark平臺(tái)上的批處理，交互式處理。因?yàn)樗麄儽旧矶际腔赗DD模型的，并且Spark Streaming的設(shè)計(jì)者也做了比較好的封裝和兼容。所以我說RDD是個(gè)很強(qiáng)大的框，能把各種場(chǎng)景都給框住，這就是高度抽象和思考后的結(jié)果。

玫瑰之機(jī)器學(xué)習(xí)

如果你使用Spark MLlib 做模型訓(xùn)練。恭喜你，首先是很多算法已經(jīng)支持Spark Streaming，譬如k-means 就支持流式數(shù)據(jù)更新模型。 其次，你也可以在Spark Streaming中直接將離線計(jì)算好的模型load進(jìn)來，然后對(duì)新進(jìn)來的數(shù)據(jù)做實(shí)時(shí)的Predict操作。

玫瑰之SQL支持

Spark Streaming 里天然就可以使用 sql/dataframe/datasets 等。而且時(shí)間窗口的使用可以極大擴(kuò)展這種使用場(chǎng)景，譬如各種系統(tǒng)預(yù)警等。類似Storm則需要額外的開發(fā)與支持。

玫瑰之吞吐和實(shí)時(shí)的有效控制

Spark Streaming 可以很好的控制實(shí)時(shí)的程度(小時(shí)，分鐘，秒)。極端情況可以設(shè)置到毫秒。

玫瑰之概述

Spark Streaming 可以很好的和Spark其他組件進(jìn)行交互，獲取其支持。同時(shí)Spark 生態(tài)圈的快速發(fā)展，亦能從中受益。

刺篇

刺篇就是描述Spark Streaming 的一些問題，做選型前關(guān)注這些問題可以有效的降低使用風(fēng)險(xiǎn)。

checkpoint 之刺

checkpoint 是個(gè)很好的恢復(fù)機(jī)制。但是方案比較粗暴，直接通過序列化的機(jī)制寫入到文件系統(tǒng)，導(dǎo)致代碼變更和配置變更無法生效。實(shí)際場(chǎng)景是升級(jí)往往比系統(tǒng)崩潰的頻率高太多。但是升級(jí)需要能夠無縫的銜接上一次的偏移量。所以spark streaming在無法容忍數(shù)據(jù)有丟失的情況下，你需要自己記錄偏移量，然后從上一次進(jìn)行恢復(fù)。

我們目前是重寫了相關(guān)的代碼，每次記錄偏移量，不過只有在升級(jí)的時(shí)候才會(huì)讀取自己記錄的偏移量，其他情況都是依然采用checkpoint機(jī)制。

Kafka 之刺

這個(gè)和Spark Streaming相關(guān)，也不太相關(guān)。說相關(guān)是因?yàn)镾park 對(duì)很多異常處理比較簡(jiǎn)單。很多是和Kafka配置相關(guān)的。我舉個(gè)例子：

如果消息體太大了，超過 fetch.message.max.bytes=1m ,那么Spark Streaming會(huì)直接拋出OffsetOutOfRangeException異常，然后停止服務(wù)。

對(duì)應(yīng)的錯(cuò)誤會(huì)從這行代碼拋出：

if (!iter.hasNext) { 
assert(requestOffset == part.untilOffset, errRanOutBeforeEnd(part)) 
finished = true  
null.asInstanceOf[R]  
} 

其實(shí)就是消費(fèi)的完成后 實(shí)際的消費(fèi)數(shù)據(jù)量和預(yù)先估計(jì)的量不一致。

你在日志中看到的信息其實(shí)是這個(gè)代碼答應(yīng)出來的：

private def errRanOutBeforeEnd(part: KafkaRDDPartition): String =

s"Ran out of messages before reaching ending offset ${part.untilOffset} " +

s"for topic ${part.topic} partition ${part.partition} start ${part.fromOffset}." +

" This should not happen, and indicates that messages may have been lost"

解決辦法自然是把 fetch.message.max.bytes 設(shè)置大些。

如果你使用Spark Streaming去追數(shù)據(jù)，從頭開始消費(fèi)kafka,而Kafka因?yàn)槟撤N原因，老數(shù)據(jù)快速的被清理掉，也會(huì)引發(fā)OffsetOutOfRangeException錯(cuò)誤。并且使得Spark Streaming程序異常的終止。

解決辦法是事先記錄kafka偏移量和時(shí)間的關(guān)系(可以隔幾秒記錄一次)，然后根據(jù)時(shí)間找到一個(gè)較大的偏移量開始消費(fèi)。

或者你根據(jù)目前Kafka新增數(shù)據(jù)的消費(fèi)速度，給smallest獲取到的偏移量再加一個(gè)較大的值，避免出現(xiàn)Spark Streaming 在fetch的時(shí)候數(shù)據(jù)不存在的情況。

Kafka partition 映射 RDD partition 之刺

Kafka的分區(qū)數(shù)決定了你的并行度(我們假設(shè)你使用Direct Approach的模式集成)。為了獲得更大的并行度，則需要進(jìn)行一次repartition，而repartition 就意味著需要發(fā)生Shuffle,在流式計(jì)算里，可能會(huì)消耗掉我們寶貴的時(shí)間。

為了能夠避免Shuffle,并且提高Spark Streaming處理的并行度，我們重寫了 DirectKafkaInputDStream,KafkaRDD,KafkaUtils等類，實(shí)現(xiàn)了一個(gè)Kafka partition 可以映射為多個(gè)RDD partition的功能。譬如你有M個(gè)Kafka partitions,則可映射成 M*N個(gè) RDD partitions。 其中N 為>1 的正整數(shù)。

我們期望官方能夠?qū)崿F(xiàn)將一個(gè)Kafka的partitions 映射為多個(gè)Spark 的partitions,避免發(fā)生Shuffle而導(dǎo)致多次的數(shù)據(jù)移動(dòng)。

textFileStream

其實(shí)使用textFileStream 的人應(yīng)該也不少。因?yàn)榭梢院芊奖愕谋O(jiān)控HDFS上某個(gè)文件夾下的文件，并且進(jìn)行計(jì)算。這里我們遇到的一個(gè)問題是，如果底層比如是壓縮文件，遇到有順壞的文件，你是跳不過去的，直接會(huì)讓Spark Streaming 異常退出。 官方并沒有提供合適的方式讓你跳過損壞的文件。

以NewHadoopRDD為例，里面有這么幾行代碼，獲取一條新的數(shù)據(jù)：

override def getNext(): (K, V) = { 
try { 
finished = !reader.next(key, value) 
} catch { 
case eof: EOFException => 
finished = true 
} 
if (!finished) { 
inputMetrics.incRecordsRead(1) 
} 
(key, value) 
} 

通過reader 獲取下一條記錄的時(shí)候，譬如是一個(gè)損壞的gzip文件，可能就會(huì)拋出異常，而這個(gè)異常是用戶catch不到的，直接讓Spark Streaming程序掛掉了。

而在 HadoopRDD類中，對(duì)應(yīng)的實(shí)現(xiàn)如下：

override def getNext(): (K, V) = { 
try { 
finished = !reader.next(key, value) 
} catch { 
case eof: EOFException => 
finished = true 
} 
if (!finished) { 
inputMetrics.incRecordsRead(1) 
} 
(key, value) 
} 

這里好歹做了個(gè)EOFException。然而，如果是一個(gè)壓縮文件，解壓的時(shí)候就直接產(chǎn)生錯(cuò)誤了，一般而言是 IOException,而不是EOFException了，這個(gè)時(shí)候也就歇菜了。

個(gè)人認(rèn)為應(yīng)該添加一些配置，允許用戶可以選擇如何對(duì)待這種有損壞或者無法解壓的文件。

因?yàn)楝F(xiàn)階段我們并沒有維護(hù)一個(gè)Spark的私有版本，所以是通過重寫FileInputDStream,NewHadoopRDD 等相關(guān)類來修正該問題。

Shuffle 之刺

Shuffle (尤其是每個(gè)周期數(shù)據(jù)量很大的情況)是Spark Streaming 不可避免的疼痛,尤其是數(shù)據(jù)量極大的情況，因?yàn)镾park Streaming對(duì)處理的時(shí)間是有限制的。我們有一個(gè)場(chǎng)景，是五分鐘一個(gè)周期，我們僅僅是做了一個(gè)repartion，耗時(shí)就達(dá)到2.1分鐘(包括到 Kafka取數(shù)據(jù))。現(xiàn)階段Spark 的Shuffle實(shí)現(xiàn)都需要落磁盤，并且Shuffle Write 和 Shuffle Read 階段是完全分開，后者必須等到前者都完成才能開始工作。我認(rèn)為Spark Streaming有必要單獨(dú)開發(fā)一個(gè)更快速，完全基于內(nèi)存的Shuffle方案。

內(nèi)存之刺

在Spark Streaming中，你也會(huì)遇到在Spark中常見的問題，典型如Executor Lost 相關(guān)的問題(shuffle fetch 失敗，Task失敗重試等)。這就意味著發(fā)生了內(nèi)存不足或者數(shù)據(jù)傾斜的問題。這個(gè)目前你需要考慮如下幾個(gè)點(diǎn)以期獲得解決方案：

相同資源下，增加partition數(shù)可以減少內(nèi)存問題。 原因如下：通過增加partition數(shù)，每個(gè)task要處理的數(shù)據(jù)少了，同一時(shí)間內(nèi)，所有正在 運(yùn)行的task要處理的數(shù)量少了很多，所有Executor占用的內(nèi)存也變小了。這可以緩解數(shù)據(jù)傾斜以及內(nèi)存不足的壓力。

關(guān)注shuffle read 階段的并行數(shù)。例如reduce,group 之類的函數(shù)，其實(shí)他們都有第二個(gè)參數(shù)，并行度(partition數(shù))，只是大家一般都不設(shè)置。不過出了問題再設(shè)置一下，也不錯(cuò)。

給一個(gè)Executor 核數(shù)設(shè)置的太多，也就意味著同一時(shí)刻，在該Executor 的內(nèi)存壓力會(huì)更大，GC也會(huì)更頻繁。我一般會(huì)控制在3個(gè)左右。然后通過提高Executor數(shù)量來保持資源的總量不變。

監(jiān)控之刺

Spark Streaming 的UI 上的Executors Tab缺少一個(gè)***的監(jiān)控，就是Worker內(nèi)存GC詳情。雖然我們可以將這些信息導(dǎo)入到 第三方監(jiān)控中，然而終究是不如在 Spark UI上展現(xiàn)更加方便。 為此我們也將該功能列入研發(fā)計(jì)劃。

總結(jié)

目前Spark Streaming 可以應(yīng)對(duì)的場(chǎng)景不少，但是在很多場(chǎng)景上，還是有這樣那樣的問題。建議調(diào)研后都進(jìn)一步做測(cè)試再做出是否遷移到該平臺(tái)的決定。