做Hadoop這塊的不管你是剛跑通WordCount還是已經(jīng)在維護幾十個節(jié)點的集群早晚都會撞上Shuffle這個詞。很多人對Shuffle的理解停留在“Map和Reduce之間的那一段”但真正能把Shuffle講清楚、能把Shuffle調(diào)明白的人不多。原因也簡單Shuffle不是單一環(huán)節(jié)而是跨JVM、跨節(jié)點、跨磁盤的一系列操作問題藏在細節(jié)里光看監(jiān)控面板看不出來。這篇文章我從Map端寫到Reduce端把每個關(guān)鍵機制的原理、參數(shù)、踩坑點揉碎了講最后給一批能直接用的優(yōu)化思路。文章可能有點長但每一步都值得細看尤其是那些帶“我試過”字樣的段落都是真金白銀換來的經(jīng)驗。1. Shuffle到底在解決什么問題先把MR全流程跑通先對齊一下最基礎(chǔ)的東西。MapReduce作業(yè)跑起來之后Map Task把輸入數(shù)據(jù)切成一條條記錄執(zhí)行完用戶寫的map邏輯輸出的是形如key, value的中間結(jié)果。Reduce Task要做的是把相同key的所有value聚合在一起再執(zhí)行reduce邏輯。問題來了同一個key可能散落在不同節(jié)點上的不同Map Task里怎么把這些數(shù)據(jù)高效歸攏到同一個Reduce Task手里這就是Shuffle要解決的核心問題。很多教材把Shuffle拆成“Map端Shuffle”和“Reduce端Shuffle”這么說沒錯但容易讓人忽略一個關(guān)鍵點Shuffle是一個跨節(jié)點的數(shù)據(jù)搬運過程搬運的不是文件本身而是數(shù)據(jù)流的控制權(quán)。Map端負責(zé)把中間結(jié)果按照分區(qū)規(guī)則寫好、排好序、盡量壓縮Reduce端負責(zé)把屬于自己分區(qū)的數(shù)據(jù)拉過來、合并、再排序。兩邊各干各的靠的是一套約定好的協(xié)議和文件格式。從網(wǎng)絡(luò)視角看Shuffle還涉及一次“數(shù)據(jù)本地性”的放棄。Map Task輸出時數(shù)據(jù)還留在本機的臨時目錄里Reduce Task一開始拉數(shù)據(jù)數(shù)據(jù)就從計算節(jié)點流動到了另一個可能沒有任何輸入數(shù)據(jù)副本的節(jié)點上。這必然帶來網(wǎng)絡(luò)IO和磁盤IO開銷所以Shuffle也一直是MR作業(yè)性能瓶頸的高發(fā)區(qū)。我見過不少作業(yè)Map階段跑得飛快Reduce階段直接卡死根因十有八九都在Shuffle上。2. Map端Shuffle的核心機制與細節(jié)拆解2.1 環(huán)形緩沖區(qū)中間結(jié)果先寫內(nèi)存不是直接落盤Map Task跑完一個map()函數(shù)后輸出并不會直接寫入磁盤而是先進入一個環(huán)形緩沖區(qū)。這個緩沖區(qū)的默認大小由mapreduce.task.io.sort.mb控制老版本叫io.sort.mb默認100MB。緩沖區(qū)設(shè)計成環(huán)形是因為Map輸出是持續(xù)產(chǎn)生的寫指針和溢寫指針交替前進能最大程度復(fù)用已分配的內(nèi)存空間避免頻繁申請和釋放。緩沖區(qū)的使用比例有個閾值由mapreduce.map.sort.spill.percent控制默認0.80也就是80MB時觸發(fā)一次溢寫。溢寫不是把整個緩沖區(qū)清空而是把緩沖區(qū)里已經(jīng)排序好的那部分數(shù)據(jù)刷到磁盤。注意這里的“排序好的”底層是Map Task在寫入緩沖區(qū)的同時會啟動后臺線程對分區(qū)內(nèi)的key做排序。所以每次溢寫產(chǎn)生的文件內(nèi)部已經(jīng)是按分區(qū)和key排好序的。我踩過一個坑把mapreduce.task.io.sort.mb調(diào)得過大比如500MB但沒有同步調(diào)整JVM堆內(nèi)存。Map Task的JVM堆就那么大緩沖區(qū)占太多留給用戶代碼的堆就少了結(jié)果跑復(fù)雜map邏輯時頻繁Full GC反而比調(diào)參之前更慢。后面我會再說內(nèi)存參數(shù)怎么配才匹配。2.2 分區(qū)、排序與Combiner的協(xié)作邏輯分區(qū)器Partitioner決定每一條map輸出記錄應(yīng)該進哪個Reduce Task。默認是HashPartitioner對key做hash后再對Reduce數(shù)量取模。這個邏輯很簡單但實際生產(chǎn)里經(jīng)常要改。比如數(shù)據(jù)傾斜嚴重時同樣的key集中在一個Reduce上其他Reduce閑著這時候就必須自定義分區(qū)器把熱點key拆開。Combiner是個容易被低估的優(yōu)化點。它的本質(zhì)是在Map端做一次局部reduce把key, [v1, v2, v3...]提前合并成key, [V_total]從而減少溢寫文件里需要排序和傳輸?shù)臄?shù)據(jù)量。Combiner必須滿足一個硬性條件它的輸入輸出類型要和reduce函數(shù)一致而且重復(fù)應(yīng)用要等價于一次性應(yīng)用。比如求和、最大值、去重都可以用Combiner但求平均值不行除非你改成傳遞sum, count再做二次計算。這里有個細節(jié)Combiner不是只在地圖端溢寫時運行一次。你可以在兩個位置觸發(fā)它一是在溢寫文件內(nèi)部做局部合并時二是合并多個溢寫文件生成最終分區(qū)文件時。每個位置都跑一遍累加的效果是對的但要注意重復(fù)執(zhí)行的語義沖突。我自己被這個坑過一次求平均的時候按直覺寫了Combiner結(jié)果輸出完全不對。2.3 溢寫文件的合并與Index文件生成每次溢寫會生成兩個文件一個是數(shù)據(jù)文件另一個是Index文件。Index文件記錄了每個分區(qū)在數(shù)據(jù)文件里的偏移量和長度這樣Reduce端拉取時可以直接定位不用全文件掃描。溢寫次數(shù)一多Map Task結(jié)束前還會對多個溢寫文件做一次合并merge合并因子由mapreduce.task.io.sort.factor控制默認10。也就是說最多同時合并10個文件流。合并后的最終文件會等待Reduce端來拉。但數(shù)據(jù)文件本身不會做完整的全局排序它只保證分區(qū)與分區(qū)之間、分區(qū)內(nèi)部的key有序。注意一個容易誤解的點Map端最終輸出是有多個分區(qū)的Reduce端只會拉取屬于自己分區(qū)的那一段因此Map端并不需要把所有分區(qū)排成一個整體大文件而是按分區(qū)索引管理。從這個角度看Map端Shuffle的本質(zhì)是做“分區(qū) 局部排序 盡可能壓縮”目標是讓Reduce端拉取時盡量少傳數(shù)據(jù)、少做排序。很多優(yōu)化點都圍繞“減少落盤數(shù)據(jù)量”展開包括開啟壓縮、用Combiner、合理控制溢寫次數(shù)。3. Reduce端Shuffle拉取、合并與歸并排序的完整鏈路3.1 拉取階段的并發(fā)數(shù)與限流控制Reduce Task啟動后并不會等所有Map Task都跑完才開始拉數(shù)據(jù)而是會啟動若干Fetcher線程去已完成Map Task的節(jié)點上拉取屬于自己分區(qū)的數(shù)據(jù)。并發(fā)拉取的數(shù)量由mapreduce.reduce.shuffle.parallelcopies控制默認是5這個值很多資料建議調(diào)到10到15但要看集群網(wǎng)絡(luò)帶寬。這里有個交互細節(jié)Reduce Task內(nèi)部有一個Http ShuffleHandler運行在NodeManager進程里負責(zé)響應(yīng)Reduce端的拉取請求。拉回來的數(shù)據(jù)會先放到Reduce端的JVM內(nèi)存中內(nèi)存上限由mapreduce.reduce.shuffle.input.buffer.percent控制默認0.70也就是Reduce堆內(nèi)存的70%。超過這個比例數(shù)據(jù)就會被迫落盤等待合并。拉取階段還涉及一個容易迷惑的參數(shù)mapreduce.reduce.shuffle.connect.timeout與read.timeout。這兩個超時時間控制的是繞過NodeManager時的一致性默認值180秒實際上對慢節(jié)點是大問題。比如某個Map Task所在節(jié)點負載很高響應(yīng)延遲超過超時時間Reduce端會判定拉取失敗反復(fù)重試。所以你在集群里看到大量“Shuffle Error”日志時不一定是網(wǎng)絡(luò)壞了先看是不是超時設(shè)置不合理。3.2 內(nèi)存合并模式和落盤合并模式的選擇Reduce端接收數(shù)據(jù)后需要把來自多個Map Task的同一分區(qū)數(shù)據(jù)合并成一個有序的輸入。這里涉及兩種模式內(nèi)存合并和磁盤合并。默認情況下Reduce端會優(yōu)先在內(nèi)存中進行多路歸并內(nèi)存不足時才寫臨時文件然后再對這些臨時文件做合并。觸發(fā)落盤合并的條件有兩個內(nèi)存已滿由mapreduce.reduce.merge.memtomem.enabled和閾值控制或者Map輸出的數(shù)量達到一定閾值。mapreduce.reduce.merge.inmem.threshold默認1000意思是內(nèi)存中緩存了1000個Map輸出片段時就觸發(fā)一次合并。這個值并不需要調(diào)大因為頻繁的mem-to-mem合并會浪費CPU。真正值得關(guān)注的是mapreduce.reduce.memory.total默認由JVM堆的mapreduce.reduce.java.opts決定以及mapreduce.reduce.shuffle.memory.limit.percent等參數(shù)。在很多版本里Reduce端的Shuffle內(nèi)存是動態(tài)計算的不當(dāng)?shù)亩言O(shè)置會直接導(dǎo)致Reduce階段頻繁溢寫磁盤IO飆升。我見過一臺機器上Reduce階段半小時不動日志全是“Merger”和“Spill”就是內(nèi)存分配不合理導(dǎo)致的。3.3 Copy、Sort與歸并排序的先后關(guān)系這里要明確一點Reduce端Shuffle并不是“先全部拉完再統(tǒng)一排序”而是邊拉邊合。拉取的每個Map輸出片段單獨看已經(jīng)是有序的Reduce端需要把這些有序片段做歸并排序最終得到整個分區(qū)內(nèi)key有序的輸入流。歸并排序的輸入分段數(shù)直接決定性能。如果設(shè)置了mapreduce.reduce.input.buffer.percent并開啟了內(nèi)存合并分段數(shù)會更多集中到內(nèi)存里否則會落到磁盤由多個臨時文件參與歸并。mapreduce.task.io.sort.factor同樣影響Reduce端的歸并路數(shù)調(diào)大會減少歸并輪數(shù)但會提高單次歸并的內(nèi)存占用。簡化一下這里的策略想要Reduce端少落盤就要盡量擴大Shuffle內(nèi)存想要Reduce端跑得快就要讓歸并輪數(shù)盡量少也就是調(diào)大排序因子。兩個目標要平衡不能顧頭不顧尾。4. Shuffle性能優(yōu)化實戰(zhàn)參數(shù)調(diào)優(yōu)與方案選型4.1 內(nèi)存類參數(shù)別只盯著mapreduce.map.memory.mb做Shuffle優(yōu)化的人第一反應(yīng)往往是調(diào)大mapreduce.map.memory.mb和mapreduce.reduce.memory.mb但這只是給容器用的內(nèi)存上限。真正影響Shuffle行為的是JVM堆大小、Shuffle緩沖占比、排序占比這些參數(shù)。我建議一套我自己在用的配比思路方便你直接參考。假設(shè)Map容器內(nèi)存是2GB可以設(shè)置mapreduce.map.java.opts-Xmx1536m mapreduce.task.io.sort.mb800 mapreduce.map.sort.spill.percent0.80 mapreduce.task.io.sort.factor32這里的邏輯是JVM堆1.5GB留給Shuffle排序緩沖區(qū)800MB占比約53%剩下留給用戶代碼。這個比例在大多數(shù)場景下比較穩(wěn)既不會讓用戶代碼因為堆太小頻繁GC也不會讓Shuffle緩沖區(qū)太小導(dǎo)致頻繁溢寫。如果你的Map邏輯內(nèi)存開銷特別大需要把mapreduce.task.io.sort.mb降下來比如降到512MB而不是硬撐800MB。Reduce端同理我常用的配比是mapreduce.reduce.java.opts-Xmx2048m mapreduce.reduce.shuffle.input.buffer.percent0.70 mapreduce.reduce.shuffle.parallelcopies10 mapreduce.reduce.merge.inmem.threshold1200 mapreduce.task.io.sort.factor32注意mapreduce.reduce.java.opts的堆大小不要超過容器內(nèi)存的80%否則YARN會直接殺掉容器。這個“80%安全線”是硬規(guī)則超過就OOM別去賭。4.2 Combiner與壓縮編碼減少落盤量和網(wǎng)絡(luò)IO的杠桿很多人調(diào)Shuffle只看內(nèi)存忽略了數(shù)據(jù)體積才是根本問題。Map端輸出的數(shù)據(jù)如果能在源頭壓縮溢寫文件更小、Reduce拉取的數(shù)據(jù)更少、磁盤IO和網(wǎng)絡(luò)IO都會下降。Hadoop原生支持的壓縮編碼里如果追求速度和壓縮比的平衡Snappy是最常用的如果追求高壓縮比且CPU有富余可以用LZ4或者Bzip2。開啟Shuffle階段壓縮需要兩個開關(guān)mapreduce.map.output.compresstrue mapreduce.map.output.compress.codecorg.apache.hadoop.io.compress.SnappyCodec注意這個壓縮只針對Map和Reduce之間的Shuffle數(shù)據(jù)不影響最終輸出落盤文件的壓縮兩者要分開配。我在生產(chǎn)環(huán)境跑過對比一個TB級別的作業(yè)開啟Snappy后Shuffle時間下降約30%代價是Map階段CPU使用率上升了不到5%總體非常劃算。Combiner的作用前面已經(jīng)講了這里再說一個經(jīng)驗不要只在本地測試時加Combiner因為本地模式樣本太小看不出差別。壓測時先不加Combiner跑一輪記下Shuffle階段數(shù)據(jù)量和耗時然后加上Combiner再來一輪用對比結(jié)果說話。如果Combiner本身邏輯復(fù)雜比如正則匹配、對象創(chuàng)建多的場景可能反而拖慢Map端那就要權(quán)衡。4.3 數(shù)據(jù)傾斜的識別與處理從分區(qū)邏輯動刀Shuffle階段最讓人頭疼的問題之一就是數(shù)據(jù)傾斜。癥狀很典型某個Reduce的Shuffle數(shù)據(jù)量是其他幾個Reduce的上百倍整個作業(yè)卡在最后一個Reduce上。原因大多是默認分區(qū)器對各key的Hash分布不均勻少數(shù)幾個key占據(jù)了大量數(shù)據(jù)。先要能識別。用Hadoop自帶的計數(shù)器看Reduce端輸入記錄數(shù)或者直接在ResourceManager頁面觀察各Reduce的Shuffle字節(jié)數(shù)。確認是傾斜后有三種常用手段如果傾斜key的語義允許讓Mapper端做二次聚合比如把key加上隨機前綴然后再做一輪不含隨機前綴的Reduce。這個辦法能分散數(shù)據(jù)但會增加一次MapReduce輪次改動較大。如果只是少數(shù)熱點key可以在自定義Partitioner里將熱點key單獨路由到某個Reduce其余key走默認Hash。這種做法要保證熱點key的數(shù)據(jù)不超出單個Reduce的處理能力。如果Reduce端必須做全局排序或全局聚合那么傾斜問題本質(zhì)上要靠重構(gòu)算法解決比如換成兩個階段先把數(shù)據(jù)打散聚合再做全局匯總。我實際用下來最省事的是先統(tǒng)計key分布確認熱點key數(shù)量再寫一個帶白名單的Partitioner。比如在分區(qū)之前先判斷key是否是熱點是就打特定的Reduce編號否則走默認邏輯。這個方案不改變Reduce數(shù)量也不改動業(yè)務(wù)邏輯代碼量大概一百行以內(nèi)。5. 常見問題與排查技巧實錄5.1 任務(wù)卡在Reduce Shuffle到99%的案例有一回我排查一個奇怪的任務(wù)Reduce階段進度條跑到99%就停住日志里沒有任何明顯的Exception作業(yè)一直掛著直到超時。這種“99%卡死”在社區(qū)里出現(xiàn)頻率非常高我遇到的根本原因不是Reduce端不干活而是某個Map Task的臨時輸出遲遲沒有清理Reduce端一直等待那個Map的輸出。Map節(jié)點可能已經(jīng)被殺掉但相關(guān)文件殘留或者NodeManager狀態(tài)異常。排查步驟是這樣的先去ResourceManager頁面看還有哪些Map Task處于“Running”或“KILLED”狀態(tài)重點看未被ACK的Map輸出。然后到對應(yīng)節(jié)點的nodemanager日志里搜ShuffleHandler的錯誤信息看是否報“Failed to fetch output”。最后用hdfs dfs -du檢查Map的中間輸出目錄/tmp/hadoop-yarn/staging/...看是否有殘留文件。解決路徑通常是清理掉殘留文件并重啟對應(yīng)節(jié)點或者調(diào)大mapreduce.reduce.shuffle.connect.timeout和read.timeout。這個案例也提醒我Shuffle階段的日志不要只盯著Task的stdout要看NodeManager的system.err里面往往有被吞掉的異常。5.2 Container OOM與頻繁GC的排查思路Reduce階段頻繁Full GC往往是Shuffle內(nèi)存占比不當(dāng)?shù)牡湫捅憩F(xiàn)。癥狀是CPU占用飆高、GC日志長期持續(xù)同時Container日志里出現(xiàn)java.lang.OutOfMemoryError: Java heap space。這里要先區(qū)分是用戶代碼內(nèi)存不足還是Shuffle緩沖擠占了用戶堆。有一個快速的區(qū)分方法臨時把mapreduce.reduce.java.opts調(diào)大比如從2GB調(diào)到3GB如果GC頻率明顯下降說明用戶代碼堆不夠如果沒什么變化則說明Shuffle緩沖邏輯有問題。另一個方法是打開mapreduce.reduce.memory.total日志看Reduce端各階段的內(nèi)存占用。我建議集群上默認開啟JVM監(jiān)控并把mapreduce.map.java.opts和mapreduce.reduce.java.opts顯式寫死不要依賴默認值。默認的-Xmx往往是容器內(nèi)存的75%或者更低實際不足以支撐Shuffle緩沖加用戶代碼的雙重消耗。寫死一堆Java參數(shù)到日志里也是后面排查的速度關(guān)鍵。5.3 失敗重試與推測執(zhí)行對Shuffle的次生影響Shuffle階段發(fā)生節(jié)點故障時Map輸出物往往已經(jīng)落盤Reduce端會重新發(fā)起拉取。但這里有個坑如果失敗發(fā)生在Shuffle進行中Reduce端可能已經(jīng)拉了一部分數(shù)據(jù)重試時需要重新拉取整個分區(qū)數(shù)據(jù)而不是增量拉取。這在大數(shù)據(jù)分區(qū)場景下會放大失敗代價。另一個隱蔽問題是推測執(zhí)行。當(dāng)mapreduce.map.speculative開啟時如果一個Map Task執(zhí)行得特別慢系統(tǒng)會啟動一個重復(fù)Map Task。兩個Map Task的中間結(jié)果都可能被Reduce端拉到失去“同一key只有一個輸出”的約束進而導(dǎo)致一些下游對順序敏感的邏輯出錯。我自己遇到過一次一個帶全局計數(shù)器的作業(yè)在開啟推測執(zhí)行后計數(shù)翻倍排查半天才發(fā)現(xiàn)是重復(fù)執(zhí)行導(dǎo)致的。所以如果你的作業(yè)對精確性要求極高或者Reduce邏輯依賴Map Tash數(shù)量做判斷建議在作業(yè)級別把推測執(zhí)行關(guān)掉或者把specific參數(shù)的閾值調(diào)高。這個參數(shù)默認是關(guān)閉的但有些發(fā)行版為了優(yōu)化速度會默認開啟檢查一遍沒壞處。5.4 常見問題速查表現(xiàn)象可能原因排查手段解決思路Reduce卡在99%Map輸出文件缺失或NodeManager異常查看ResourceManager的Task狀態(tài)和NodeManager錯誤日志清理殘留文件、重試節(jié)點、調(diào)整超時時間Reduce階段頻繁溢出、磁盤IO高Shuffle內(nèi)存占比不足或堆設(shè)置過小比較mapreduce.reduce.shuffle.input.buffer.percent和實際堆調(diào)大Reduce堆適當(dāng)調(diào)高緩沖占比作業(yè)卡在Copy階段網(wǎng)絡(luò)帶寬受限、parallelcopies過小觀察每個Reduce的拉取速度調(diào)大parallelcopies到10-15結(jié)合帶寬測試輸出結(jié)果計數(shù)翻倍推測執(zhí)行導(dǎo)致重復(fù)任務(wù)貢獻輸出檢查作業(yè)日志中是否存在重復(fù)TaskAttempt關(guān)閉推測執(zhí)行或提高觸發(fā)閾值單個Reduce數(shù)據(jù)量遠大于其他Reduce數(shù)據(jù)傾斜用計數(shù)器或UI頁面看各Reduce輸入量自定義分區(qū)器或二次聚合Map端溢寫次數(shù)遠超預(yù)期sort.mb不足或spill.percent過低觀察MR控制臺的Spill Counter增大sort.mb調(diào)整spill.percent和merge factorNodeManager被容器殺死java.opts超過容器內(nèi)存查看NM日志將java.opts控制在容器內(nèi)存的80%以內(nèi)6. 調(diào)優(yōu)實操總結(jié)與個人經(jīng)驗分享一套我每次遇到Shuffle問題都會先跑一遍的檢查清單。先用ResourceManager頁面看總Shuffle數(shù)據(jù)量和各Reduce的分布判斷有沒有明顯傾斜再看NM日志確認沒有ShuffleHandler報錯然后檢查作業(yè)里的mapreduce.map.output.compress是否開啟最后再看內(nèi)存相關(guān)參數(shù)。參數(shù)上我現(xiàn)在更傾向于在作業(yè)代碼里用Configuration.set動態(tài)設(shè)置而不是全局改yarn-site.xml。好處是每個作業(yè)可以根據(jù)自己的數(shù)據(jù)量、復(fù)雜度做差異化配置不至于把一次調(diào)優(yōu)的激進參數(shù)帶到所有作業(yè)里去。如果要給一個通用起點對于10GB到100GB輸入的中等作業(yè)這套配置可以先用起來mapreduce.map.output.compresstrue mapreduce.map.output.compress.codecorg.apache.hadoop.io.compress.SnappyCodec mapreduce.task.io.sort.mb600 mapreduce.reduce.shuffle.parallelcopies10 mapreduce.reduce.shuffle.input.buffer.percent0.70 mapreduce.reduce.java.opts-Xmx2048m mapreduce.task.io.sort.factor32這套參數(shù)不是萬能藥但它是我在多種業(yè)務(wù)場景下驗證過“最不容易出錯”的起點。再往下微調(diào)時一次只改一個參數(shù)觀察兩個完整作業(yè)不要為了圖快同時改一堆否則你根本不知道是哪一個參數(shù)起的作用。Shuffle階段的調(diào)優(yōu)更像是一個“木桶工程”內(nèi)存、網(wǎng)絡(luò)、磁盤、CPU每一項都可能成為瓶頸。沒有一套固定模板能覆蓋所有場景但只要你把機制吃透遇到問題時從“數(shù)據(jù)量減少、拉取速度提高、內(nèi)存利用率提升”這三條線去分析大部分問題都能找到答案。最后再說一個實用小技巧在寫作業(yè)時盡量在Reducer的cleanup階段把Shuffle相關(guān)計數(shù)器以自定義Counter的形式輸出到日志比如Shuffle字節(jié)數(shù)、溢出文件數(shù)。平時你可能覺得無所謂但等到你要做集群級調(diào)優(yōu)時這些歷史記錄會幫你快速定位哪個時間段、哪個作業(yè)的Shuffle行為發(fā)生了異常。這個習(xí)慣幫我省過不少排查時間值得養(yǎng)成。