測試hadoop版本：2.4 

Map端聚合的應用場景：當我們只關心所有資料中的部分資料時，並且資料可以放入記憶體中。

使用的好處：可以大大減小網路資料的傳輸量，提高效率；

一般程式設計思路：在Mapper的map函式中讀入所有資料，然後新增到一個List（佇列）中，然後在cleanup函式中對list進行處理，輸出我們關係的少量資料。

例項：

在map函式中使用空格分隔每行資料，然後把每個單詞新增到一個堆疊中，在cleanup函式中輸出堆疊中單詞次數比較多的單詞以及次數；

package fz.inmap.aggregation;

import java.io.IOException;
import java.util.ArrayList;
import java.util.PriorityQueue;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


public class InMapArrgegationDriver extends Configured implements Tool{
	public static Logger log = LoggerFactory.getLogger(InMapArrgegationDriver.class);
	/**
	 * @throws Exception 
	 * 
	 */
	public static void main(String[] args) throws Exception {
		ToolRunner.run(new Configuration(), new InMapArrgegationDriver(),args);
	}

	@Override
	public int run(String[] arg0) throws Exception {
		if(arg0.length!=3){
			System.err.println("Usage:\nfz.inmap.aggregation.InMapArrgegationDriver <in> <out> <maxNum>");
			return -1;
		}
		Configuration conf = getConf();
		
//		System.out.println(conf.get("fs.defaultFS"));
		Path in = new Path(arg0[0]);
		Path out= new Path(arg0[1]);
		out.getFileSystem(conf).delete(out, true);
		conf.set("maxResult", arg0[2]);
		Job job = Job.getInstance(conf,"in map arrgegation job");
		job.setJarByClass(getClass());
		
		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);
		
		job.setMapperClass(InMapMapper.class);
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
//		job.setOutputKeyClass(LongWritable.class);
//		job.setOutputValueClass(VectorWritable.class);
		job.setNumReduceTasks(0);
//		System.out.println(job.getConfiguration().get("mapreduce.job.reduces"));
//		System.out.println(conf.get("mapreduce.job.reduces"));
		FileInputFormat.setInputPaths(job, in);
		FileOutputFormat.setOutputPath(job, out);
		
		return job.waitForCompletion(true)?0:-1;
	}
	
	protected static class InMapMapper extends Mapper<LongWritable,Text,Text,IntWritable>{
		private ArrayList<Word> words = new ArrayList<Word>();
		private PriorityQueue<Word> queue;
		private int maxResult;
		
		protected void setup(Context cxt){
			maxResult = cxt.getConfiguration().getInt("maxResult", 10);
		}
		
		protected void map(LongWritable key, Text value,Context cxt){
			String  [] line = value.toString().split(" "); // use blank to split
			for(String word:line){
				Word curr = new Word(word,1);
				if(words.contains(curr)){
					// increase the exists word's frequency
					for(Word w:words){
						if(w.equals(curr)){
							w.frequency++;
							break;
						}
					}
				}else{
					words.add(curr);
				}
			}
		}
		protected void cleanup(Context cxt) throws InterruptedException,IOException{
			Text outputKey = new Text();
			IntWritable outputValue = new IntWritable();
			
			queue = new PriorityQueue<Word>(words.size());
			queue.addAll(words);
			for(int i=0;i< maxResult;i++){
				Word tail = queue.poll();
				if(tail!=null){
					outputKey.set(tail.value);
					outputValue.set(tail.frequency);
					log.info("key is {},value is {}", outputKey,outputValue);
					cxt.write(outputKey, outputValue);
					
				}
			}
		}
	}

}
 

package fz.inmap.aggregation;

public class Word implements Comparable<Word>{

	public String value;
	public int frequency;
	
	public Word(String value,int frequency){
		this.value=value;
		this.frequency=frequency;
	}
	@Override
	public int compareTo(Word o) {
		return o.frequency-this.frequency;
	}
	@Override
	public boolean equals(Object obj){
		if(obj instanceof Word){
			return value.equalsIgnoreCase(((Word)obj).value);
		}else{
			return false;
		}
	}
}
 

或者檢視輸出結果：

總結：使用map端聚合，雖然可以大大減小網路資料傳輸量，提高效率，但是我們在應用的時候還是需要考慮實際的應用環境。比如，如果使用上面的演算法來計算最大單詞頻率的前10個，然後還是使用上面的程式碼，就會有問題。每個mapper會處理並輸出自己的單詞詞頻最大的10個單詞，並沒有考慮到所有資料，這樣在reducer端整合的時候就會可能會忽略部分資料，造成最終結果的錯誤。

分享，成長，快樂