package cn.aparke.mr.topn;

import java.io.IOException;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

/**
 * 2017/07/28 qq.com/a
 * 2017/07/28 qq.com/bx
 * 2017/07/28 qq.com/by 需求分析
 * 求出上面每个页面的统计次数前topn
 * 
 * @author aparke
 */
public class PageTopnMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
	Text keyOut = new Text();
	IntWritable valueOut=new IntWritable(1);
	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

		String[] split = value.toString().split(" ");
		keyOut.set(split[1]);
		context.write(keyOut, valueOut);

	}
}

package cn.aparke.mr.topn;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;


//map传来的数据
//输入为一个页面和次数（1），输出为页面和总数
public class PageTopnReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

	Text outKey=new Text();
	IntWritable outValue=new IntWritable();

	//通过TreeMap先拿到每组的key,values 一组一组的拿，当所有数据拿完后，执行clean()，实现输出前几条
	TreeMap<PageTopnBean, Object> treeMap = new TreeMap<>();
	//通过TreeSet先拿到每组的值
	TreeSet<PageTopnBean> treeSet=new TreeSet<>();
	
	@Override
	protected void reduce(Text key, Iterable<IntWritable> values,
			Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
		//reduce拿到统计结果后，先不实现context的write方法
		//将reduce的结果拿到后存到treemap中
		int count=0;
		for (IntWritable value : values) {
			count+=value.get();
		}
		
		//实例化一个pageTopnBean对象，实现compareto方法（倒序）
		PageTopnBean pageTopnBean = new PageTopnBean();
		
		pageTopnBean.set(key.toString(), count);//通过pageTopnBean 实现排序

//		treeMap.put(pageTopnBean, null);//放进treemap中按字典顺序排序
		treeSet.add(pageTopnBean);
	}
	
	@Override
	protected void cleanup (Context context)
			throws IOException, InterruptedException {
		
		//通过context传进来拿到topn的参数
		Configuration conf = context.getConfiguration();
		int topn = conf.getInt("top.n",5);//如果拿不到默认为3
		
		int i=0;
		for (PageTopnBean value : treeSet) {
			outKey.set(value.getPage());
			outValue.set(value.getCount());
			context.write(outKey, outValue);
			i++;
			if (i==topn) {
				return;
			}
		}
		//使用treemap缓存
//		Set<Entry<PageTopnBean, Object>> entrySet = treeMap.entrySet();
//		int i=0;
//		for (Entry<PageTopnBean, Object> entry : entrySet) {
//			
//			outKey.set(entry.getKey().getPage());
//			outValue.set(entry.getKey().getCount());
//			context.write(outKey, outValue);
//			i++;
//			if (i==topn) {
//				return;
//			}
//		}
		
	}
}

package cn.aparke.mr.topn;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class JobSubmitter {

	public static void main(String[] args) throws Exception {

		Configuration conf = new Configuration();
		//conf.addResource("xx-oo.xml");//通过加载classpath下的*-site.xml文件解析参数
		

		conf.setInt("top.n" , 3);// 通过代码设置参数
		//conf.setInt("top.n", Integer.parseInt(args[0]));
		
		//	 通过java反射机制拿到属性配置文件获取参数
		/*
		Properties props = new Properties();
		props.load(JobSubmitter.class.getClassLoader().getResourceAsStream("topn.properties"));
		conf.setInt("top.n", Integer.parseInt(props.getProperty("top.n")));
		*/
		
		Job job = Job.getInstance(conf);

		job.setJarByClass(JobSubmitter.class);

		job.setMapperClass(PageTopnMapper.class);
		job.setReducerClass(PageTopnReducer.class);

		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);

		Path inpath=new Path("d:/testdata/topn/input");
		Path outpath=new Path("d:/testdata/topn/outputnew2");
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outpath)) {
			fs.delete(outpath, true);
		}
		
		FileInputFormat.setInputPaths(job, inpath);
		FileOutputFormat.setOutputPath(job, outpath);
		
		boolean waitForCompletion = job.waitForCompletion(true);
		System.exit(waitForCompletion?0:1);

	}
}

package cn.aparke.mr.topn;

//因为value要求出topn所以需要使用Comparable接口实现compareTo方法
//但是此类为作为输出的key或value所有不需要实现Writable接口实现序列化以及反序列化方法
public class PageTopnBean implements Comparable<PageTopnBean>{

	private String page;
	private int count;
	
	public void set(String page, int count) {
		this.page = page;
		this.count = count;
	}
	public String getPage() {
		return page;
	}
	public void setPage(String page) {
		this.page = page;
	}
	public int getCount() {
		return count;
	}
	public void setCount(int count) {
		this.count = count;
	}
	@Override
	public int compareTo(PageTopnBean o) {
		
		//如果访问次数相等，就比较访问页面
		return o.getCount()-this.count==0?this.page.compareTo(o.getPage()):o.getCount()-this.count;
			}
	
}