溫馨提示×
您好，登錄后才能下訂單哦！
點擊 登錄注冊 即表示同意《億速云用戶服務條款》
您好，登錄后才能下訂單哦！
Spark2.x中三種ShuffleWriter觸發條件分別是什么，針對這個問題，這篇文章詳細介紹了相對應的分析和解答，希望可以幫助更多想解決這個問題的小伙伴找到更簡單易行的方法。
一、概述
基于Sort的Shuffle Writer有三種實現：BypassMergeSortShuffleWriter、UnsafeShuffleWriter、SortShuffleWriter，下面我們主要看看在Spark2.x是如何選擇使用哪種Shuffler Writer處理的。
二、Shuffle觸發條件
1.我們結合源碼來看ShuffleManager是如何來決定選擇實例化哪一種ShuffleHandle,如果注冊的是SerializedShuffleHandle，就獲取UnsafeShuffleWriter；如果注冊的是BypassMergeSortShuffleHandle，就獲取BypassMergeSortShuffleWriter；如果注冊的是BaseShuffleHandle，就獲取SortShuffleWriter，registerShuffle()函數代碼如下：
override def registerShuffle[K, V, C]( shuffleId: Int, numMaps: Int, dependency: ShuffleDependency[K, V, C]): ShuffleHandle = { /*先判斷是否滿足BypassMergeSortShuffleHandle條件*/ if (SortShuffleWriter.shouldBypassMergeSort(conf, dependency)) { new BypassMergeSortShuffleHandle[K, V]( shuffleId, numMaps, dependency.asInstanceOf[ShuffleDependency[K, V, V]]) /*再判斷是否滿足SerializedShuffleHandle條件*/ } else if (SortShuffleManager.canUseSerializedShuffle(dependency)) { // Otherwise, try to buffer map outputs in a serialized form, since this is more efficient: new SerializedShuffleHandle[K, V]( shuffleId, numMaps, dependency.asInstanceOf[ShuffleDependency[K, V, V]]) /*最后才是BaseShuffleHandle*/ } else { // Otherwise, buffer map outputs in a deserialized form: new BaseShuffleHandle(shuffleId, numMaps, dependency) } }
2.上面registerShuffle()函數的各個策略的判斷條件，這里詳細說一下：
1).BypassMergeSortShuffleWriter(對應BypassMergeSortHandle)的觸發條件如下：
a.map端不能進行aggregation聚合操作
b.不能指定ordering，即分區內數據不能排序
c.分區的數目必須小于 spark.shuffle.sort.bypassMergeThrshold指定的閥值，默認值是200
2).UnsafeShuffleWriter觸發條件(對應SerializedShuffleHandle)的觸發條件如下：
a.shuffle依賴不帶有聚合操作
b.支持序列化值的重新定位
c.分區數量少于16777216個
3).SortShuffleWriter(對應BaseShuffleHandle)
上面如果不滿足，默認是這個策略，用于處理所有其他情況。
關于Spark2.x中三種ShuffleWriter觸發條件分別是什么問題的解答就分享到這里了，希望以上內容可以對大家有一定的幫助，如果你還有很多疑惑沒有解開，可以關注億速云行業資訊頻道了解更多相關知識。
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
