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開銷。