SparkStreaming 性能調優

在開發Spark Streaming應用程序時，要結合集羣中各節點的配置狀況儘量地提升數據處理的實時性。在調優的過程當中，一方面要儘量利用集羣資源來減小每一個批處理的時間；另外一方面要確保接收到的數據能及時處理掉。

運行時間優化

• 設置合理的批處理時間和窗口大小

Spark Streaming中做業之間一般存在依賴關係，後面的做業必須確保前面的做業執行結束後才能提交，若前面的做業的執行時間超過了設置的批處理時間間隔，那麼後續的做業將沒法按時提交執行，形成做業的堵塞。也就是說若想Spark Streaming應用程序穩定地在集羣中運行，對於接收到的數據必須儘快處理掉。例如若設定批處理時間爲1秒鐘，那麼系統每1秒鐘生成一個RDD，若是系統計算一個RDD的時間大於1秒，那麼當前的RDD還沒來得及處理，後續的RDD已經提交上來在等待處理了，這就產生了堵塞。所以須要設置一個合理的批處理時間間隔以確保做業可以在這個批處理時間間隔時間內結束。許多實驗數據代表，500毫秒對大多Spark Streaming應用而言是較好的批處理時間間隔。

相似地，對於窗口操做，滑動時間間隔對於性能也有很大的影響。當單批次數據計算代價太高時，能夠考慮適當增大滑動時間間隔。

對於批處理時間和窗口大小的設定，並無統一的標準。一般是先從一個比較大的批處理時間（10秒左右）開始，而後不斷地使用更小的值進行對比測試。若是Spark Streaming用戶界面中顯示的處理時間保持不變，則能夠進一步設定更小的值；若是處理時間開始增長，則可能已經達到了應用的極限，再減少該值則可能會影響系統的性能。

• 提升並行度

提升並行度也是一種減小批處理所消耗時間的常見方法。有如下三種方式能夠提升並行度。一種方法是增長接收器數目。若是獲取的數據太多，則可能致使單個節點來不及對數據進行讀入與分發，使得接收器成爲系統瓶頸。這時能夠經過建立多個輸入DStream來增長接收器數目，而後再使用union來把數據合併爲一個數據源。第二種方法是將收到的數據顯式地從新分區。若是接收器數目沒法再增長，能夠經過使用DStream.repartition、spark.streaming.blocklnterval等參數顯式地對Dstream進行從新分區。第三種方法是提升聚合計算的並行度。對於會致使shuffle的操做，例如reduceByKey、reduceByKeyAndWindow等操做，可經過顯示設置更高的行度參數確保更爲充分地使用集羣資源。

內存使用與垃圾回收

• 控制批處理時間間隔內的數據量

Spark Streaming會把批處理時間間隔內獲取到的全部數據存放在Spark內部可用的內存中。所以必須確保在當前節點上SparkStreaming可用的內存容量至少能容下一個批處理時間間隔內全部的數據。好比一個批處理時間間隔是1秒，可是1秒產生了1GB的數據，那麼要確保當前的節點上至少有可供SparkStreaming使用的1GB內存。

• 及時清理再也不使用的數據

對於內存中處理過的、再也不須要的數據應及時清理，以確保Spark Streaming可以擁有足夠的內存空間可使用。一種方法是能夠經過設置合理的spark.cleaner.ttl時長來及時清理超時的無用數據，但該方法應慎重使用，以避免後續數據在須要時被錯誤清理。另外一種方法是將spark.streaming.unpersist設置爲true，系統將自動清理已經不須要的RDD。該方法能顯著減小RDD對內存的須要，同時潛在地提升GC的性能。此外用戶還能夠經過配置參數streamingContext.remember爲數據設置更長的保留時間。

• 減小序列化與反序列化的負擔

SparkStreaming默認將接收到的數據序列化後放入內存，以減小內存使用。序列化和反序列化須要更多的CPU資源，所以使用適當的序列化工具（例如Kryo）和自定義的序列化接口能夠更高效地使用CPU。除了使用更好的序列化工具外還能夠結合壓縮機制，經過配置spark.rdd.compress，以CPU的時間開銷來換取內存資源，下降GC開銷。