@Author  : Spinach | GHB
@Link    : http://blog.csdn.net/bocai8058

MapReduce的資料傾斜和解決方式

資料傾斜

根據MapReduce的執行流程以發現，在map端是不會產生資料傾斜的，因為它會根據分片規則將資料進行均勻的切分成一個個固定大小的資料片段，而且每一個數據片段都會有一個maptask去處理，因此，資料傾斜只會在reduce端產生，因為map程式會將資料處理成鍵值對的形式，這樣就有可能出現有的鍵相同的資料多，有的鍵相同的資料少。因此在進行partition分割槽的時候會將鍵相同的資料分配到同一個reduce當中去，這樣就會導致有的reduce要處理資料會很多，有的就會很少，從而導致資料傾斜問題的產生。所以我們需要根據實際資料的特點自定義partition分割槽的規則，這樣就能很好的解決資料傾斜的問題。

解決方法

重寫Partitioner類的getPartition

public class Screw2Partition extends Partitioner<Text, IntWritable> 
{
     @Override
     public int getPartition(Text text, IntWritable intWritable, int numPartitions) 
     {
         Random r = new Random();
         //返回的是分割槽的隨機的一個ID
         return r.nextInt(numPartitions);
     }
}
 

• 自定義partition
• 隨機分割槽

MapReduce的優化方式

1. Map和Reduce的數量
2. 出現數據傾斜，需要自定義partitioner
3. 自定義combiner，還可以調參min.num.spill.for.combine，預設是3，也就是說spill的檔案數在預設情況下由三個的時候就要進行combine操作，最終減少磁碟資料；
4. MapReduce中Kvbuffer的大小預設100M，可以通過mapreduce.task.io.sort.mb（default：100）引數來調整。可以根據不同的硬體尤其是記憶體的大小來調整，調大的話，會減少磁碟spill的次數此時如果記憶體足夠的話，一般都會顯著提升效能。spill一般會在Buffer空間大小的80%開始進行spill（因為spill的時候還有可能別的執行緒在往裡寫資料，因為還預留空間，有可能有正在寫到Buffer中的資料），可以通過mapreduce.map.sort.spill.percent（default：0.80）進行調整，Map Task在計算的時候會不斷產生很多spill檔案，在Map Task結束前會對這些spill檔案進行合併，這個過程就是merge的過程。mapreduce.task.io.sort.factor（default：10），代表進行merge的時候最多能同時merge多少spill，如果有100個spill個檔案，此時就無法一次完成整個merge的過程，這個時候需要調大mapreduce.task.io.sort.factor（default：10）來減少merge的次數，從而減少磁碟的操作；
5. 減少磁碟IO和網路IO還可以進行：壓縮，對spill，merge檔案都可以進行壓縮。中間結果非常的大，IO成為瓶頸的時候壓縮就非常有用，可以通過mapreduce.map.output.compress（default：false）設定為true進行壓縮，資料會被壓縮寫入磁碟，讀資料讀的是壓縮資料需要解壓，在實際經驗中Hive在Hadoop的執行的瓶頸一般都是IO而不是CPU，壓縮一般可以10倍的減少IO操作，壓縮的方式Gzip，Lzo,BZip2,Lzma等，其中Lzo是一種比較平衡選擇，mapreduce.map.output.compress.codec（default：org.apache.hadoop.io.compress.DefaultCodec）引數設定。但這個過程會消耗CPU，適合IO瓶頸比較大。
6. 增大記憶體大小
7. copier時可以調節並行度。預設情況下，每個Reducer只會有5個map端並行的下載執行緒在從map下資料，如果一個時間段內job完成的map有100個或者更多，那麼reduce也最多隻能同時下載5個map的資料，所以這個引數比較適合map很多並且完成的比較快的job的情況下調大，有利於reduce更快的獲取屬於自己部分的資料。 在Reducer記憶體和網路都比較好的情況下，可以調大該引數；