spark利用cache優化shuffle

cache表，資料放記憶體，資料被廣播到executor，

將多份資料進行關聯是資料處理過程中非常普遍的用法，不過在分布式計算系統中，這個問題往往會變的非常麻煩，因為框架提供的 join 操作一般會將所有資料根據 key 傳送到所有的 reduce 分割槽中去，也就是 shuffle 的過程。造成大量的網路以及磁碟io消耗，執行效率極其低下，這個過程一般被稱為 reduce-side-join。

如果其中有張表較小的話，我們則可以自己實現在 map 端實現資料關聯，跳過大量資料進行 shuffle 的過程，執行時間得到大量縮短，根據不同資料可能會有幾倍到數十倍的效能提公升，這個過程是map-side-join。

reduce-side-join 的缺陷在於會將key相同的資料傳送到同乙個partition中進行運算，大資料集的傳輸需要長時間的io，同時任務併發度收到限制，還可能造成資料傾斜。

reduce-side-join 執行圖如下

map-side-join 執行圖如下

//快取全表
sqlcontext.sql("cache table activity")
//快取過濾結果
sqlcontext.sql("cache table activity_cached as select * from activity where ...")

cache table 是即時生效的，如果你想等到乙個action操作再快取資料可以使用 cache lazy table,這樣操作會直到乙個 action 操作才被觸發，例如 count(*)

sqlcontext.sql("cache lazy table ...")

取消hive表快取資料

sqlcontext.sql("uncache table activity")

示例:

我們也需要注意cachetable與uncachetable的使用時機，cachetable主要用於快取中間表結果，它的特點是少量資料且被後續計算（sql）頻繁使用；如果中間表結果使用完畢，我們應該立即使用uncachetable釋放快取空間，用於快取其它資料

val df = sqlcontext.sql("select * from activity")
df.registertemptable("activity_cached")
sqlcontext.cachetable("activity_cached")
tip:cachetable操作是lazy的，需要乙個action操作來觸發快取操作。

對應的uncachetable可以取消快取

sqlcontext.uncachetable("activity_cached")

val df = sqlcontext.sql("select * from tablename")
df.cache()

added rdd_xx_x in memory on ...

如果記憶體不足，則會存入磁碟中,提示如下:

added rdd_xx_x on disk on ...

快取資料後可以在storage上看到快取的資料

spark.sql.autobroadcastjointhreshold

該引數預設為10m,在進行join等聚合操作時，將小於該值的表broadcast到每台worker，消除了大量的shuffle操作。

spark.rdd.compress true

將rdd存入mem或disk前再進行一次壓縮，效果顯著，我使用cachetable了一張表，沒有開啟該引數前總共cache了54g資料,開啟這個引數後只34g,可是執行速度並沒有收到太大的影響。

spark.sql.shuffle.partitions

這個引數預設為200，是join等聚合操作的並行度，如果有大量的資料進行操作，造成單個任務比較重,執行時間過長的時候，會報如下的錯誤:

org.apache.spark.shuffle.fetchfailedexception: connection from /192.168.xx.***:53450 closed

這個時候需要提高該值。

spark利用cache優化shuffle

spark的cache和checkpoint的區別

spark中cache和checkpoint使用

效能優化之cache

spark利用cache優化shuffle

spark的cache和checkpoint的區別

spark中cache和checkpoint使用

效能優化之cache

相關推薦