在最初接觸mapreduce時(shí)，top n 問(wèn)題的解決辦法是將mapreduce輸出（排序后）放入一個(gè)集合中，取前n個(gè)，但這種寫法過(guò)于簡(jiǎn)單，內(nèi)存能夠加載的集合的大小是有上限的，一旦數(shù)據(jù)量大，很容易出現(xiàn)內(nèi)存溢出。

今天在這里介紹另一種實(shí)現(xiàn)方式，當(dāng)然這也不是***的方式，不過(guò)正所謂一步一個(gè)腳印，邁好每一步，以后的步伐才能更堅(jiān)定，哈哈說(shuō)了點(diǎn)題外話。恩恩，以后還會(huì)有更好的方式需求，得到top ***的前n條記錄。

這里只給出一些核心的代碼，其他job等配置的代碼略 。

Configuration conf = new Configuration(); 
conf.setInt("N", 5); 

初始化job之前需要 conf.setInt("N",5); 意在在mapreduce階段讀取N,N就代表著top N。

以下是map

package com.lzz.one; 
import java.io.IOException; 
import java.util.Arrays; 
import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.io.LongWritable; 
import org.apache.hadoop.io.Text; 
import org.apache.hadoop.mapreduce.Mapper; 
  
  
/** 
 * topN 
*  #orderid,userid,payment,productid 
* [root@x00 hd]# cat seventeen_a.txt 
* 1,9819,100,121 
* 2,8918,2000,111 
* 3,2813,1234,22 
* 4,9100,10,1101 
* 5,3210,490,111 
* 6,1298,28,1211 
* 7,1010,281,90 
* 8,1818,9000,20 
* [root@x00 hd]# cat seventeen_b.txt 
* 100,3333,10,100 
* 101,9321,1000,293 
* 102,3881,701,20 
* 103,6791,910,30 
* 104,8888,11,39 
   
* 預(yù)測(cè)結(jié)果：（求 Top N=5 的結(jié)果） 
* 1 9000 
* 2 2000 
* 3 1234 
* 4 1000 
* 5 910 
 * @author Administrator 
 * 
 */ 
public class TopNMapper extends Mapper<LongWritable, Text, IntWritable, IntWritable>{ 
    int len; 
    int top[]; 
    @Override 
    public void setup(Context context) throws IOException,InterruptedException { 
        len = context.getConfiguration().getInt("N", 10); 
        top = new int[len+1]; 
    } 
   
    @Override 
public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException { 
    String line = value.toString(); 
    String arr []= line.split(","); 
    if(arr != null && arr.length == 4){ 
        int pay = Integer.parseInt(arr[2]); 
        add(pay); 
    } 
} 
  
  
public void add(int pay){ 
    top[0] = pay; 
    Arrays.sort(top); 
} 
   
@Override 
public void cleanup(Context context) throws IOException,InterruptedException { 
    for(int i=1;i<=len;i++){ 
        <span></span>context.write(new IntWritable(top[i]),new IntWritable(top[i])); 
    <span></span>} 
 } 
   
} 
  
   
   
   
   
 <div> 
  
  
   
   
   
   
 </div> 

接下來(lái)是reduce

package com.lzz.one; 
 
import java.io.IOException; 
import java.util.Arrays; 
 
import org.apache.hadoop.io.IntWritable; 
import org.apache.hadoop.mapreduce.Reducer; 
 
public class TopNReduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{ 
int len; 
int top[]; 
@Override 
public void setup(Context context) 
throws IOException, InterruptedException { 
len = context.getConfiguration().getInt("N", 10); 
top = new int[len+1]; 
} 
 
@Override 
public void reduce(IntWritable key, Iterable<IntWritable> values, 
Context context) 
throws IOException, InterruptedException { 
for(IntWritable val : values){ 
add(val.get()); 
} 
} 
 
public void add(int pay){ 
top[0] = pay; 
Arrays.sort(top); 
} 
 
@Override 
public void cleanup(Context context) 
throws IOException, InterruptedException { 
for(int i=len;i>0;i--){ 
context.write(new IntWritable(len-i+1),new IntWritable(top[i])); 
} 
} 
} 

說(shuō)一下邏輯，雖然畫圖比較清晰，但是時(shí)間有限，畫圖水平有限，只用語(yǔ)言來(lái)描述吧，希望能說(shuō)的明白。

如果要取top 5，則應(yīng)該定義一個(gè)長(zhǎng)度為為6的數(shù)組，map所要做的事情就是將每條日志的那個(gè)需要排序的字段放入數(shù)組***個(gè)元素中，調(diào)用Arrays.sort(Array[])方法可以將數(shù)組按照正序，從數(shù)字角度說(shuō)是從小到大排序，比如***條記錄是9000，那么排序結(jié)果是[0,0,0,0,0,9000]，第二條日志記錄是8000，排序結(jié)果是[0,0,0,0,8000,9000]，第三條日志記錄是8500，排序結(jié)果是[0,0,0,8000,8500,9000]，以此類推，每次放進(jìn)去一個(gè)數(shù)字如果大于數(shù)組里面最小的元素，相當(dāng)于將最小的覆蓋掉了，也就是說(shuō)數(shù)組中元素永遠(yuǎn)是拿到日志中***的那些個(gè)記錄。

ok，map將數(shù)組原封不動(dòng)按照順序輸出，reduce接收到從每個(gè)map拿到的五個(gè)排好序的元素，在進(jìn)行跟map一樣的排序，排序后數(shù)組里面就是按照從小到大排好序的元素，將這些元素倒序輸出就是最終我們要的結(jié)果了。

與之前的方式做個(gè)比較，之前的map做的事情很少，在reduce中排序后哪前5條，reduce的壓力是很大的，要把所有的數(shù)據(jù)都處理一遍，而一般設(shè)置reduce的個(gè)數(shù)較少，一旦數(shù)據(jù)較多，reduce就會(huì)承受不了，悲劇了。而現(xiàn)在的方式巧妙的將reduce的壓力轉(zhuǎn)移到了map，而map是集群效應(yīng)的，很多臺(tái)服務(wù)器來(lái)做這件事情，減少了一臺(tái)機(jī)器上的負(fù)擔(dān)，每個(gè)map其實(shí)只是輸出了5個(gè)元素而已，如果有5個(gè)map，其實(shí)reduce才對(duì)5*5個(gè)數(shù)據(jù)進(jìn)行了操作，也就不會(huì)出現(xiàn)內(nèi)存溢出等問(wèn)題了。

原文出自：http://my.oschina.net/u/1378204/blog/343666