前言

繼基礎(chǔ)篇講解了每個Spark開發(fā)人員都必須熟知的開發(fā)調(diào)優(yōu)與資源調(diào)優(yōu)之后，本文作為《Spark性能優(yōu)化指南》的高級篇船侧，將深入分析數(shù)據(jù)傾斜調(diào)優(yōu)與shuffle調(diào)優(yōu)，以解決更加棘手的性能問題厅各。

數(shù)據(jù)傾斜調(diào)優(yōu)

調(diào)優(yōu)概述

有的時(shí)候镜撩，我們可能會遇到大數(shù)據(jù)計(jì)算中一個最棘手的問題——數(shù)據(jù)傾斜，此時(shí)Spark作業(yè)的性能會比期望差很多讯检。數(shù)據(jù)傾斜調(diào)優(yōu)琐鲁，就是使用各種技術(shù)方案解決不同類型的數(shù)據(jù)傾斜問題，以保證Spark作業(yè)的性能人灼。

數(shù)據(jù)傾斜發(fā)生時(shí)的現(xiàn)象

絕大多數(shù)task執(zhí)行得都非澄Ф危快，但個別task執(zhí)行極慢投放。比如奈泪，總共有1000個task，997個task都在1分鐘之內(nèi)執(zhí)行完了，但是剩余兩三個task卻要一兩個小時(shí)涝桅。這種情況很常見拜姿。

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

數(shù)據(jù)傾斜發(fā)生的原理

數(shù)據(jù)傾斜的原理很簡單：在進(jìn)行shuffle的時(shí)候壁却，必須將各個節(jié)點(diǎn)上相同的key拉取到某個節(jié)點(diǎn)上的一個task來進(jìn)行處理，比如按照key進(jìn)行聚合或join等操作裸准。此時(shí)如果某個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秒鐘就運(yùn)行完了权悟；但是個別task可能分配到了100萬數(shù)據(jù)砸王，要運(yùn)行一兩個小時(shí)。因此峦阁，整個Spark作業(yè)的運(yùn)行進(jìn)度是由運(yùn)行時(shí)間最長的那個task決定的处硬。

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

下圖就是一個很清晰的例子：hello這個key凿跳，在三個節(jié)點(diǎn)上對應(yīng)了總共7條數(shù)據(jù)件豌，這些數(shù)據(jù)都會被拉取到同一個task中進(jìn)行處理；而world和you這兩個key分別才對應(yīng)1條數(shù)據(jù)控嗜，所以另外兩個task只要分別處理1條數(shù)據(jù)即可茧彤。此時(shí)第一個task的運(yùn)行時(shí)間可能是另外兩個task的7倍，而整個stage的運(yùn)行速度也由運(yùn)行最慢的那個task所決定疆栏。

如何定位導(dǎo)致數(shù)據(jù)傾斜的代碼

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

某個task執(zhí)行特別慢的情況

首先要看的自阱，就是數(shù)據(jù)傾斜發(fā)生在第幾個stage中。

如果是用yarn-client模式提交米酬，那么本地是直接可以看到log的沛豌，可以在log中找到當(dāng)前運(yùn)行到了第幾個stage；如果是用yarn-cluster模式提交赃额，則可以通過Spark Web UI來查看當(dāng)前運(yùn)行到了第幾個stage加派。此外，無論是使用yarn-client模式還是yarn-cluster模式爬早，我們都可以在Spark Web UI上深入看一下當(dāng)前這個stage各個task分配的數(shù)據(jù)量哼丈，從而進(jìn)一步確定是不是task分配的數(shù)據(jù)不均勻?qū)е铝藬?shù)據(jù)傾斜。

比如下圖中筛严，倒數(shù)第三列顯示了每個task的運(yùn)行時(shí)間醉旦。明顯可以看到，有的task運(yùn)行特別快桨啃，只需要幾秒鐘就可以運(yùn)行完车胡；而有的task運(yùn)行特別慢，需要幾分鐘才能運(yùn)行完照瘾，此時(shí)單從運(yùn)行時(shí)間上看就已經(jīng)能夠確定發(fā)生數(shù)據(jù)傾斜了匈棘。此外，倒數(shù)第一列顯示了每個task處理的數(shù)據(jù)量析命，明顯可以看到主卫，運(yùn)行時(shí)間特別短的task只需要處理幾百KB的數(shù)據(jù)即可，而運(yùn)行時(shí)間特別長的task需要處理幾千KB的數(shù)據(jù)鹃愤，處理的數(shù)據(jù)量差了10倍簇搅。此時(shí)更加能夠確定是發(fā)生了數(shù)據(jù)傾斜。

知道數(shù)據(jù)傾斜發(fā)生在哪一個stage之后软吐，接著我們就需要根據(jù)stage劃分原理瘩将，推算出來發(fā)生傾斜的那個stage對應(yīng)代碼中的哪一部分，這部分代碼中肯定會有一個shuffle類算子凹耙。精準(zhǔn)推算stage與代碼的對應(yīng)關(guān)系姿现，需要對Spark的源碼有深入的理解，這里我們可以介紹一個相對簡單實(shí)用的推算方法：只要看到Spark代碼中出現(xiàn)了一個shuffle類算子或者是Spark SQL的SQL語句中出現(xiàn)了會導(dǎo)致shuffle的語句（比如group by語句）肖抱，那么就可以判定备典，以那個地方為界限劃分出了前后兩個stage。

這里我們就以Spark最基礎(chǔ)的入門程序——單詞計(jì)數(shù)來舉例意述，如何用最簡單的方法大致推算出一個stage對應(yīng)的代碼熊经。如下示例泽艘，在整個代碼中，只有一個reduceByKey是會發(fā)生shuffle的算子镐依，因此就可以認(rèn)為匹涮，以這個算子為界限，會劃分出前后兩個stage槐壳。

stage0然低，主要是執(zhí)行從textFile到map操作，以及執(zhí)行shuffle write操作务唐。shuffle write操作雳攘，我們可以簡單理解為對pairs RDD中的數(shù)據(jù)進(jìn)行分區(qū)操作，每個task處理的數(shù)據(jù)中枫笛，相同的key會寫入同一個磁盤文件內(nèi)吨灭。

stage1，主要是執(zhí)行從reduceByKey到collect操作刑巧，stage1的各個task一開始運(yùn)行喧兄，就會首先執(zhí)行shuffle read操作。執(zhí)行shuffle read操作的task啊楚，會從stage0的各個task所在節(jié)點(diǎn)拉取屬于自己處理的那些key吠冤，然后對同一個key進(jìn)行全局性的聚合或join等操作，在這里就是對key的value值進(jìn)行累加恭理。stage1在執(zhí)行完reduceByKey算子之后拯辙，就計(jì)算出了最終的wordCounts RDD，然后會執(zhí)行collect算子颜价，將所有數(shù)據(jù)拉取到Driver上涯保，供我們遍歷和打印輸出。

1 val conf = new SparkConf()

2 val sc = new SparkContext(conf)

3

4 val lines = sc.textFile("hdfs://...")

5 val words = lines.flatMap(_.split(" "))

6 val pairs = words.map((_, 1))

7 val wordCounts = pairs.reduceByKey(_ + _)

8

9 wordCounts.collect().foreach(println(_))

通過對單詞計(jì)數(shù)程序的分析周伦，希望能夠讓大家了解最基本的stage劃分的原理遭赂，以及stage劃分后shuffle操作是如何在兩個stage的邊界處執(zhí)行的。然后我們就知道如何快速定位出發(fā)生數(shù)據(jù)傾斜的stage對應(yīng)代碼的哪一個部分了横辆。比如我們在Spark Web UI或者本地log中發(fā)現(xiàn)，stage1的某幾個task執(zhí)行得特別慢茄猫，判定stage1出現(xiàn)了數(shù)據(jù)傾斜狈蚤，那么就可以回到代碼中定位出stage1主要包括了reduceByKey這個shuffle類算子，此時(shí)基本就可以確定是由educeByKey算子導(dǎo)致的數(shù)據(jù)傾斜問題划纽。比如某個單詞出現(xiàn)了100萬次脆侮，其他單詞才出現(xiàn)10次，那么stage1的某個task就要處理100萬數(shù)據(jù)勇劣，整個stage的速度就會被這個task拖慢靖避。

某個task莫名其妙內(nèi)存溢出的情況

這種情況下去定位出問題的代碼就比較容易了潭枣。我們建議直接看yarn-client模式下本地log的異常棧锣咒，或者是通過YARN查看yarn-cluster模式下的log中的異常棧塑顺。一般來說默垄，通過異常棧信息就可以定位到你的代碼中哪一行發(fā)生了內(nèi)存溢出病附。然后在那行代碼附近找找蹋凝，一般也會有shuffle類算子溅呢，此時(shí)很可能就是這個算子導(dǎo)致了數(shù)據(jù)傾斜骡湖。

但是大家要注意的是幢竹，不能單純靠偶然的內(nèi)存溢出就判定發(fā)生了數(shù)據(jù)傾斜榛臼。因?yàn)樽约壕帉懙拇a的bug伊佃，以及偶然出現(xiàn)的數(shù)據(jù)異常，也可能會導(dǎo)致內(nèi)存溢出沛善。因此還是要按照上面所講的方法航揉，通過Spark Web UI查看報(bào)錯的那個stage的各個task的運(yùn)行時(shí)間以及分配的數(shù)據(jù)量，才能確定是否是由于數(shù)據(jù)傾斜才導(dǎo)致了這次內(nèi)存溢出金刁。

查看導(dǎo)致數(shù)據(jù)傾斜的key的數(shù)據(jù)分布情況

知道了數(shù)據(jù)傾斜發(fā)生在哪里之后帅涂，通常需要分析一下那個執(zhí)行了shuffle操作并且導(dǎo)致了數(shù)據(jù)傾斜的RDD/Hive表，查看一下其中key的分布情況胀葱。這主要是為之后選擇哪一種技術(shù)方案提供依據(jù)漠秋。針對不同的key分布與不同的shuffle算子組合起來的各種情況，可能需要選擇不同的技術(shù)方案來解決抵屿。

此時(shí)根據(jù)你執(zhí)行操作的情況不同庆锦，可以有很多種查看key分布的方式：

如果是Spark SQL中的group by、join語句導(dǎo)致的數(shù)據(jù)傾斜轧葛，那么就查詢一下SQL中使用的表的key分布情況搂抒。

如果是對Spark RDD執(zhí)行shuffle算子導(dǎo)致的數(shù)據(jù)傾斜，那么可以在Spark作業(yè)中加入查看key分布的代碼尿扯，比如RDD.countByKey()求晶。然后對統(tǒng)計(jì)出來的各個key出現(xiàn)的次數(shù)，collect/take到客戶端打印一下衷笋，就可以看到key的分布情況芳杏。

舉例來說，對于上面所說的單詞計(jì)數(shù)程序辟宗，如果確定了是stage1的reduceByKey算子導(dǎo)致了數(shù)據(jù)傾斜爵赵，那么就應(yīng)該看看進(jìn)行reduceByKey操作的RDD中的key分布情況，在這個例子中指的就是pairs RDD泊脐。如下示例空幻，我們可以先對pairs采樣10%的樣本數(shù)據(jù)，然后使用countByKey算子統(tǒng)計(jì)出每個key出現(xiàn)的次數(shù)容客，最后在客戶端遍歷和打印樣本數(shù)據(jù)中各個key的出現(xiàn)次數(shù)鬓梅。

1 val sampledPairs = pairs.sample(false, 0.1)

2 val sampledWordCounts = sampledPairs.countByKey()

3 sampledWordCounts.foreach(println(_))

數(shù)據(jù)傾斜的解決方案

解決方案一：使用Hive ETL預(yù)處理數(shù)據(jù)

方案適用場景：導(dǎo)致數(shù)據(jù)傾斜的是Hive表。如果該Hive表中的數(shù)據(jù)本身很不均勻（比如某個key對應(yīng)了100萬數(shù)據(jù)镜遣，其他key才對應(yīng)了10條數(shù)據(jù)）己肮，而且業(yè)務(wù)場景需要頻繁使用Spark對Hive表執(zhí)行某個分析操作，那么比較適合使用這種技術(shù)方案悲关。

方案實(shí)現(xiàn)思路：此時(shí)可以評估一下寓辱，是否可以通過Hive來進(jìn)行數(shù)據(jù)預(yù)處理（即通過Hive ETL預(yù)先對數(shù)據(jù)按照key進(jìn)行聚合艘绍，或者是預(yù)先和其他表進(jìn)行join），然后在Spark作業(yè)中針對的數(shù)據(jù)源就不是原來的Hive表了这敬，而是預(yù)處理后的Hive表航夺。此時(shí)由于數(shù)據(jù)已經(jīng)預(yù)先進(jìn)行過聚合或join操作了，那么在Spark作業(yè)中也就不需要使用原先的shuffle類算子執(zhí)行這類操作了崔涂。

方案實(shí)現(xiàn)原理：這種方案從根源上解決了數(shù)據(jù)傾斜阳掐，因?yàn)閺氐妆苊饬嗽赟park中執(zhí)行shuffle類算子，那么肯定就不會有數(shù)據(jù)傾斜的問題了冷蚂。但是這里也要提醒一下大家缭保，這種方式屬于治標(biāo)不治本。因?yàn)楫吘箶?shù)據(jù)本身就存在分布不均勻的問題蝙茶，所以Hive ETL中進(jìn)行g(shù)roup by或者join等shuffle操作時(shí)艺骂，還是會出現(xiàn)數(shù)據(jù)傾斜，導(dǎo)致Hive ETL的速度很慢隆夯。我們只是把數(shù)據(jù)傾斜的發(fā)生提前到了Hive ETL中钳恕，避免Spark程序發(fā)生數(shù)據(jù)傾斜而已。

方案優(yōu)點(diǎn)：實(shí)現(xiàn)起來簡單便捷蹄衷，效果還非常好忧额，完全規(guī)避掉了數(shù)據(jù)傾斜，Spark作業(yè)的性能會大幅度提升宦芦。

方案缺點(diǎn)：治標(biāo)不治本，Hive ETL中還是會發(fā)生數(shù)據(jù)傾斜轴脐。

方案實(shí)踐經(jīng)驗(yàn)：在一些Java系統(tǒng)與Spark結(jié)合使用的項(xiàng)目中调卑，會出現(xiàn)Java代碼頻繁調(diào)用Spark作業(yè)的場景抡砂，而且對Spark作業(yè)的執(zhí)行性能要求很高，就比較適合使用這種方案恬涧。將數(shù)據(jù)傾斜提前到上游的Hive ETL注益，每天僅執(zhí)行一次，只有那一次是比較慢的溯捆，而之后每次Java調(diào)用Spark作業(yè)時(shí)丑搔，執(zhí)行速度都會很快，能夠提供更好的用戶體驗(yàn)提揍。

項(xiàng)目實(shí)踐經(jīng)驗(yàn)：在美團(tuán)·點(diǎn)評的交互式用戶行為分析系統(tǒng)中使用了這種方案啤月，該系統(tǒng)主要是允許用戶通過Java Web系統(tǒng)提交數(shù)據(jù)分析統(tǒng)計(jì)任務(wù)，后端通過Java提交Spark作業(yè)進(jìn)行數(shù)據(jù)分析統(tǒng)計(jì)劳跃。要求Spark作業(yè)速度必須要快谎仲，盡量在10分鐘以內(nèi)，否則速度太慢刨仑，用戶體驗(yàn)會很差郑诺。所以我們將有些Spark作業(yè)的shuffle操作提前到了Hive ETL中，從而讓Spark直接使用預(yù)處理的Hive中間表杉武，盡可能地減少Spark的shuffle操作辙诞，大幅度提升了性能，將部分作業(yè)的性能提升了6倍以上轻抱。

解決方案二：過濾少數(shù)導(dǎo)致傾斜的key

方案適用場景：如果發(fā)現(xiàn)導(dǎo)致傾斜的key就少數(shù)幾個飞涂，而且對計(jì)算本身的影響并不大的話，那么很適合使用這種方案十拣。比如99%的key就對應(yīng)10條數(shù)據(jù)封拧，但是只有一個key對應(yīng)了100萬數(shù)據(jù)，從而導(dǎo)致了數(shù)據(jù)傾斜夭问。

方案實(shí)現(xiàn)思路：如果我們判斷那少數(shù)幾個數(shù)據(jù)量特別多的key泽西，對作業(yè)的執(zhí)行和計(jì)算結(jié)果不是特別重要的話，那么干脆就直接過濾掉那少數(shù)幾個key缰趋。比如捧杉，在Spark SQL中可以使用where子句過濾掉這些key或者在Spark Core中對RDD執(zhí)行filter算子過濾掉這些key。如果需要每次作業(yè)執(zhí)行時(shí)秘血，動態(tài)判定哪些key的數(shù)據(jù)量最多然后再進(jìn)行過濾味抖，那么可以使用sample算子對RDD進(jìn)行采樣，然后計(jì)算出每個key的數(shù)量灰粮，取數(shù)據(jù)量最多的key過濾掉即可仔涩。

方案實(shí)現(xiàn)原理：將導(dǎo)致數(shù)據(jù)傾斜的key給過濾掉之后，這些key就不會參與計(jì)算了粘舟，自然不可能產(chǎn)生數(shù)據(jù)傾斜熔脂。

方案優(yōu)點(diǎn)：實(shí)現(xiàn)簡單佩研，而且效果也很好，可以完全規(guī)避掉數(shù)據(jù)傾斜霞揉。

方案缺點(diǎn)：適用場景不多旬薯，大多數(shù)情況下，導(dǎo)致傾斜的key還是很多的适秩，并不是只有少數(shù)幾個绊序。

方案實(shí)踐經(jīng)驗(yàn)：在項(xiàng)目中我們也采用過這種方案解決數(shù)據(jù)傾斜。有一次發(fā)現(xiàn)某一天Spark作業(yè)在運(yùn)行的時(shí)候突然OOM了秽荞，追查之后發(fā)現(xiàn)骤公，是Hive表中的某一個key在那天數(shù)據(jù)異常，導(dǎo)致數(shù)據(jù)量暴增蚂会。因此就采取每次執(zhí)行前先進(jìn)行采樣淋样，計(jì)算出樣本中數(shù)據(jù)量最大的幾個key之后，直接在程序中將那些key給過濾掉胁住。

解決方案三：提高shuffle操作的并行度

方案適用場景：如果我們必須要對數(shù)據(jù)傾斜迎難而上趁猴，那么建議優(yōu)先使用這種方案，因?yàn)檫@是處理數(shù)據(jù)傾斜最簡單的一種方案彪见。

方案實(shí)現(xiàn)思路：在對RDD執(zhí)行shuffle算子時(shí)儡司，給shuffle算子傳入一個參數(shù)，比如reduceByKey(1000)余指，該參數(shù)就設(shè)置了這個shuffle算子執(zhí)行時(shí)shuffle read task的數(shù)量捕犬。對于Spark SQL中的shuffle類語句，比如group by酵镜、join等碉碉，需要設(shè)置一個參數(shù)，即spark.sql.shuffle.partitions淮韭，該參數(shù)代表了shuffle read task的并行度垢粮，該值默認(rèn)是200，對于很多場景來說都有點(diǎn)過小靠粪。

方案實(shí)現(xiàn)原理：增加shuffle read task的數(shù)量蜡吧，可以讓原本分配給一個task的多個key分配給多個task，從而讓每個task處理比原來更少的數(shù)據(jù)占键。舉例來說昔善，如果原本有5個key，每個key對應(yīng)10條數(shù)據(jù)畔乙，這5個key都是分配給一個task的君仆，那么這個task就要處理50條數(shù)據(jù)。而增加了shuffle read task以后，每個task就分配到一個key返咱，即每個task就處理10條數(shù)據(jù)氮帐，那么自然每個task的執(zhí)行時(shí)間都會變短了。具體原理如下圖所示洛姑。

方案優(yōu)點(diǎn)：實(shí)現(xiàn)起來比較簡單，可以有效緩解和減輕數(shù)據(jù)傾斜的影響皮服。

方案缺點(diǎn)：只是緩解了數(shù)據(jù)傾斜而已楞艾，沒有徹底根除問題，根據(jù)實(shí)踐經(jīng)驗(yàn)來看龄广，其效果有限硫眯。

方案實(shí)踐經(jīng)驗(yàn)：該方案通常無法徹底解決數(shù)據(jù)傾斜，因?yàn)槿绻霈F(xiàn)一些極端情況择同，比如某個key對應(yīng)的數(shù)據(jù)量有100萬两入，那么無論你的task數(shù)量增加到多少，這個對應(yīng)著100萬數(shù)據(jù)的key肯定還是會分配到一個task中去處理敲才，因此注定還是會發(fā)生數(shù)據(jù)傾斜的裹纳。所以這種方案只能說是在發(fā)現(xiàn)數(shù)據(jù)傾斜時(shí)嘗試使用的第一種手段，嘗試去用嘴簡單的方法緩解數(shù)據(jù)傾斜而已紧武，或者是和其他方案結(jié)合起來使用剃氧。

解決方案四：兩階段聚合（局部聚合+全局聚合）

方案適用場景：對RDD執(zhí)行reduceByKey等聚合類shuffle算子或者在Spark SQL中使用group by語句進(jìn)行分組聚合時(shí)，比較適用這種方案阻星。

方案實(shí)現(xiàn)思路：這個方案的核心實(shí)現(xiàn)思路就是進(jìn)行兩階段聚合朋鞍。第一次是局部聚合，先給每個key都打上一個隨機(jī)數(shù)妥箕，比如10以內(nèi)的隨機(jī)數(shù)滥酥，此時(shí)原先一樣的key就變成不一樣的了，比如(hello, 1) (hello, 1) (hello, 1) (hello, 1)畦幢，就會變成(1_hello, 1) (1_hello, 1) (2_hello, 1) (2_hello, 1)坎吻。接著對打上隨機(jī)數(shù)后的數(shù)據(jù)，執(zhí)行reduceByKey等聚合操作呛讲，進(jìn)行局部聚合禾怠，那么局部聚合結(jié)果，就會變成了(1_hello, 2) (2_hello, 2)贝搁。然后將各個key的前綴給去掉吗氏，就會變成(hello,2)(hello,2)，再次進(jìn)行全局聚合操作雷逆，就可以得到最終結(jié)果了弦讽，比如(hello, 4)。

方案實(shí)現(xiàn)原理：將原本相同的key通過附加隨機(jī)前綴的方式，變成多個不同的key往产，就可以讓原本被一個task處理的數(shù)據(jù)分散到多個task上去做局部聚合被碗，進(jìn)而解決單個task處理數(shù)據(jù)量過多的問題。接著去除掉隨機(jī)前綴仿村，再次進(jìn)行全局聚合锐朴，就可以得到最終的結(jié)果。具體原理見下圖蔼囊。

方案優(yōu)點(diǎn)：對于聚合類的shuffle操作導(dǎo)致的數(shù)據(jù)傾斜焚志，效果是非常不錯的。通常都可以解決掉數(shù)據(jù)傾斜畏鼓，或者至少是大幅度緩解數(shù)據(jù)傾斜酱酬，將Spark作業(yè)的性能提升數(shù)倍以上。

方案缺點(diǎn)：僅僅適用于聚合類的shuffle操作云矫，適用范圍相對較窄膳沽。如果是join類的shuffle操作，還得用其他的解決方案让禀。

1 // 第一步挑社，給RDD中的每個key都打上一個隨機(jī)前綴。

2 JavaPairRDD randomPrefixRdd = rdd.mapToPair(

3? ? ? ? new PairFunction, String, Long>() {

4? ? ? ? ? ? private static final long serialVersionUID = 1L;

5? ? ? ? ? ? @Override

6? ? ? ? ? ? public Tuple2 call(Tuple2 tuple)

7? ? ? ? ? ? ? ? ? ? throws Exception {

8? ? ? ? ? ? ? ? Random random = new Random();

9? ? ? ? ? ? ? ? int prefix = random.nextInt(10);

10? ? ? ? ? ? ? ? return new Tuple2(prefix + "_" + tuple._1, tuple._2);

11? ? ? ? ? ? }

12? ? ? ? });

13

14 // 第二步巡揍，對打上隨機(jī)前綴的key進(jìn)行局部聚合滔灶。

15 JavaPairRDD localAggrRdd = randomPrefixRdd.reduceByKey(

16? ? ? ? new Function2() {

17? ? ? ? ? ? private static final long serialVersionUID = 1L;

18? ? ? ? ? ? @Override

19? ? ? ? ? ? public Long call(Long v1, Long v2) throws Exception {

20? ? ? ? ? ? ? ? return v1 + v2;

21? ? ? ? ? ? }

22? ? ? ? });

23

24 // 第三步，去除RDD中每個key的隨機(jī)前綴吼肥。

25 JavaPairRDD removedRandomPrefixRdd = localAggrRdd.mapToPair(

26? ? ? ? new PairFunction, Long, Long>() {

27? ? ? ? ? ? private static final long serialVersionUID = 1L;

28? ? ? ? ? ? @Override

29? ? ? ? ? ? public Tuple2 call(Tuple2 tuple)

30? ? ? ? ? ? ? ? ? ? throws Exception {

31? ? ? ? ? ? ? ? long originalKey = Long.valueOf(tuple._1.split("_")[1]);

32? ? ? ? ? ? ? ? return new Tuple2(originalKey, tuple._2);

33? ? ? ? ? ? }

34? ? ? ? });

35

36 // 第四步录平，對去除了隨機(jī)前綴的RDD進(jìn)行全局聚合。

37 JavaPairRDD globalAggrRdd = removedRandomPrefixRdd.reduceByKey(

38? ? ? ? new Function2() {

39? ? ? ? ? ? private static final long serialVersionUID = 1L;

40? ? ? ? ? ? @Override

41? ? ? ? ? ? public Long call(Long v1, Long v2) throws Exception {

42? ? ? ? ? ? ? ? return v1 + v2;

43? ? ? ? ? ? }

44? ? ? ? });

解決方案五：將reduce join轉(zhuǎn)為map join

方案適用場景：在對RDD使用join類操作缀皱，或者是在Spark SQL中使用join語句時(shí)斗这，而且join操作中的一個RDD或表的數(shù)據(jù)量比較小（比如幾百M(fèi)或者一兩G）啤斗，比較適用此方案表箭。

方案實(shí)現(xiàn)思路：不使用join算子進(jìn)行連接操作，而使用Broadcast變量與map類算子實(shí)現(xiàn)join操作钮莲，進(jìn)而完全規(guī)避掉shuffle類的操作免钻，徹底避免數(shù)據(jù)傾斜的發(fā)生和出現(xiàn)。將較小RDD中的數(shù)據(jù)直接通過collect算子拉取到Driver端的內(nèi)存中來崔拥，然后對其創(chuàng)建一個Broadcast變量极舔；接著對另外一個RDD執(zhí)行map類算子，在算子函數(shù)內(nèi)链瓦，從Broadcast變量中獲取較小RDD的全量數(shù)據(jù)拆魏，與當(dāng)前RDD的每一條數(shù)據(jù)按照連接key進(jìn)行比對盯桦，如果連接key相同的話，那么就將兩個RDD的數(shù)據(jù)用你需要的方式連接起來渤刃。

方案實(shí)現(xiàn)原理：普通的join是會走shuffle過程的拥峦，而一旦shuffle，就相當(dāng)于會將相同key的數(shù)據(jù)拉取到一個shuffle read task中再進(jìn)行join卖子，此時(shí)就是reduce join略号。但是如果一個RDD是比較小的，則可以采用廣播小RDD全量數(shù)據(jù)+map算子來實(shí)現(xiàn)與join同樣的效果洋闽，也就是map join璃哟，此時(shí)就不會發(fā)生shuffle操作，也就不會發(fā)生數(shù)據(jù)傾斜喊递。具體原理如下圖所示。

方案優(yōu)點(diǎn)：對join操作導(dǎo)致的數(shù)據(jù)傾斜阳似，效果非常好骚勘，因?yàn)楦揪筒粫l(fā)生shuffle，也就根本不會發(fā)生數(shù)據(jù)傾斜撮奏。

方案缺點(diǎn)：適用場景較少俏讹，因?yàn)檫@個方案只適用于一個大表和一個小表的情況。畢竟我們需要將小表進(jìn)行廣播畜吊，此時(shí)會比較消耗內(nèi)存資源泽疆，driver和每個Executor內(nèi)存中都會駐留一份小RDD的全量數(shù)據(jù)。如果我們廣播出去的RDD數(shù)據(jù)比較大玲献，比如10G以上殉疼，那么就可能發(fā)生內(nèi)存溢出了。因此并不適合兩個都是大表的情況捌年。

1 // 首先將數(shù)據(jù)量比較小的RDD的數(shù)據(jù)瓢娜，collect到Driver中來。

2 List> rdd1Data = rdd1.collect()

3 // 然后使用Spark的廣播功能礼预，將小RDD的數(shù)據(jù)轉(zhuǎn)換成廣播變量眠砾，這樣每個Executor就只有一份RDD的數(shù)據(jù)。

4 // 可以盡可能節(jié)省內(nèi)存空間托酸，并且減少網(wǎng)絡(luò)傳輸性能開銷褒颈。

5 final Broadcast>> rdd1DataBroadcast = sc.broadcast(rdd1Data);

6

7 // 對另外一個RDD執(zhí)行map類操作，而不再是join類操作励堡。

8 JavaPairRDD> joinedRdd = rdd2.mapToPair(

9? ? ? ? new PairFunction, String, Tuple2>() {

10? ? ? ? ? ? private static final long serialVersionUID = 1L;

11? ? ? ? ? ? @Override

12? ? ? ? ? ? public Tuple2> call(Tuple2 tuple)

13? ? ? ? ? ? ? ? ? ? throws Exception {

14? ? ? ? ? ? ? ? // 在算子函數(shù)中谷丸，通過廣播變量，獲取到本地Executor中的rdd1數(shù)據(jù)应结。

15? ? ? ? ? ? ? ? List> rdd1Data = rdd1DataBroadcast.value();

16? ? ? ? ? ? ? ? // 可以將rdd1的數(shù)據(jù)轉(zhuǎn)換為一個Map淤井，便于后面進(jìn)行join操作。

17? ? ? ? ? ? ? ? Map rdd1DataMap = new HashMap();

18? ? ? ? ? ? ? ? for(Tuple2 data : rdd1Data) {

19? ? ? ? ? ? ? ? ? ? rdd1DataMap.put(data._1, data._2);

20? ? ? ? ? ? ? ? }

21? ? ? ? ? ? ? ? // 獲取當(dāng)前RDD數(shù)據(jù)的key以及value。

22? ? ? ? ? ? ? ? String key = tuple._1;

23? ? ? ? ? ? ? ? String value = tuple._2;

24? ? ? ? ? ? ? ? // 從rdd1數(shù)據(jù)Map中币狠，根據(jù)key獲取到可以join到的數(shù)據(jù)游两。

25? ? ? ? ? ? ? ? Row rdd1Value = rdd1DataMap.get(key);

26? ? ? ? ? ? ? ? return new Tuple2(key, new Tuple2(value, rdd1Value));

27? ? ? ? ? ? }

28? ? ? ? });

29

30 // 這里得提示一下。

31 // 上面的做法漩绵，僅僅適用于rdd1中的key沒有重復(fù)贱案，全部是唯一的場景。

32 // 如果rdd1中有多個相同的key止吐，那么就得用flatMap類的操作宝踪，在進(jìn)行join的時(shí)候不能用map，而是得遍歷rdd1所有數(shù)據(jù)進(jìn)行join碍扔。

33 // rdd2中每條數(shù)據(jù)都可能會返回多條join后的數(shù)據(jù)瘩燥。

解決方案六：采樣傾斜key并分拆join操作

方案適用場景：兩個RDD/Hive表進(jìn)行join的時(shí)候，如果數(shù)據(jù)量都比較大不同，無法采用“解決方案五”厉膀，那么此時(shí)可以看一下兩個RDD/Hive表中的key分布情況。如果出現(xiàn)數(shù)據(jù)傾斜二拐，是因?yàn)槠渲心骋粋€RDD/Hive表中的少數(shù)幾個key的數(shù)據(jù)量過大服鹅，而另一個RDD/Hive表中的所有key都分布比較均勻，那么采用這個解決方案是比較合適的百新。

方案實(shí)現(xiàn)思路：

對包含少數(shù)幾個數(shù)據(jù)量過大的key的那個RDD企软，通過sample算子采樣出一份樣本來，然后統(tǒng)計(jì)一下每個key的數(shù)量饭望，計(jì)算出來數(shù)據(jù)量最大的是哪幾個key仗哨。

然后將這幾個key對應(yīng)的數(shù)據(jù)從原來的RDD中拆分出來，形成一個單獨(dú)的RDD铅辞，并給每個key都打上n以內(nèi)的隨機(jī)數(shù)作為前綴藻治，而不會導(dǎo)致傾斜的大部分key形成另外一個RDD。

接著將需要join的另一個RDD巷挥，也過濾出來那幾個傾斜key對應(yīng)的數(shù)據(jù)并形成一個單獨(dú)的RDD桩卵，將每條數(shù)據(jù)膨脹成n條數(shù)據(jù)，這n條數(shù)據(jù)都按順序附加一個0~n的前綴倍宾，不會導(dǎo)致傾斜的大部分key也形成另外一個RDD雏节。

再將附加了隨機(jī)前綴的獨(dú)立RDD與另一個膨脹n倍的獨(dú)立RDD進(jìn)行join，此時(shí)就可以將原先相同的key打散成n份高职，分散到多個task中去進(jìn)行join了钩乍。

而另外兩個普通的RDD就照常join即可。

最后將兩次join的結(jié)果使用union算子合并起來即可怔锌，就是最終的join結(jié)果寥粹。

方案實(shí)現(xiàn)原理：對于join導(dǎo)致的數(shù)據(jù)傾斜变过，如果只是某幾個key導(dǎo)致了傾斜，可以將少數(shù)幾個key分拆成獨(dú)立RDD涝涤，并附加隨機(jī)前綴打散成n份去進(jìn)行join媚狰，此時(shí)這幾個key對應(yīng)的數(shù)據(jù)就不會集中在少數(shù)幾個task上，而是分散到多個task進(jìn)行join了阔拳。具體原理見下圖崭孤。

方案優(yōu)點(diǎn)：對于join導(dǎo)致的數(shù)據(jù)傾斜，如果只是某幾個key導(dǎo)致了傾斜糊肠，采用該方式可以用最有效的方式打散key進(jìn)行join辨宠。而且只需要針對少數(shù)傾斜key對應(yīng)的數(shù)據(jù)進(jìn)行擴(kuò)容n倍，不需要對全量數(shù)據(jù)進(jìn)行擴(kuò)容货裹。避免了占用過多內(nèi)存嗤形。

方案缺點(diǎn)：如果導(dǎo)致傾斜的key特別多的話，比如成千上萬個key都導(dǎo)致數(shù)據(jù)傾斜弧圆，那么這種方式也不適合赋兵。

? 1 // 首先從包含了少數(shù)幾個導(dǎo)致數(shù)據(jù)傾斜key的rdd1中，采樣10%的樣本數(shù)據(jù)墓阀。

? 2 JavaPairRDD sampledRDD = rdd1.sample(false, 0.1);

? 3

? 4 // 對樣本數(shù)據(jù)RDD統(tǒng)計(jì)出每個key的出現(xiàn)次數(shù)，并按出現(xiàn)次數(shù)降序排序拓轻。

? 5 // 對降序排序后的數(shù)據(jù)斯撮，取出top 1或者top 100的數(shù)據(jù)，也就是key最多的前n個數(shù)據(jù)扶叉。

? 6 // 具體取出多少個數(shù)據(jù)量最多的key勿锅，由大家自己決定，我們這里就取1個作為示范枣氧。

? 7 JavaPairRDD mappedSampledRDD = sampledRDD.mapToPair(

? 8? ? ? ? new PairFunction, Long, Long>() {

? 9? ? ? ? ? ? private static final long serialVersionUID = 1L;

10? ? ? ? ? ? @Override

11? ? ? ? ? ? public Tuple2 call(Tuple2 tuple)

12? ? ? ? ? ? ? ? ? ? throws Exception {

13? ? ? ? ? ? ? ? return new Tuple2(tuple._1, 1L);

14? ? ? ? ? ? }? ?

15? ? ? ? });

16 JavaPairRDD countedSampledRDD = mappedSampledRDD.reduceByKey(

17? ? ? ? new Function2() {

18? ? ? ? ? ? private static final long serialVersionUID = 1L;

19? ? ? ? ? ? @Override

20? ? ? ? ? ? public Long call(Long v1, Long v2) throws Exception {

21? ? ? ? ? ? ? ? return v1 + v2;

22? ? ? ? ? ? }

23? ? ? ? });

24 JavaPairRDD reversedSampledRDD = countedSampledRDD.mapToPair(

25? ? ? ? new PairFunction, Long, Long>() {

26? ? ? ? ? ? private static final long serialVersionUID = 1L;

27? ? ? ? ? ? @Override

28? ? ? ? ? ? public Tuple2 call(Tuple2 tuple)

29? ? ? ? ? ? ? ? ? ? throws Exception {

30? ? ? ? ? ? ? ? return new Tuple2(tuple._2, tuple._1);

31? ? ? ? ? ? }

32? ? ? ? });

33 final Long skewedUserid = reversedSampledRDD.sortByKey(false).take(1).get(0)._2;

34

35 // 從rdd1中分拆出導(dǎo)致數(shù)據(jù)傾斜的key溢十，形成獨(dú)立的RDD。

36 JavaPairRDD skewedRDD = rdd1.filter(

37? ? ? ? new Function, Boolean>() {

38? ? ? ? ? ? private static final long serialVersionUID = 1L;

39? ? ? ? ? ? @Override

40? ? ? ? ? ? public Boolean call(Tuple2 tuple) throws Exception {

41? ? ? ? ? ? ? ? return tuple._1.equals(skewedUserid);

42? ? ? ? ? ? }

43? ? ? ? });

44 // 從rdd1中分拆出不導(dǎo)致數(shù)據(jù)傾斜的普通key达吞，形成獨(dú)立的RDD张弛。

45 JavaPairRDD commonRDD = rdd1.filter(

46? ? ? ? new Function, Boolean>() {

47? ? ? ? ? ? private static final long serialVersionUID = 1L;

48? ? ? ? ? ? @Override

49? ? ? ? ? ? public Boolean call(Tuple2 tuple) throws Exception {

50? ? ? ? ? ? ? ? return !tuple._1.equals(skewedUserid);

51? ? ? ? ? ? }

52? ? ? ? });

53

54 // rdd2，就是那個所有key的分布相對較為均勻的rdd酪劫。

55 // 這里將rdd2中吞鸭，前面獲取到的key對應(yīng)的數(shù)據(jù)，過濾出來覆糟，分拆成單獨(dú)的rdd刻剥，并對rdd中的數(shù)據(jù)使用flatMap算子都擴(kuò)容100倍。

56 // 對擴(kuò)容的每條數(shù)據(jù)滩字，都打上0～100的前綴造虏。

57 JavaPairRDD skewedRdd2 = rdd2.filter(

58? ? ? ? ? new Function, Boolean>() {

59? ? ? ? ? ? private static final long serialVersionUID = 1L;

60? ? ? ? ? ? @Override

61? ? ? ? ? ? public Boolean call(Tuple2 tuple) throws Exception {

62? ? ? ? ? ? ? ? return tuple._1.equals(skewedUserid);

63? ? ? ? ? ? }

64? ? ? ? }).flatMapToPair(new PairFlatMapFunction, String, Row>() {

65? ? ? ? ? ? private static final long serialVersionUID = 1L;

66? ? ? ? ? ? @Override

67? ? ? ? ? ? public Iterable> call(

68? ? ? ? ? ? ? ? ? ? Tuple2 tuple) throws Exception {

69? ? ? ? ? ? ? ? Random random = new Random();

70? ? ? ? ? ? ? ? List> list = new ArrayList>();

71? ? ? ? ? ? ? ? for(int i = 0; i < 100; i++) {

72? ? ? ? ? ? ? ? ? ? list.add(new Tuple2(i + "_" + tuple._1, tuple._2));

73? ? ? ? ? ? ? ? }

74? ? ? ? ? ? ? ? return list;

75? ? ? ? ? ? }

76

77? ? ? ? });

78

79 // 將rdd1中分拆出來的導(dǎo)致傾斜的key的獨(dú)立rdd御吞，每條數(shù)據(jù)都打上100以內(nèi)的隨機(jī)前綴。

80 // 然后將這個rdd1中分拆出來的獨(dú)立rdd漓藕，與上面rdd2中分拆出來的獨(dú)立rdd陶珠，進(jìn)行join。

81 JavaPairRDD> joinedRDD1 = skewedRDD.mapToPair(

82? ? ? ? new PairFunction, String, String>() {

83? ? ? ? ? ? private static final long serialVersionUID = 1L;

84? ? ? ? ? ? @Override

85? ? ? ? ? ? public Tuple2 call(Tuple2 tuple)

86? ? ? ? ? ? ? ? ? ? throws Exception {

87? ? ? ? ? ? ? ? Random random = new Random();

88? ? ? ? ? ? ? ? int prefix = random.nextInt(100);

89? ? ? ? ? ? ? ? return new Tuple2(prefix + "_" + tuple._1, tuple._2);

90? ? ? ? ? ? }

91? ? ? ? })

92? ? ? ? .join(skewedUserid2infoRDD)

93? ? ? ? .mapToPair(new PairFunction>, Long, Tuple2>() {

94? ? ? ? ? ? ? ? ? ? ? ? private static final long serialVersionUID = 1L;

95? ? ? ? ? ? ? ? ? ? ? ? @Override

96? ? ? ? ? ? ? ? ? ? ? ? public Tuple2> call(

97? ? ? ? ? ? ? ? ? ? ? ? ? ? Tuple2> tuple)

98? ? ? ? ? ? ? ? ? ? ? ? ? ? throws Exception {

99? ? ? ? ? ? ? ? ? ? ? ? ? ? long key = Long.valueOf(tuple._1.split("_")[1]);

100? ? ? ? ? ? ? ? ? ? ? ? ? ? return new Tuple2>(key, tuple._2);

101? ? ? ? ? ? ? ? ? ? ? ? }

102? ? ? ? ? ? ? ? ? ? });

103

104 // 將rdd1中分拆出來的包含普通key的獨(dú)立rdd撵术，直接與rdd2進(jìn)行join背率。

105 JavaPairRDD> joinedRDD2 = commonRDD.join(rdd2);

106

107 // 將傾斜key join后的結(jié)果與普通key join后的結(jié)果，uinon起來嫩与。

108 // 就是最終的join結(jié)果寝姿。

109 JavaPairRDD> joinedRDD = joinedRDD1.union(joinedRDD2);

解決方案七：使用隨機(jī)前綴和擴(kuò)容RDD進(jìn)行join

方案適用場景：如果在進(jìn)行join操作時(shí)，RDD中有大量的key導(dǎo)致數(shù)據(jù)傾斜划滋，那么進(jìn)行分拆key也沒什么意義饵筑，此時(shí)就只能使用最后一種方案來解決問題了。

方案實(shí)現(xiàn)思路：

該方案的實(shí)現(xiàn)思路基本和“解決方案六”類似处坪，首先查看RDD/Hive表中的數(shù)據(jù)分布情況根资，找到那個造成數(shù)據(jù)傾斜的RDD/Hive表，比如有多個key都對應(yīng)了超過1萬條數(shù)據(jù)同窘。

然后將該RDD的每條數(shù)據(jù)都打上一個n以內(nèi)的隨機(jī)前綴玄帕。

同時(shí)對另外一個正常的RDD進(jìn)行擴(kuò)容，將每條數(shù)據(jù)都擴(kuò)容成n條數(shù)據(jù)想邦，擴(kuò)容出來的每條數(shù)據(jù)都依次打上一個0~n的前綴裤纹。

最后將兩個處理后的RDD進(jìn)行join即可。

方案實(shí)現(xiàn)原理：將原先一樣的key通過附加隨機(jī)前綴變成不一樣的key丧没，然后就可以將這些處理后的“不同key”分散到多個task中去處理鹰椒，而不是讓一個task處理大量的相同key。該方案與“解決方案六”的不同之處就在于呕童，上一種方案是盡量只對少數(shù)傾斜key對應(yīng)的數(shù)據(jù)進(jìn)行特殊處理漆际，由于處理過程需要擴(kuò)容RDD殖侵，因此上一種方案擴(kuò)容RDD后對內(nèi)存的占用并不大留瞳；而這一種方案是針對有大量傾斜key的情況，沒法將部分key拆分出來進(jìn)行單獨(dú)處理秀姐，因此只能對整個RDD進(jìn)行數(shù)據(jù)擴(kuò)容往声，對內(nèi)存資源要求很高茫蛹。

方案優(yōu)點(diǎn)：對join類型的數(shù)據(jù)傾斜基本都可以處理，而且效果也相對比較顯著烁挟，性能提升效果非常不錯婴洼。

方案缺點(diǎn)：該方案更多的是緩解數(shù)據(jù)傾斜，而不是徹底避免數(shù)據(jù)傾斜撼嗓。而且需要對整個RDD進(jìn)行擴(kuò)容柬采，對內(nèi)存資源要求很高欢唾。

方案實(shí)踐經(jīng)驗(yàn)：曾經(jīng)開發(fā)一個數(shù)據(jù)需求的時(shí)候，發(fā)現(xiàn)一個join導(dǎo)致了數(shù)據(jù)傾斜粉捻。優(yōu)化之前礁遣，作業(yè)的執(zhí)行時(shí)間大約是60分鐘左右；使用該方案優(yōu)化之后肩刃，執(zhí)行時(shí)間縮短到10分鐘左右祟霍，性能提升了6倍。

1 // 首先將其中一個key分布相對較為均勻的RDD膨脹100倍盈包。

2 JavaPairRDD expandedRDD = rdd1.flatMapToPair(

3? ? ? ? new PairFlatMapFunction, String, Row>() {

4? ? ? ? ? ? private static final long serialVersionUID = 1L;

5? ? ? ? ? ? @Override

6? ? ? ? ? ? public Iterable> call(Tuple2 tuple)

7? ? ? ? ? ? ? ? ? ? throws Exception {

8? ? ? ? ? ? ? ? List> list = new ArrayList>();

9? ? ? ? ? ? ? ? for(int i = 0; i < 100; i++) {

10? ? ? ? ? ? ? ? ? ? list.add(new Tuple2(0 + "_" + tuple._1, tuple._2));

11? ? ? ? ? ? ? ? }

12? ? ? ? ? ? ? ? return list;

13? ? ? ? ? ? }

14? ? ? ? });

15

16 // 其次沸呐，將另一個有數(shù)據(jù)傾斜key的RDD，每條數(shù)據(jù)都打上100以內(nèi)的隨機(jī)前綴呢燥。

17 JavaPairRDD mappedRDD = rdd2.mapToPair(

18? ? ? ? new PairFunction, String, String>() {

19? ? ? ? ? ? private static final long serialVersionUID = 1L;

20? ? ? ? ? ? @Override

21? ? ? ? ? ? public Tuple2 call(Tuple2 tuple)

22? ? ? ? ? ? ? ? ? ? throws Exception {

23? ? ? ? ? ? ? ? Random random = new Random();

24? ? ? ? ? ? ? ? int prefix = random.nextInt(100);

25? ? ? ? ? ? ? ? return new Tuple2(prefix + "_" + tuple._1, tuple._2);

26? ? ? ? ? ? }

27? ? ? ? });

28

29 // 將兩個處理后的RDD進(jìn)行join即可崭添。

30 JavaPairRDD> joinedRDD = mappedRDD.join(expandedRDD);

解決方案八：多種方案組合使用

在實(shí)踐中發(fā)現(xiàn)，很多情況下叛氨，如果只是處理較為簡單的數(shù)據(jù)傾斜場景呼渣，那么使用上述方案中的某一種基本就可以解決。但是如果要處理一個較為復(fù)雜的數(shù)據(jù)傾斜場景寞埠，那么可能需要將多種方案組合起來使用屁置。比如說，我們針對出現(xiàn)了多個數(shù)據(jù)傾斜環(huán)節(jié)的Spark作業(yè)仁连，可以先運(yùn)用解決方案一和二蓝角，預(yù)處理一部分?jǐn)?shù)據(jù)，并過濾一部分?jǐn)?shù)據(jù)來緩解怖糊；其次可以對某些shuffle操作提升并行度帅容，優(yōu)化其性能颇象；最后還可以針對不同的聚合或join操作伍伤，選擇一種方案來優(yōu)化其性能。大家需要對這些方案的思路和原理都透徹理解之后遣钳，在實(shí)踐中根據(jù)各種不同的情況扰魂，靈活運(yùn)用多種方案，來解決自己的數(shù)據(jù)傾斜問題蕴茴。

shuffle調(diào)優(yōu)

調(diào)優(yōu)概述

大多數(shù)Spark作業(yè)的性能主要就是消耗在了shuffle環(huán)節(jié)劝评，因?yàn)樵摥h(huán)節(jié)包含了大量的磁盤IO、序列化倦淀、網(wǎng)絡(luò)數(shù)據(jù)傳輸?shù)炔僮鹘蟆Ｒ虼耍绻屪鳂I(yè)的性能更上一層樓撞叽，就有必要對shuffle過程進(jìn)行調(diào)優(yōu)姻成。但是也必須提醒大家的是插龄，影響一個Spark作業(yè)性能的因素，主要還是代碼開發(fā)科展、資源參數(shù)以及數(shù)據(jù)傾斜均牢，shuffle調(diào)優(yōu)只能在整個Spark的性能調(diào)優(yōu)中占到一小部分而已。因此大家務(wù)必把握住調(diào)優(yōu)的基本原則才睹，千萬不要舍本逐末徘跪。下面我們就給大家詳細(xì)講解shuffle的原理，以及相關(guān)參數(shù)的說明琅攘，同時(shí)給出各個參數(shù)的調(diào)優(yōu)建議垮庐。

ShuffleManager發(fā)展概述

在Spark的源碼中，負(fù)責(zé)shuffle過程的執(zhí)行乎澄、計(jì)算和處理的組件主要就是ShuffleManager突硝，也即shuffle管理器。而隨著Spark的版本的發(fā)展置济，ShuffleManager也在不斷迭代解恰，變得越來越先進(jìn)。

在Spark 1.2以前浙于，默認(rèn)的shuffle計(jì)算引擎是HashShuffleManager护盈。該ShuffleManager而HashShuffleManager有著一個非常嚴(yán)重的弊端，就是會產(chǎn)生大量的中間磁盤文件羞酗，進(jìn)而由大量的磁盤IO操作影響了性能腐宋。

因此在Spark 1.2以后的版本中，默認(rèn)的ShuffleManager改成了SortShuffleManager檀轨。SortShuffleManager相較于HashShuffleManager來說胸竞，有了一定的改進(jìn)。主要就在于参萄，每個Task在進(jìn)行shuffle操作時(shí)卫枝，雖然也會產(chǎn)生較多的臨時(shí)磁盤文件，但是最后會將所有的臨時(shí)文件合并（merge）成一個磁盤文件讹挎，因此每個Task就只有一個磁盤文件校赤。在下一個stage的shuffle read task拉取自己的數(shù)據(jù)時(shí)，只要根據(jù)索引讀取每個磁盤文件中的部分?jǐn)?shù)據(jù)即可筒溃。

下面我們詳細(xì)分析一下HashShuffleManager和SortShuffleManager的原理马篮。

HashShuffleManager運(yùn)行原理

未經(jīng)優(yōu)化的HashShuffleManager

下圖說明了未經(jīng)優(yōu)化的HashShuffleManager的原理。這里我們先明確一個假設(shè)前提：每個Executor只有1個CPU core怜奖，也就是說浑测，無論這個Executor上分配多少個task線程，同一時(shí)間都只能執(zhí)行一個task線程歪玲。

我們先從shuffle write開始說起迁央。shuffle write階段怎顾，主要就是在一個stage結(jié)束計(jì)算之后，為了下一個stage可以執(zhí)行shuffle類的算子（比如reduceByKey）漱贱，而將每個task處理的數(shù)據(jù)按key進(jìn)行“分類”槐雾。所謂“分類”，就是對相同的key執(zhí)行hash算法幅狮，從而將相同key都寫入同一個磁盤文件中募强，而每一個磁盤文件都只屬于下游stage的一個task。在將數(shù)據(jù)寫入磁盤之前崇摄，會先將數(shù)據(jù)寫入內(nèi)存緩沖中擎值，當(dāng)內(nèi)存緩沖填滿之后，才會溢寫到磁盤文件中去逐抑。

那么每個執(zhí)行shuffle write的task鸠儿，要為下一個stage創(chuàng)建多少個磁盤文件呢？很簡單厕氨，下一個stage的task有多少個进每，當(dāng)前stage的每個task就要創(chuàng)建多少份磁盤文件。比如下一個stage總共有100個task命斧，那么當(dāng)前stage的每個task都要創(chuàng)建100份磁盤文件田晚。如果當(dāng)前stage有50個task，總共有10個Executor国葬，每個Executor執(zhí)行5個Task贤徒，那么每個Executor上總共就要創(chuàng)建500個磁盤文件，所有Executor上會創(chuàng)建5000個磁盤文件汇四。由此可見接奈，未經(jīng)優(yōu)化的shuffle write操作所產(chǎn)生的磁盤文件的數(shù)量是極其驚人的。

接著我們來說說shuffle read通孽。shuffle read序宦，通常就是一個stage剛開始時(shí)要做的事情。此時(shí)該stage的每一個task就需要將上一個stage的計(jì)算結(jié)果中的所有相同key利虫，從各個節(jié)點(diǎn)上通過網(wǎng)絡(luò)都拉取到自己所在的節(jié)點(diǎn)上挨厚，然后進(jìn)行key的聚合或連接等操作堡僻。由于shuffle write的過程中糠惫，task給下游stage的每個task都創(chuàng)建了一個磁盤文件，因此shuffle read的過程中钉疫，每個task只要從上游stage的所有task所在節(jié)點(diǎn)上硼讽，拉取屬于自己的那一個磁盤文件即可。

shuffle read的拉取過程是一邊拉取一邊進(jìn)行聚合的牲阁。每個shuffle read task都會有一個自己的buffer緩沖固阁，每次都只能拉取與buffer緩沖相同大小的數(shù)據(jù)壤躲，然后通過內(nèi)存中的一個Map進(jìn)行聚合等操作。聚合完一批數(shù)據(jù)后备燃，再拉取下一批數(shù)據(jù)碉克，并放到buffer緩沖中進(jìn)行聚合操作。以此類推并齐，直到最后將所有數(shù)據(jù)到拉取完漏麦，并得到最終的結(jié)果。

SortShuffleManager運(yùn)行原理

SortShuffleManager的運(yùn)行機(jī)制主要分成兩種况褪，一種是普通運(yùn)行機(jī)制撕贞，另一種是bypass運(yùn)行機(jī)制。當(dāng)shuffle read task的數(shù)量小于等于spark.shuffle.sort.bypassMergeThreshold參數(shù)的值時(shí)（默認(rèn)為200）测垛，就會啟用bypass機(jī)制捏膨。

普通運(yùn)行機(jī)制

下圖說明了普通的SortShuffleManager的原理。在該模式下食侮，數(shù)據(jù)會先寫入一個內(nèi)存數(shù)據(jù)結(jié)構(gòu)中号涯，此時(shí)根據(jù)不同的shuffle算子，可能選用不同的數(shù)據(jù)結(jié)構(gòu)锯七。如果是reduceByKey這種聚合類的shuffle算子诚隙，那么會選用Map數(shù)據(jù)結(jié)構(gòu)，一邊通過Map進(jìn)行聚合起胰，一邊寫入內(nèi)存久又；如果是join這種普通的shuffle算子，那么會選用Array數(shù)據(jù)結(jié)構(gòu)效五，直接寫入內(nèi)存地消。接著，每寫一條數(shù)據(jù)進(jìn)入內(nèi)存數(shù)據(jù)結(jié)構(gòu)之后畏妖，就會判斷一下脉执，是否達(dá)到了某個臨界閾值。如果達(dá)到臨界閾值的話戒劫，那么就會嘗試將內(nèi)存數(shù)據(jù)結(jié)構(gòu)中的數(shù)據(jù)溢寫到磁盤半夷，然后清空內(nèi)存數(shù)據(jù)結(jié)構(gòu)。

在溢寫到磁盤文件之前迅细，會先根據(jù)key對內(nèi)存數(shù)據(jù)結(jié)構(gòu)中已有的數(shù)據(jù)進(jìn)行排序巫橄。排序過后，會分批將數(shù)據(jù)寫入磁盤文件茵典。默認(rèn)的batch數(shù)量是10000條湘换，也就是說，排序好的數(shù)據(jù)，會以每批1萬條數(shù)據(jù)的形式分批寫入磁盤文件彩倚。寫入磁盤文件是通過Java的BufferedOutputStream實(shí)現(xiàn)的筹我。BufferedOutputStream是Java的緩沖輸出流，首先會將數(shù)據(jù)緩沖在內(nèi)存中帆离，當(dāng)內(nèi)存緩沖滿溢之后再一次寫入磁盤文件中蔬蕊，這樣可以減少磁盤IO次數(shù)，提升性能哥谷。

一個task將所有數(shù)據(jù)寫入內(nèi)存數(shù)據(jù)結(jié)構(gòu)的過程中袁串，會發(fā)生多次磁盤溢寫操作，也就會產(chǎn)生多個臨時(shí)文件呼巷。最后會將之前所有的臨時(shí)磁盤文件都進(jìn)行合并囱修，這就是merge過程，此時(shí)會將之前所有臨時(shí)磁盤文件中的數(shù)據(jù)讀取出來王悍，然后依次寫入最終的磁盤文件之中破镰。此外，由于一個task就只對應(yīng)一個磁盤文件压储，也就意味著該task為下游stage的task準(zhǔn)備的數(shù)據(jù)都在這一個文件中鲜漩，因此還會單獨(dú)寫一份索引文件，其中標(biāo)識了下游各個task的數(shù)據(jù)在文件中的start offset與end offset集惋。

SortShuffleManager由于有一個磁盤文件merge的過程孕似，因此大大減少了文件數(shù)量。比如第一個stage有50個task刮刑，總共有10個Executor喉祭，每個Executor執(zhí)行5個task，而第二個stage有100個task雷绢。由于每個task最終只有一個磁盤文件泛烙，因此此時(shí)每個Executor上只有5個磁盤文件，所有Executor只有50個磁盤文件翘紊。

bypass運(yùn)行機(jī)制

下圖說明了bypass SortShuffleManager的原理蔽氨。bypass運(yùn)行機(jī)制的觸發(fā)條件如下：

shuffle map task數(shù)量小于spark.shuffle.sort.bypassMergeThreshold參數(shù)的值。

不是聚合類的shuffle算子（比如reduceByKey）帆疟。

此時(shí)task會為每個下游task都創(chuàng)建一個臨時(shí)磁盤文件鹉究，并將數(shù)據(jù)按key進(jìn)行hash然后根據(jù)key的hash值，將key寫入對應(yīng)的磁盤文件之中踪宠。當(dāng)然自赔，寫入磁盤文件時(shí)也是先寫入內(nèi)存緩沖，緩沖寫滿之后再溢寫到磁盤文件的殴蓬。最后匿级，同樣會將所有臨時(shí)磁盤文件都合并成一個磁盤文件，并創(chuàng)建一個單獨(dú)的索引文件染厅。

該過程的磁盤寫機(jī)制其實(shí)跟未經(jīng)優(yōu)化的HashShuffleManager是一模一樣的痘绎，因?yàn)槎家獎?chuàng)建數(shù)量驚人的磁盤文件，只是在最后會做一個磁盤文件的合并而已肖粮。因此少量的最終磁盤文件孤页，也讓該機(jī)制相對未經(jīng)優(yōu)化的HashShuffleManager來說，shuffle read的性能會更好涩馆。

而該機(jī)制與普通SortShuffleManager運(yùn)行機(jī)制的不同在于：第一行施，磁盤寫機(jī)制不同；第二魂那，不會進(jìn)行排序蛾号。也就是說，啟用該機(jī)制的最大好處在于涯雅，shuffle write過程中鲜结，不需要進(jìn)行數(shù)據(jù)的排序操作，也就節(jié)省掉了這部分的性能開銷活逆。

shuffle相關(guān)參數(shù)調(diào)優(yōu)

以下是Shffule過程中的一些主要參數(shù)精刷，這里詳細(xì)講解了各個參數(shù)的功能、默認(rèn)值以及基于實(shí)踐經(jīng)驗(yàn)給出的調(diào)優(yōu)建議蔗候。

spark.shuffle.file.buffer

默認(rèn)值：32k

參數(shù)說明：該參數(shù)用于設(shè)置shuffle write task的BufferedOutputStream的buffer緩沖大小怒允。將數(shù)據(jù)寫到磁盤文件之前，會先寫入buffer緩沖中锈遥，待緩沖寫滿之后纫事，才會溢寫到磁盤。

調(diào)優(yōu)建議：如果作業(yè)可用的內(nèi)存資源較為充足的話所灸，可以適當(dāng)增加這個參數(shù)的大卸瘛（比如64k），從而減少shuffle write過程中溢寫磁盤文件的次數(shù)庆寺，也就可以減少磁盤IO次數(shù)蚊夫，進(jìn)而提升性能。在實(shí)踐中發(fā)現(xiàn)懦尝，合理調(diào)節(jié)該參數(shù)知纷，性能會有1%~5%的提升。

spark.reducer.maxSizeInFlight

默認(rèn)值：48m

參數(shù)說明：該參數(shù)用于設(shè)置shuffle read task的buffer緩沖大小陵霉，而這個buffer緩沖決定了每次能夠拉取多少數(shù)據(jù)琅轧。

調(diào)優(yōu)建議：如果作業(yè)可用的內(nèi)存資源較為充足的話，可以適當(dāng)增加這個參數(shù)的大杏荒印（比如96m）乍桂，從而減少拉取數(shù)據(jù)的次數(shù)冲杀，也就可以減少網(wǎng)絡(luò)傳輸?shù)拇螖?shù)，進(jìn)而提升性能睹酌。在實(shí)踐中發(fā)現(xiàn)权谁，合理調(diào)節(jié)該參數(shù)，性能會有1%~5%的提升憋沿。

spark.shuffle.io.maxRetries

默認(rèn)值：3

參數(shù)說明：shuffle read task從shuffle write task所在節(jié)點(diǎn)拉取屬于自己的數(shù)據(jù)時(shí)旺芽，如果因?yàn)榫W(wǎng)絡(luò)異常導(dǎo)致拉取失敗，是會自動進(jìn)行重試的辐啄。該參數(shù)就代表了可以重試的最大次數(shù)采章。如果在指定次數(shù)之內(nèi)拉取還是沒有成功，就可能會導(dǎo)致作業(yè)執(zhí)行失敗壶辜。

調(diào)優(yōu)建議：對于那些包含了特別耗時(shí)的shuffle操作的作業(yè)悯舟，建議增加重試最大次數(shù)（比如60次），以避免由于JVM的full gc或者網(wǎng)絡(luò)不穩(wěn)定等因素導(dǎo)致的數(shù)據(jù)拉取失敗砸民。在實(shí)踐中發(fā)現(xiàn)图谷，對于針對超大數(shù)據(jù)量（數(shù)十億~上百億）的shuffle過程，調(diào)節(jié)該參數(shù)可以大幅度提升穩(wěn)定性阱洪。

spark.shuffle.io.retryWait

默認(rèn)值：5s

參數(shù)說明：具體解釋同上便贵，該參數(shù)代表了每次重試?yán)?shù)據(jù)的等待間隔，默認(rèn)是5s冗荸。

調(diào)優(yōu)建議：建議加大間隔時(shí)長（比如60s）承璃，以增加shuffle操作的穩(wěn)定性。

spark.shuffle.memoryFraction

默認(rèn)值：0.2

參數(shù)說明：該參數(shù)代表了Executor內(nèi)存中蚌本，分配給shuffle read task進(jìn)行聚合操作的內(nèi)存比例盔粹，默認(rèn)是20%。

調(diào)優(yōu)建議：在資源參數(shù)調(diào)優(yōu)中講解過這個參數(shù)程癌。如果內(nèi)存充足舷嗡，而且很少使用持久化操作，建議調(diào)高這個比例嵌莉，給shuffle read的聚合操作更多內(nèi)存进萄，以避免由于內(nèi)存不足導(dǎo)致聚合過程中頻繁讀寫磁盤。在實(shí)踐中發(fā)現(xiàn)锐峭，合理調(diào)節(jié)該參數(shù)可以將性能提升10%左右中鼠。

spark.shuffle.manager

默認(rèn)值：sort

參數(shù)說明：該參數(shù)用于設(shè)置ShuffleManager的類型。Spark 1.5以后沿癞，有三個可選項(xiàng)：hash援雇、sort和tungsten-sort。HashShuffleManager是Spark 1.2以前的默認(rèn)選項(xiàng)椎扬，但是Spark 1.2以及之后的版本默認(rèn)都是SortShuffleManager了惫搏。tungsten-sort與sort類似具温，但是使用了tungsten計(jì)劃中的堆外內(nèi)存管理機(jī)制，內(nèi)存使用效率更高筐赔。

調(diào)優(yōu)建議：由于SortShuffleManager默認(rèn)會對數(shù)據(jù)進(jìn)行排序铣猩，因此如果你的業(yè)務(wù)邏輯中需要該排序機(jī)制的話，則使用默認(rèn)的SortShuffleManager就可以川陆；而如果你的業(yè)務(wù)邏輯不需要對數(shù)據(jù)進(jìn)行排序剂习，那么建議參考后面的幾個參數(shù)調(diào)優(yōu)蛮位，通過bypass機(jī)制或優(yōu)化的HashShuffleManager來避免排序操作较沪，同時(shí)提供較好的磁盤讀寫性能。這里要注意的是失仁，tungsten-sort要慎用尸曼，因?yàn)橹鞍l(fā)現(xiàn)了一些相應(yīng)的bug。

spark.shuffle.sort.bypassMergeThreshold

默認(rèn)值：200

參數(shù)說明：當(dāng)ShuffleManager為SortShuffleManager時(shí)萄焦，如果shuffle read task的數(shù)量小于這個閾值（默認(rèn)是200）控轿，則shuffle write過程中不會進(jìn)行排序操作，而是直接按照未經(jīng)優(yōu)化的HashShuffleManager的方式去寫數(shù)據(jù)拂封，但是最后會將每個task產(chǎn)生的所有臨時(shí)磁盤文件都合并成一個文件茬射，并會創(chuàng)建單獨(dú)的索引文件。

調(diào)優(yōu)建議：當(dāng)你使用SortShuffleManager時(shí)冒签，如果的確不需要排序操作在抛，那么建議將這個參數(shù)調(diào)大一些，大于shuffle read task的數(shù)量萧恕。那么此時(shí)就會自動啟用bypass機(jī)制刚梭，map-side就不會進(jìn)行排序了，減少了排序的性能開銷票唆。但是這種方式下朴读，依然會產(chǎn)生大量的磁盤文件，因此shuffle write性能有待提高走趋。

spark.shuffle.consolidateFiles

默認(rèn)值：false

參數(shù)說明：如果使用HashShuffleManager衅金，該參數(shù)有效。如果設(shè)置為true簿煌，那么就會開啟consolidate機(jī)制典挑，會大幅度合并shuffle write的輸出文件，對于shuffle read task數(shù)量特別多的情況下啦吧，這種方法可以極大地減少磁盤IO開銷您觉，提升性能。

調(diào)優(yōu)建議：如果的確不需要SortShuffleManager的排序機(jī)制授滓，那么除了使用bypass機(jī)制琳水，還可以嘗試將spark.shffle.manager參數(shù)手動指定為hash肆糕，使用HashShuffleManager，同時(shí)開啟consolidate機(jī)制在孝。在實(shí)踐中嘗試過诚啃，發(fā)現(xiàn)其性能比開啟了bypass機(jī)制的SortShuffleManager要高出10%~30%。

寫在最后的話

本文分別講解了開發(fā)過程中的優(yōu)化原則私沮、運(yùn)行前的資源參數(shù)設(shè)置調(diào)優(yōu)始赎、運(yùn)行中的數(shù)據(jù)傾斜的解決方案、為了精益求精的shuffle調(diào)優(yōu)仔燕。希望大家能夠在閱讀本文之后造垛，記住這些性能調(diào)優(yōu)的原則以及方案，在Spark作業(yè)開發(fā)晰搀、測試以及運(yùn)行的過程中多嘗試五辽，只有這樣，我們才能開發(fā)出更優(yōu)的Spark作業(yè)外恕，不斷提升其性能杆逗。