SparkStreaming 性能調(diào)優(yōu)

在開發(fā)Spark Streaming應(yīng)用程序時(shí)，要結(jié)合集群中各節(jié)點(diǎn)的配置情況盡可能地提高數(shù)據(jù)處理的實(shí)時(shí)性。在調(diào)優(yōu)的過程中，一方面要盡可能利用集群資源來減少每個(gè)批處理的時(shí)間；另一方面要確保接收到的數(shù)據(jù)能及時(shí)處理掉。

運(yùn)行時(shí)間優(yōu)化

• 設(shè)置合理的批處理時(shí)間和窗口大小

Spark Streaming中作業(yè)之間通常存在依賴關(guān)系，后面的作業(yè)必須確保前面的作業(yè)執(zhí)行結(jié)束后才能提交，若前面的作業(yè)的執(zhí)行時(shí)間超過了設(shè)置的批處理時(shí)間間隔，那么后續(xù)的作業(yè)將無法按時(shí)提交執(zhí)行，造成作業(yè)的堵塞。也就是說若想Spark Streaming應(yīng)用程序穩(wěn)定地在集群中運(yùn)行，對(duì)于接收到的數(shù)據(jù)必須盡快處理掉。例如若設(shè)定批處理時(shí)間為1秒鐘，那么系統(tǒng)每1秒鐘生成一個(gè)RDD，如果系統(tǒng)計(jì)算一個(gè)RDD的時(shí)間大于1秒，那么當(dāng)前的RDD還沒來得及處理，后續(xù)的RDD已經(jīng)提交上來在等待處理了，這就產(chǎn)生了堵塞。因此需要設(shè)置一個(gè)合理的批處理時(shí)間間隔以確保作業(yè)能夠在這個(gè)批處理時(shí)間間隔時(shí)間內(nèi)結(jié)束。許多實(shí)驗(yàn)數(shù)據(jù)表明，500毫秒對(duì)大多Spark Streaming應(yīng)用而言是較好的批處理時(shí)間間隔。

類似地，對(duì)于窗口操作，滑動(dòng)時(shí)間間隔對(duì)于性能也有很大的影響。當(dāng)單批次數(shù)據(jù)計(jì)算代價(jià)過高時(shí)，可以考慮適當(dāng)增大滑動(dòng)時(shí)間間隔。

對(duì)于批處理時(shí)間和窗口大小的設(shè)定，并沒有統(tǒng)一的標(biāo)準(zhǔn)。通常是先從一個(gè)比較大的批處理時(shí)間（10秒左右）開始，然后不斷地使用更小的值進(jìn)行對(duì)比測試。如果Spark Streaming用戶界面中顯示的處理時(shí)間保持不變，則可以進(jìn)一步設(shè)定更小的值；如果處理時(shí)間開始增加，則可能已經(jīng)達(dá)到了應(yīng)用的極限，再減小該值則可能會(huì)影響系統(tǒng)的性能。

• 提高并行度

提高并行度也是一種減少批處理所消耗時(shí)間的常見方法。有以下三種方式可以提高并行度。一種方法是增加接收器數(shù)目。如果獲取的數(shù)據(jù)太多，則可能導(dǎo)致單個(gè)節(jié)點(diǎn)來不及對(duì)數(shù)據(jù)進(jìn)行讀入與分發(fā)，使得接收器成為系統(tǒng)瓶頸。這時(shí)可以通過創(chuàng)建多個(gè)輸入DStream來增加接收器數(shù)目，然后再使用union來把數(shù)據(jù)合并為一個(gè)數(shù)據(jù)源。第二種方法是將收到的數(shù)據(jù)顯式地重新分區(qū)。如果接收器數(shù)目無法再增加，可以通過使用DStream.repartition、spark.streaming.blocklnterval等參數(shù)顯式地對(duì)Dstream進(jìn)行重新分區(qū)。第三種方法是提高聚合計(jì)算的并行度。對(duì)于會(huì)導(dǎo)致shuffle的操作，例如reduceByKey、reduceByKeyAndWindow等操作，可通過顯示設(shè)置更高的行度參數(shù)確保更為充分地使用集群資源。

內(nèi)存使用與垃圾回收

• 控制批處理時(shí)間間隔內(nèi)的數(shù)據(jù)量

Spark Streaming會(huì)把批處理時(shí)間間隔內(nèi)獲取到的所有數(shù)據(jù)存放在Spark內(nèi)部可用的內(nèi)存中。因此必須確保在當(dāng)前節(jié)點(diǎn)上SparkStreaming可用的內(nèi)存容量至少能容下一個(gè)批處理時(shí)間間隔內(nèi)所有的數(shù)據(jù)。比如一個(gè)批處理時(shí)間間隔是1秒，但是1秒產(chǎn)生了1GB的數(shù)據(jù)，那么要確保當(dāng)前的節(jié)點(diǎn)上至少有可供SparkStreaming使用的1GB內(nèi)存。

• 及時(shí)清理不再使用的數(shù)據(jù)

對(duì)于內(nèi)存中處理過的、不再需要的數(shù)據(jù)應(yīng)及時(shí)清理，以確保Spark Streaming能夠擁有足夠的內(nèi)存空間可以使用。一種方法是可以通過設(shè)置合理的spark.cleaner.ttl時(shí)長來及時(shí)清理超時(shí)的無用數(shù)據(jù)，但該方法應(yīng)慎重使用，以免后續(xù)數(shù)據(jù)在需要時(shí)被錯(cuò)誤清理。另一種方法是將spark.streaming.unpersist設(shè)置為true，系統(tǒng)將自動(dòng)清理已經(jīng)不需要的RDD。該方法能顯著減少RDD對(duì)內(nèi)存的需要，同時(shí)潛在地提高GC的性能。此外用戶還可以通過配置參數(shù)streamingContext.remember為數(shù)據(jù)設(shè)置更長的保留時(shí)間。

• 減少序列化與反序列化的負(fù)擔(dān)

SparkStreaming默認(rèn)將接收到的數(shù)據(jù)序列化后放入內(nèi)存，以減少內(nèi)存使用。序列化和反序列化需要更多的CPU資源，因此使用適當(dāng)?shù)男蛄谢ぞ撸ɡ鏚ryo）和自定義的序列化接口可以更高效地使用CPU。除了使用更好的序列化工具外還可以結(jié)合壓縮機(jī)制，通過配置spark.rdd.compress，以CPU的時(shí)間開銷來換取內(nèi)存資源，降低GC開銷。