Spark數(shù)據(jù)傾斜及其解決方案-創(chuàng)新互聯(lián)

• 本文首發(fā)于 vivo互聯(lián)網(wǎng)技術(shù) 微信公眾號 https://mp.weixin.qq.com/s/lqMu6lfk-Ny1ZHYruEeBdA
• 作者簡介：鄭志彬，畢業(yè)于華南理工大學(xué)計算機(jī)科學(xué)與技術(shù)（雙語班）。先后從事過電子商務(wù)、開放平臺、移動瀏覽器、推薦廣告和大數(shù)據(jù)、人工智能等相關(guān)開發(fā)和架構(gòu)。目前在vivo智能平臺中心從事 AI中臺建設(shè)以及廣告推薦業(yè)務(wù)。擅長各種業(yè)務(wù)形態(tài)的業(yè)務(wù)架構(gòu)、平臺化以及各種業(yè)務(wù)解決方案。

本文從數(shù)據(jù)傾斜的危害、現(xiàn)象、原因等方面，由淺入深闡述Spark數(shù)據(jù)傾斜及其解決方案。

成都創(chuàng)新互聯(lián)公司科技有限公司專業(yè)互聯(lián)網(wǎng)基礎(chǔ)服務(wù)商，為您提供德陽機(jī)房托管，高防主機(jī)，成都IDC機(jī)房托管，成都主機(jī)托管等互聯(lián)網(wǎng)服務(wù)。

一、什么是數(shù)據(jù)傾斜

對 Spark/Hadoop 這樣的分布式大數(shù)據(jù)系統(tǒng)來講，數(shù)據(jù)量大并不可怕，可怕的是數(shù)據(jù)傾斜。

對于分布式系統(tǒng)而言，理想情況下，隨著系統(tǒng)規(guī)模（節(jié)點數(shù)量）的增加，應(yīng)用整體耗時線性下降。如果一臺機(jī)器處理一批大量數(shù)據(jù)需要120分鐘，當(dāng)機(jī)器數(shù)量增加到3臺時，理想的耗時為120 / 3 = 40分鐘。但是，想做到分布式情況下每臺機(jī)器執(zhí)行時間是單機(jī)時的1 / N，就必須保證每臺機(jī)器的任務(wù)量相等。不幸的是，很多時候，任務(wù)的分配是不均勻的，甚至不均勻到大部分任務(wù)被分配到個別機(jī)器上，其它大部分機(jī)器所分配的任務(wù)量只占總得的小部分。比如一臺機(jī)器負(fù)責(zé)處理 80% 的任務(wù)，另外兩臺機(jī)器各處理 10% 的任務(wù)。

『不患多而患不均』，這是分布式環(huán)境下大的問題。意味著計算能力不是線性擴(kuò)展的，而是存在短板效應(yīng): 一個 Stage 所耗費的時間，是由最慢的那個 Task 決定。

由于同一個 Stage 內(nèi)的所有 task 執(zhí)行相同的計算，在排除不同計算節(jié)點計算能力差異的前提下，不同 task 之間耗時的差異主要由該 task 所處理的數(shù)據(jù)量決定。所以，要想發(fā)揮分布式系統(tǒng)并行計算的優(yōu)勢，就必須解決數(shù)據(jù)傾斜問題。

二、數(shù)據(jù)傾斜的危害

當(dāng)出現(xiàn)數(shù)據(jù)傾斜時，小量任務(wù)耗時遠(yuǎn)高于其它任務(wù)，從而使得整體耗時過大，未能充分發(fā)揮分布式系統(tǒng)的并行計算優(yōu)勢?！　?/p>

另外，當(dāng)發(fā)生數(shù)據(jù)傾斜時，部分任務(wù)處理的數(shù)據(jù)量過大，可能造成內(nèi)存不足使得任務(wù)失敗，并進(jìn)而引進(jìn)整個應(yīng)用失敗?！　?/p>

三、數(shù)據(jù)傾斜的現(xiàn)象

當(dāng)發(fā)現(xiàn)如下現(xiàn)象時，十有八九是發(fā)生數(shù)據(jù)傾斜了:

• 絕大多數(shù) task 執(zhí)行得都非?？欤珎€別 task 執(zhí)行極慢，整體任務(wù)卡在某個階段不能結(jié)束。

• 原本能夠正常執(zhí)行的 Spark 作業(yè)，某天突然報出 OOM（內(nèi)存溢出）異常，觀察異常棧，是我們寫的業(yè)務(wù)代碼造成的。這種情況比較少見。

TIPS

在 Spark streaming 程序中，數(shù)據(jù)傾斜更容易出現(xiàn)，特別是在程序中包含一些類似 sql 的 join、group 這種操作的時候。因為 Spark Streaming 程序在運行的時候，我們一般不會分配特別多的內(nèi)存，因此一旦在這個過程中出現(xiàn)一些數(shù)據(jù)傾斜，就十分容易造成 OOM。

四、數(shù)據(jù)傾斜的原因

在進(jìn)行 shuffle 的時候，必須將各個節(jié)點上相同的 key 拉取到某個節(jié)點上的一個 task 來進(jìn)行處理，比如按照 key 進(jìn)行聚合或 join 等操作。此時如果某個 key 對應(yīng)的數(shù)據(jù)量特別大的話，就會發(fā)生數(shù)據(jù)傾斜。比如大部分 key 對應(yīng)10條數(shù)據(jù)，但是個別 key 卻對應(yīng)了100萬條數(shù)據(jù)，那么大部分 task 可能就只會分配到10條數(shù)據(jù)，然后1秒鐘就運行完了；但是個別 task 可能分配到了100萬數(shù)據(jù)，要運行一兩個小時。

因此出現(xiàn)數(shù)據(jù)傾斜的時候，Spark 作業(yè)看起來會運行得非常緩慢，甚至可能因為某個 task 處理的數(shù)據(jù)量過大導(dǎo)致內(nèi)存溢出。

五、問題發(fā)現(xiàn)與定位

1、通過 Spark Web UI

通過 Spark Web UI 來查看當(dāng)前運行的 stage 各個 task 分配的數(shù)據(jù)量（Shuffle Read Size/Records），從而進(jìn)一步確定是不是 task 分配的數(shù)據(jù)不均勻?qū)е铝藬?shù)據(jù)傾斜。

知道數(shù)據(jù)傾斜發(fā)生在哪一個 stage 之后，接著我們就需要根據(jù) stage 劃分原理，推算出來發(fā)生傾斜的那個 stage 對應(yīng)代碼中的哪一部分，這部分代碼中肯定會有一個 shuffle 類算子?？梢酝ㄟ^ countByKey 查看各個 key 的分布。

TIPS

數(shù)據(jù)傾斜只會發(fā)生在 shuffle 過程中。這里給大家羅列一些常用的并且可能會觸發(fā) shuffle 操作的算子: distinct、groupByKey、reduceByKey、aggregateByKey、join、cogroup、repartition 等。出現(xiàn)數(shù)據(jù)傾斜時，可能就是你的代碼中使用了這些算子中的某一個所導(dǎo)致的。

2、通過 key 統(tǒng)計

也可以通過抽樣統(tǒng)計 key 的出現(xiàn)次數(shù)驗證。

由于數(shù)據(jù)量巨大，可以采用抽樣的方式，對數(shù)據(jù)進(jìn)行抽樣，統(tǒng)計出現(xiàn)的次數(shù)，根據(jù)出現(xiàn)次數(shù)大小排序取出前幾個:

df.select("key").sample(false, 0.1)           // 數(shù)據(jù)采樣
    .(k => (k, 1)).reduceBykey(_ + _)         // 統(tǒng)計 key 出現(xiàn)的次數(shù)
    .map(k => (k._2, k._1)).sortByKey(false)  // 根據(jù) key 出現(xiàn)次數(shù)進(jìn)行排序
    .take(10)                                 // 取前 10 個。

如果發(fā)現(xiàn)多數(shù)數(shù)據(jù)分布都較為平均，而個別數(shù)據(jù)比其他數(shù)據(jù)大上若干個數(shù)量級，則說明發(fā)生了數(shù)據(jù)傾斜。

六、如何緩解數(shù)據(jù)傾斜

基本思路

• 業(yè)務(wù)邏輯: 我們從業(yè)務(wù)邏輯的層面上來優(yōu)化數(shù)據(jù)傾斜，比如要統(tǒng)計不同城市的訂單情況，那么我們單獨對這一線城市來做 count，最后和其它城市做整合。

• 程序?qū)崿F(xiàn): 比如說在 Hive 中，經(jīng)常遇到 count（distinct）操作，這樣會導(dǎo)致最終只有一個 reduce，我們可以先 group 再在外面包一層 count，就可以了；在 Spark 中使用 reduceByKey 替代 groupByKey 等。

• 參數(shù)調(diào)優(yōu): Hadoop 和 Spark 都自帶了很多的參數(shù)和機(jī)制來調(diào)節(jié)數(shù)據(jù)傾斜，合理利用它們就能解決大部分問題。

思路1. 過濾異常數(shù)據(jù)

如果導(dǎo)致數(shù)據(jù)傾斜的 key 是異常數(shù)據(jù)，那么簡單的過濾掉就可以了。

首先要對 key 進(jìn)行分析，判斷是哪些 key 造成數(shù)據(jù)傾斜。具體方法上面已經(jīng)介紹過了，這里不贅述。

然后對這些 key 對應(yīng)的記錄進(jìn)行分析:

1. 空值或者異常值之類的，大多是這個原因引起

2. 無效數(shù)據(jù)，大量重復(fù)的測試數(shù)據(jù)或是對結(jié)果影響不大的有效數(shù)據(jù)

3. 有效數(shù)據(jù)，業(yè)務(wù)導(dǎo)致的正常數(shù)據(jù)分布

解決方案

對于第 1，2 種情況，直接對數(shù)據(jù)進(jìn)行過濾即可。

第3種情況則需要特殊的處理，具體我們下面詳細(xì)介紹。

思路2. 提高 shuffle 并行度

Spark 在做 Shuffle 時，默認(rèn)使用 HashPartitioner（非 Hash Shuffle）對數(shù)據(jù)進(jìn)行分區(qū)。如果并行度設(shè)置的不合適，可能造成大量不相同的 Key 對應(yīng)的數(shù)據(jù)被分配到了同一個 Task 上，造成該 Task 所處理的數(shù)據(jù)遠(yuǎn)大于其它 Task，從而造成數(shù)據(jù)傾斜。

如果調(diào)整 Shuffle 時的并行度，使得原本被分配到同一 Task 的不同 Key 發(fā)配到不同 Task 上處理，則可降低原 Task 所需處理的數(shù)據(jù)量，從而緩解數(shù)據(jù)傾斜問題造成的短板效應(yīng)。

（1）操作流程

RDD 操作 可在需要 Shuffle 的操作算子上直接設(shè)置并行度或者使用 spark.default.parallelism 設(shè)置。如果是 Spark SQL，還可通過 SET spark.sql.shuffle.partitions=[num_tasks]?設(shè)置并行度。默認(rèn)參數(shù)由不同的 Cluster Manager 控制。

dataFrame 和 sparkSql 可以設(shè)置 spark.sql.shuffle.partitions=[num_tasks]?參數(shù)控制 shuffle 的并發(fā)度，默認(rèn)為200。

（2）適用場景

大量不同的 Key 被分配到了相同的 Task 造成該 Task 數(shù)據(jù)量過大。

（3）解決方案

調(diào)整并行度。一般是增大并行度，但有時如減小并行度也可達(dá)到效果。

（4）優(yōu)勢

實現(xiàn)簡單，只需要參數(shù)調(diào)優(yōu)。可用最小的代價解決問題。一般如果出現(xiàn)數(shù)據(jù)傾斜，都可以通過這種方法先試驗幾次，如果問題未解決，再嘗試其它方法。

（5）劣勢

適用場景少，只是讓每個 task 執(zhí)行更少的不同的key。無法解決個別key特別大的情況造成的傾斜，如果某些 key 的大小非常大，即使一個 task 單獨執(zhí)行它，也會受到數(shù)據(jù)傾斜的困擾。并且該方法一般只能緩解數(shù)據(jù)傾斜，沒有徹底消除問題。從實踐經(jīng)驗來看，其效果一般。

TIPS 可以把數(shù)據(jù)傾斜類比為 hash 沖突。提高并行度就類似于 提高 hash 表的大小。

思路3. 自定義 Partitioner

（1）原理

使用自定義的 Partitioner（默認(rèn)為 HashPartitioner），將原本被分配到同一個 Task 的不同 Key 分配到不同 Task。

例如，我們在?groupByKey?算子上，使用自定義的 Partitioner:

.groupByKey(new?Partitioner()?{
  @Override
  public int numPartitions() {
    return 12;
  }

  @Override
  public int getPartition(Object key) {
    int id = Integer.parseInt(key.toString());
    if(id >= 9500000 && id <= 9500084 && ((id - 9500000) % 12) == 0) {
      return (id - 9500000) / 12;
    } else {
      return id % 12;
    }
  }
})

TIPS 這個做法相當(dāng)于自定義 hash 表的 哈希函數(shù)。

（2）適用場景

大量不同的 Key 被分配到了相同的 Task 造成該 Task 數(shù)據(jù)量過大。

（3）解決方案

使用自定義的 Partitioner 實現(xiàn)類代替默認(rèn)的 HashPartitioner，盡量將所有不同的 Key 均勻分配到不同的 Task 中。

（4）優(yōu)勢

不影響原有的并行度設(shè)計。如果改變并行度，后續(xù) Stage 的并行度也會默認(rèn)改變，可能會影響后續(xù) Stage。

（5）劣勢

適用場景有限，只能將不同 Key 分散開，對于同一 Key 對應(yīng)數(shù)據(jù)集非常大的場景不適用。效果與調(diào)整并行度類似，只能緩解數(shù)據(jù)傾斜而不能完全消除數(shù)據(jù)傾斜。而且需要根據(jù)數(shù)據(jù)特點自定義專用的 Partitioner，不夠靈活。

思路4. Reduce 端 Join 轉(zhuǎn)化為 Map 端 Join

通過 Spark 的 Broadcast 機(jī)制，將 Reduce 端 Join 轉(zhuǎn)化為 Map 端 Join，這意味著 Spark 現(xiàn)在不需要跨節(jié)點做 shuffle 而是直接通過本地文件進(jìn)行 join，從而完全消除 Shuffle 帶來的數(shù)據(jù)傾斜。

from?pyspark.sql.functions?import?broadcast
result = broadcast(A).join(B, ["join_col"], "left")

其中 A 是比較小的 dataframe 并且能夠整個存放在 executor 內(nèi)存中。

（1）適用場景

參與Join的一邊數(shù)據(jù)集足夠小，可被加載進(jìn) Driver 并通過 Broadcast 方法廣播到各個 Executor 中。

（2）解決方案

在 Java/Scala 代碼中將小數(shù)據(jù)集數(shù)據(jù)拉取到 Driver，然后通過 Broadcast 方案將小數(shù)據(jù)集的數(shù)據(jù)廣播到各 Executor?；蛘咴谑褂?SQL 前，將 Broadcast 的閾值調(diào)整得足夠大，從而使 Broadcast 生效。進(jìn)而將 Reduce Join 替換為 Map Join。

（3）優(yōu)勢

避免了 Shuffle，徹底消除了數(shù)據(jù)傾斜產(chǎn)生的條件，可極大提升性能。

（4）劣勢

因為是先將小數(shù)據(jù)通過 Broadcase 發(fā)送到每個 executor 上，所以需要參與 Join 的一方數(shù)據(jù)集足夠小，并且主要適用于 Join 的場景，不適合聚合的場景，適用條件有限。

NOTES

使用Spark SQL時需要通過 SET spark.sql.autoBroadcastJoinThreshold=104857600?將 Broadcast 的閾值設(shè)置得足夠大，才會生效。

思路5. 拆分 join 再 union

思路很簡單，就是將一個 join 拆分成 傾斜數(shù)據(jù)集 Join 和 非傾斜數(shù)據(jù)集 Join，最后進(jìn)行 union:

1. 對包含少數(shù)幾個數(shù)據(jù)量過大的 key 的那個 RDD (假設(shè)是 leftRDD)，通過 sample 算子采樣出一份樣本來，然后統(tǒng)計一下每個 key 的數(shù)量，計算出來數(shù)據(jù)量大的是哪幾個 key。具體方法上面已經(jīng)介紹過了，這里不贅述。

2. 然后將這 k 個 key 對應(yīng)的數(shù)據(jù)從 leftRDD 中單獨過濾出來，并給每個 key 都打上 1~n 以內(nèi)的隨機(jī)數(shù)作為前綴，形成一個單獨的 leftSkewRDD；而不會導(dǎo)致傾斜的大部分 key 形成另外一個 leftUnSkewRDD。

3. 接著將需要 join 的另一個 rightRDD，也過濾出來那幾個傾斜 key 并通過 flatMap 操作將該數(shù)據(jù)集中每條數(shù)據(jù)均轉(zhuǎn)換為 n 條數(shù)據(jù)（這 n 條數(shù)據(jù)都按順序附加一個 0~n 的前綴），形成單獨的 rightSkewRDD；不會導(dǎo)致傾斜的大部分 key 也形成另外一個 rightUnSkewRDD。

4. 現(xiàn)在將 leftSkewRDD 與 膨脹 n 倍的 rightSkewRDD 進(jìn)行 join，且在 Join 過程中將隨機(jī)前綴去掉，得到傾斜數(shù)據(jù)集的 Join 結(jié)果 skewedJoinRDD。注意到此時我們已經(jīng)成功將原先相同的 key 打散成 n 份，分散到多個 task 中去進(jìn)行 join 了。

5. 對 leftUnSkewRDD 與 rightUnRDD 進(jìn)行Join，得到 Join 結(jié)果 unskewedJoinRDD。

6. 通過 union 算子將 skewedJoinRDD 與 unskewedJoinRDD 進(jìn)行合并，從而得到完整的 Join 結(jié)果集。

TIPS

1. rightRDD 與傾斜 Key 對應(yīng)的部分?jǐn)?shù)據(jù)，需要與隨機(jī)前綴集 (1~n) 作笛卡爾乘積 (即將數(shù)據(jù)量擴(kuò)大 n 倍），從而保證無論數(shù)據(jù)傾斜側(cè)傾斜 Key 如何加前綴，都能與之正常 Join。
2. skewRDD 的 join 并行度可以設(shè)置為 n * k (k 為 topSkewkey 的個數(shù))。
3. 由于傾斜Key與非傾斜Key的操作完全獨立，可并行進(jìn)行。

（1）適用場景

兩張表都比較大，無法使用 Map 端 Join。其中一個 RDD 有少數(shù)幾個 Key 的數(shù)據(jù)量過大，另外一個 RDD 的 Key 分布較為均勻。

（2）解決方案

將有數(shù)據(jù)傾斜的 RDD 中傾斜 Key 對應(yīng)的數(shù)據(jù)集單獨抽取出來加上隨機(jī)前綴，另外一個 RDD 每條數(shù)據(jù)分別與隨機(jī)前綴結(jié)合形成新的RDD（相當(dāng)于將其數(shù)據(jù)增到到原來的N倍，N即為隨機(jī)前綴的總個數(shù)），然后將二者Join并去掉前綴。然后將不包含傾斜Key的剩余數(shù)據(jù)進(jìn)行Join。最后將兩次Join的結(jié)果集通過union合并，即可得到全部Join結(jié)果。

（3）優(yōu)勢

相對于 Map 則 Join，更能適應(yīng)大數(shù)據(jù)集的 Join。如果資源充足，傾斜部分?jǐn)?shù)據(jù)集與非傾斜部分?jǐn)?shù)據(jù)集可并行進(jìn)行，效率提升明顯。且只針對傾斜部分的數(shù)據(jù)做數(shù)據(jù)擴(kuò)展，增加的資源消耗有限。

（4）劣勢

如果傾斜 Key 非常多，則另一側(cè)數(shù)據(jù)膨脹非常大，此方案不適用。而且此時對傾斜 Key 與非傾斜 Key 分開處理，需要掃描數(shù)據(jù)集兩遍，增加了開銷。

思路6. 大表 key 加鹽，小表擴(kuò)大 N 倍 jion

如果出現(xiàn)數(shù)據(jù)傾斜的 Key 比較多，上一種方法將這些大量的傾斜 Key 分拆出來，意義不大。此時更適合直接對存在數(shù)據(jù)傾斜的數(shù)據(jù)集全部加上隨機(jī)前綴，然后對另外一個不存在嚴(yán)重數(shù)據(jù)傾斜的數(shù)據(jù)集整體與隨機(jī)前綴集作笛卡爾乘積（即將數(shù)據(jù)量擴(kuò)大N倍）。

其實就是上一個方法的特例或者簡化。少了拆分，也就沒有 union。

（1）適用場景

一個數(shù)據(jù)集存在的傾斜 Key 比較多，另外一個數(shù)據(jù)集數(shù)據(jù)分布比較均勻。

（2）優(yōu)勢

對大部分場景都適用，效果不錯。

（3）劣勢

需要將一個數(shù)據(jù)集整體擴(kuò)大 N 倍，會增加資源消耗。

思路7. map 端先局部聚合

在 map 端加個 combiner 函數(shù)進(jìn)行局部聚合。加上 combiner 相當(dāng)于提前進(jìn)行 reduce ,就會把一個 mapper 中的相同 key 進(jìn)行聚合，減少 shuffle 過程中數(shù)據(jù)量 以及 reduce 端的計算量。這種方法可以有效的緩解數(shù)據(jù)傾斜問題，但是如果導(dǎo)致數(shù)據(jù)傾斜的 key 大量分布在不同的 mapper 的時候，這種方法就不是很有效了。

TIPS 使用 reduceByKey 而不是 groupByKey。

思路8. 加鹽局部聚合 + 去鹽全局聚合

這個方案的核心實現(xiàn)思路就是進(jìn)行兩階段聚合。第一次是局部聚合，先給每個 key 都打上一個 1~n 的隨機(jī)數(shù)，比如 3 以內(nèi)的隨機(jī)數(shù)，此時原先一樣的 key 就變成不一樣的了，比如?(hello, 1) (hello, 1) (hello, 1) (hello, 1) (hello, 1)，就會變成?(1_hello, 1) (3_hello, 1) (2_hello, 1) (1_hello, 1) (2_hello, 1)。接著對打上隨機(jī)數(shù)后的數(shù)據(jù)，執(zhí)行 reduceByKey 等聚合操作，進(jìn)行局部聚合，那么局部聚合結(jié)果，就會變成了?(1_hello, 2) (2_hello, 2) (3_hello, 1)。然后將各個 key 的前綴給去掉，就會變成?(hello, 2) (hello, 2) (hello, 1)，再次進(jìn)行全局聚合操作，就可以得到最終結(jié)果了，比如?(hello, 5)。

def antiSkew(): RDD[(String, Int)] = {
    val SPLIT = "-"
    val prefix = new Random().nextInt(10)
    pairs.map(t => ( prefix + SPLIT + t._1, 1))
        .reduceByKey((v1, v2) => v1 + v2)
        .map(t => (t._1.split(SPLIT)(1), t2._2))
        .reduceByKey((v1, v2) => v1 + v2)
}

不過進(jìn)行兩次 mapreduce，性能稍微比一次的差些。

七、Hadoop 中的數(shù)據(jù)傾斜

Hadoop 中直接貼近用戶使用的是 Mapreduce 程序和 Hive 程序，雖說 Hive 最后也是用 MR 來執(zhí)行（至少目前 Hive 內(nèi)存計算并不普及），但是畢竟寫的內(nèi)容邏輯區(qū)別很大，一個是程序，一個是Sql，因此這里稍作區(qū)分。

Hadoop 中的數(shù)據(jù)傾斜主要表現(xiàn)在 ruduce 階段卡在99.99%，一直99.99%不能結(jié)束。

這里如果詳細(xì)的看日志或者和監(jiān)控界面的話會發(fā)現(xiàn):

• 有一個多幾個 reduce 卡住

• 各種 container報錯 OOM

• 讀寫的數(shù)據(jù)量極大，至少遠(yuǎn)遠(yuǎn)超過其它正常的 reduce

• 伴隨著數(shù)據(jù)傾斜，會出現(xiàn)任務(wù)被 kill 等各種詭異的表現(xiàn)。

經(jīng)驗:?Hive的數(shù)據(jù)傾斜，一般都發(fā)生在 Sql 中 Group 和 On 上，而且和數(shù)據(jù)邏輯綁定比較深。

優(yōu)化方法

這里列出來一些方法和思路，具體的參數(shù)和用法在官網(wǎng)看就行了。

1. map join 方式

2. count distinct 的操作，先轉(zhuǎn)成 group，再 count

3. 參數(shù)調(diào)優(yōu)

   set hive.map.aggr=true

   set hive.groupby.skewindata=true

4. left semi jion 的使用

5. 設(shè)置 map 端輸出、中間結(jié)果壓縮。（不完全是解決數(shù)據(jù)傾斜的問題，但是減少了 IO 讀寫和網(wǎng)絡(luò)傳輸，能提高很多效率）

說明

hive.map.aggr=true: 在map中會做部分聚集操作，效率更高但需要更多的內(nèi)存。

hive.groupby.skewindata=true: 數(shù)據(jù)傾斜時負(fù)載均衡，當(dāng)選項設(shè)定為true，生成的查詢計劃會有兩個MRJob。第一個MRJob 中，Map的輸出結(jié)果集合會隨機(jī)分布到Reduce中，每個Reduce做部分聚合操作，并輸出結(jié)果，這樣處理的結(jié)果是相同的GroupBy Key有可能被分發(fā)到不同的Reduce中，從而達(dá)到負(fù)載均衡的目的；第二個MRJob再根據(jù)預(yù)處理的數(shù)據(jù)結(jié)果按照GroupBy Key分布到Reduce中（這個過程可以保證相同的GroupBy Key被分布到同一個Reduce中），最后完成最終的聚合操作。

八、參考文章

1. Spark性能優(yōu)化之道——解決Spark數(shù)據(jù)傾斜（Data Skew）的N種姿勢
2. 漫談千億級數(shù)據(jù)優(yōu)化實踐：數(shù)據(jù)傾斜（純干貨）
3. 解決spark中遇到的數(shù)據(jù)傾斜問題

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)cdcxhl.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價比高”等特點與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。

標(biāo)題名稱：Spark數(shù)據(jù)傾斜及其解決方案-創(chuàng)新互聯(lián)
標(biāo)題網(wǎng)址：http://muchs.cn/article6/dpgoig.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供網(wǎng)站內(nèi)鏈、手機(jī)網(wǎng)站建設(shè)、Google、關(guān)鍵詞優(yōu)化、微信公眾號、用戶體驗

廣告

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時間刪除。文章觀點不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時需注明來源： 創(chuàng)新互聯(lián)