mapreduce shuffle細節

來自專欄 數據研發技術

用過hadoop的同學都知道在mr 作業執行過程中存在一個shuffle環節，其實shuffle包含的子過程還是不少的，這次就來看看shuffle究竟都幹了什麼。

在map task執行完成之後會將中間結果暫存到本地，在數據落地之前其實已經經歷了好幾個環節，這幾個環節分別為partition，combine（如果有），spill_out，combine（如果有）。先來看partition，partition是用來將map的結果進行路由的，也就是分配給哪個reduce節點，用戶可以根據自己的需要定製partition，默認的partiton方法是用的哈希。在進行partition之後，得到的partition結果會和key/value一起存進環形緩衝區，在緩衝區內存儲的數據量如果超過一定閾值就會執行spill（溢出）操作，將緩衝區中的數據刷到本地文件中。不過還有一點特別重要的是如果設置了combiner（一般與reducer相同）,緩衝區中的數據會先執行一次combine，然後才是sortAndSpill，這裡是對key進行排序，使用的排序演算法為quick sort。當然，如果map out的數據量很大，那就有可能會產生多個spill_out文件，這裡還要對多個spill_out文件進行合併，在合併的過程中combine會再次被調用，同時合併後的結果也應該是有序的，所以這裡用到了多路歸併排序，多路歸併排序就是維護一個最小堆，從各個merge pass中取值放進這個堆然後取堆頂的值。多路歸併排序也有其他的實現方法，比如勝者樹，感興趣的自行查閱資料。最終的map產生的中間結果是一個數據文件，還有一個索引文件，該索引文件中包含每個partition在文件中的偏移量以便reducer讀取。

shuffle可不光跟map有關係，跟reduce同樣有聯繫，在reduce啟動後就會啟動一個ReduceCopier的線程不斷的從map產生的中間結果里取分配到自己數據。在取完數據後，reducer需要產生<key,List<v1,v2....>>的序列，那麼這裡的List中的數據也要經過排序，寫過SecondarySort的同學應該清楚這裡是怎麼回事，其實就是實現一個RawComparator，然後利用job.setGroupingComparatorClass進行設置即可。到此，整個shuffle才算結束。