MapReduce实现TopK的示例

　　由于开始学习MapReduce编程已经有一段时间了，作为一个从编程中寻找自信和乐趣以及热爱编程的孩子来讲，手开始变得很“痒”了，很想小试一下身手。于是自己编写了TopK的代码。TopK的意思就是从原文件中找出词频排名前K的所有单词。首先分析该问题，从中我们可以得到启发：要想知道词频排名前K的所有单词，那么是不是要对所有的单词进行词频的统计啊？于是我们就联想到了一个比较经典的例子：WordCount的例子。是的，没错。就是它，统计原文件中每个单词的个数就靠它。

　　但是，我们词频统计出来了，接下来需要做的就是如何找出词频排名前K的所有单词。如何找出词频排名前K呢？我们知道，WordCount得到的结果就是所有单词的词频情况，并且是已排好序的。所以，我们接下来需要做的是：

1、将所有相同词频的单词汇总，这一步就是map之后的shuffle过程可以得到相应的结果，只需要在map阶段将词频作为key，单词多作为value即可。

2、找出排名前k的词频的所有单词，并且按照词频的顺序排序，在这一步当中，很多人通过采用TreeMap的数据结构来实现，但是这里要注意点，TreeMap对于相同的键值是会进行覆盖的。因此无法操作相同键值的数据。也有些人对key进行封装，但同样还是避免不了有相同键值的结果。因此，我在这里采用的方法是将所有词频相同的单词用ArrayList存放起来，最后在将ArrayList的内容写入待hdfs中即可。

　　综上所述，要实现TopK的结果，需要用到两个MR作业，一个是WordCount作业，一个是TopK的作业。

代码如下：

1、WordCount的部分：

import java.io.IOException;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


public class MyTopK {
	
	public static class  Mymap extends Mapper<LongWritable, Text, Text, IntWritable>{
		
		private final IntWritable one =new IntWritable(1);
		private Text word =new Text();
		
		public void map(LongWritable ikey,Text ivalue,Context context) throws IOException, InterruptedException{
			StringTokenizer str=new StringTokenizer(ivalue.toString());
			while(str.hasMoreTokens()){
				word.set(str.nextToken());
				context.write(word, one);
			}
		}
	}
	
	
	public static class Myreduce extends Reducer<Text, IntWritable, Text, IntWritable>{
		private IntWritable result=new IntWritable();
		
		
		public void reduce(Text ikey,Iterable<IntWritable> ivalue,
				Context context) throws IOException, InterruptedException{
			int sum=0;
			for(IntWritable val:ivalue){
				sum+=val.get();
			}
			result.set(sum);
			context.write(ikey, result);
		}
	}
	
	//设置静态的函数，方便直接在main中通过类名 来调用
	public static boolean run(String in ,String out) throws IOException, ClassNotFoundException, InterruptedException{
		
		Configuration conf =new Configuration();
		
		Job job=new Job(conf,"Wordcount");
		job.setJarByClass(MyTopK.class);
		job.setMapperClass(Mymap.class);
		job.setReducerClass(Myreduce.class);
		
		//设置map输出类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		//设置reduce的输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		//设置输入输出路径
		FileInputFormat.addInputPath(job, new Path(in));
		FileOutputFormat.setOutputPath(job, new Path(out));
		
		return job.waitForCompletion(true);
	}
}

　　2、TopK的实现过程

import java.io.IOException;
import java.util.Map.Entry;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Set;
import java.util.StringTokenizer;
import java.util.TreeMap;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
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.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;


public class MyTopK1 {
	
	public static class MyMap extends Mapper<LongWritable, Text, IntWritable, Text>{
		
		IntWritable outkey=new IntWritable();
		Text outvalue=new Text();
		
		public void map(LongWritable ikey,Text ivalue,Context context) throws IOException, InterruptedException{
			StringTokenizer str=new StringTokenizer(ivalue.toString());
			while(str.hasMoreTokens()){
				//这个表示是输入数据的每行数据，每一行包含了单词和单词的次数，这个内容否在ivalue中，下面需要将ivalue中的单词和单词的次数进行分离。
				String element=str.nextToken();
				if(Pattern.matches("\d+", element)){//这里利用正则表达式来匹配单词的个数
					outkey.set(Integer.parseInt(element));//将单词的个数作为键值
				}else {
					outvalue.set(element);//将单词作为键值值
				}
			}
			context.write(outkey, outvalue);//在写的过程中会对单词的次数进行排序
		}
		
	}
	
	public  static TreeMap<Integer, ArrayList<String> > hm =new TreeMap<Integer, ArrayList<String> >(new Comparator<Integer>() {
		public int compare(Integer v1,Integer v2){
			return v2.compareTo(v1);
		}
	});//用来选择出topK
	private static MultipleOutputs<Text, IntWritable> mos=null;//用来进行多文件输出
	
	private static String path=null;
	
	//通过shuffle过程之后，相同次数的单词就在一起了，并将这个数据作为reduce的输入数据
	public static class Myreduce extends Reducer<IntWritable, Text, Text, IntWritable>{
		
		public void reduce(IntWritable ikey,Iterable<Text> ivalue,Context context) throws IOException, 
		InterruptedException{
			ArrayList<String> tmp=new ArrayList<String>(10);
			for(Text val:ivalue){
				context.write(val,ikey);//输出全排序的内容
//				tmp.add(val.toString());  //这里会造成占用较多的内存，这里可以优化
				
				//优化的方法就是限定列表的长度，由于是topk，所以每一个最多也就是10个即可
				if(tmp.size()<=10){
					tmp.add(val.toString());  
				}
			}
			hm.put(ikey.get(), tmp);
		}
		
		private static int topKNUM=10; //表示求最高的多少个数
		protected void cleanup(Context context) throws IOException,
				InterruptedException {
			//String path = context.getConfiguration().get("topKout");
			mos = new MultipleOutputs<Text, IntWritable>(context);
			Set<Entry<Integer, ArrayList<String> > > set =   hm.entrySet();
			
			for (Entry<Integer, ArrayList<String>> entry : set) {
				ArrayList<String> al = entry.getValue();
				if (topKNUM-al.size() > 0) {
					for (String word : al) {
						//if (topKNUM-- > 0) {
							mos.write("topKMOS", new Text(word), // 这里参数“topKMOS”表示一个属性名称
									new IntWritable(entry.getKey()), path);
						//}
					}
				}
			}
			mos.close();
		}

	}
	@SuppressWarnings("deprecation")
	public static  void run(String in,String out,String topkout) throws IOException, ClassNotFoundException, InterruptedException{
		
		Configuration conf=new Configuration();
		
		//创建作业，并制定map和reduce类

		Job job=new Job(conf);
		job.setJarByClass(MyTopK1.class);
		job.setMapperClass(MyMap.class);
		job.setReducerClass(Myreduce.class);
		
		//TopK的输出路径
		path=topkout;
		//conf.set("topKout",topkout);
		
		//设置map输出的类型
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(Text.class);
		
		//设置reduce的输出类型
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		//设置MultipleOutputs输出格式，//这里的第二个参数“topKMOS”要跟write方法中的参数相同
		MultipleOutputs.addNamedOutput(job, "topKMOS",TextOutputFormat.class, Text.class, IntWritable.class);
		
		//设置输入输出格式
		FileInputFormat.addInputPath(job, new Path(in));
		FileOutputFormat.setOutputPath(job, new Path(out));
		
		//提交作业
		job.waitForCompletion(true);
	}
}

　　3、程序的入口

import java.io.IOException;

import org.apache.log4j.PropertyConfigurator;


public class TopKmain {

	public static void main(String[] args) throws ClassNotFoundException, IOException, InterruptedException {
		// TODO Auto-generated method stub、、
		//这里要记住一点，手动加载一些log4j文件，一来可以去掉警告，二来可以在出现错误时，通过查看日志，了解错误详细内容
		String rootPath = System.getProperty("user.dir" );
		PropertyConfigurator.configure(rootPath+"\log4j.properties");
		
		//要统计字数，排序的文字
        String in = "hdfs://192.168.1.21:9000/input";
        
        //统计字数后的结果
        String wordCoutoutput = "hdfs://192.168.1.21:9000/out";
        
        //对统计完后的结果再排序后的内容
        String sort = "hdfs://192.168.1.21:9000/sort";
        
        //指定前K条输出的文件名称
        String topK = "hdfs://192.168.1.21:9000/topK";
        
        //如果统计字数的job完成后就开始排序
        if(MyTopK.run(in, wordCoutoutput)){
        	MyTopK1.run(wordCoutoutput, sort,topK);
        } 
	}

}