package cn.aparke.mr.page.reverse;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;


public class PageCountBean implements WritableComparable<PageCountBean> {

	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;
	}

	// 定义一个比较器，比较次数。两个次数做差如果等于0，如果不等于0返回他们的页面大小，否则返回他们的差
	public int compareTo(PageCountBean o) {

		return o.getCount() - this.count == 0 ? this.page.compareTo(o.getPage()) : o.getCount() - this.count;
	}

	public void readFields(DataInput in) throws IOException {
		this.page = in.readUTF();
		this.count = in.readInt();
	}

	public void write(DataOutput out) throws IOException {
		out.writeUTF(this.page);
		out.writeInt(this.count);
	}

	@Override
	public String toString() {
		return this.page + "," + this.count;
	}

}

package cn.aparke.mr.page.reverse;

import java.io.IOException;
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.LongWritable;
import org.apache.hadoop.io.NullWritable;
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;


/*需求
mapreduce实现 页面访问总次数的全局倒序排序
通过两次的mapreduce实现
1.先同通过wordcount统计出每个页面的访问次数
2.将上次的结果作为这次的输入统计次数并实现排序*/

/*
mapreduce程序内置了一个排序机制：
map worker 和reduce worker ，都会对数据按照key的大小来排序
所以最终的输出结果中，一定是按照key有顺序的结果

思路：
本案例中，就可以利用这个机制来实现需求：
1、先写一个mr程序，将每个页面的访问总次数统计出来
2、再写第二个mr程序：
map阶段： 读取第一个mr产生的结果文件，将每一条数据解析成一个java对象PageCountBean（封装着一个url和它的总次数），然后将这个对象作为key，null作为value返回
要点：这个java对象要实现WritableComparable接口，以让worker可以调用对象的compareTo方法来进行排序

reduce阶段：由于worker已经对收到的数据按照PageCountBean的compareTo方法排了序，所以，在reduce方法中，只要将数据输出即可，最后的结果自然是按总次数大小的有序结果

*/
public class PageCountReverseSort {

	public static class PageCountStepMapStep1 extends Mapper<LongWritable, Text, Text, IntWritable> {

		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] split = line.split(" ");
			context.write(new Text(split[1]), new IntWritable(1));
		}
	}

	public static class PageCountStepReduceStep1 extends Reducer<Text, IntWritable, Text, IntWritable> {

		IntWritable outVal=new IntWritable();
		@Override
		protected void reduce(Text key, Iterable<IntWritable> values,
				Reducer<Text, IntWritable, Text, IntWritable>.Context context)
				throws IOException, InterruptedException {
			int count = 0;
			for (IntWritable value : values) {
				count += value.get();
			}
			outVal.set(count);
			context.write(key, outVal);
		}
	}


	public static class PageCountStep2Mapper extends Mapper<LongWritable, Text, PageCountBean, NullWritable> {

		@Override
		protected void map(LongWritable key, Text value,
				Mapper<LongWritable, Text, PageCountBean, NullWritable>.Context context)
				throws IOException, InterruptedException {

			String[] split = value.toString().split("\t");
			PageCountBean pageCount = new PageCountBean();
			pageCount.set(split[0], Integer.parseInt(split[1]));
			context.write(pageCount, NullWritable.get());
		}
	}
	
	public static class PageCountStep2Reduce extends Reducer<PageCountBean, NullWritable, PageCountBean, NullWritable>{
		
		@Override
		protected void reduce(PageCountBean key, Iterable<NullWritable> value,
				Reducer<PageCountBean, NullWritable, PageCountBean, NullWritable>.Context context)
				throws IOException, InterruptedException {

			context.write(key, NullWritable.get());
			
		}
			}
	
	
	
	public static void main(String[] args) throws Exception {
		
		Configuration conf =new Configuration();
		conf.setInt("top.n", 5);
		Job job =Job.getInstance(conf);
		
		job.setJarByClass(PageCountReverseSort.class);

		job.setMapperClass(PageCountStepMapStep1.class);
		job.setReducerClass(PageCountStepReduceStep1.class);

		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		Path inPath = new Path("D:\\testdata\\topn\\input");
		Path outPath = new Path("D:\\testdata\\topn\\count");
		
		FileSystem fs = FileSystem.get(conf);
		if (fs.exists(outPath)) {
			fs.delete(outPath, true);
		}

		FileInputFormat.setInputPaths(job,inPath);
		FileOutputFormat.setOutputPath(job,outPath);

//		Job运行是通过job.waitForCompletion(true)，true表示将运行进度等信息及时输出给用户，false的话只是等待作业结束
		job.waitForCompletion(true);
		
		//第二个job
		Job job2 = Job.getInstance(conf);
		job2.setJarByClass(PageCountReverseSort.class);
		job2.setMapperClass(PageCountStep2Mapper.class);
		job2.setReducerClass(PageCountStep2Reduce.class);
		job2.setMapOutputKeyClass(PageCountBean.class);
		job2.setMapOutputValueClass(NullWritable.class);
		
		//job2的输入是job1的输出
		Path sortOut = new Path("D:\\testdata\\topn\\sortput");
		if (fs.exists(sortOut)) {
			fs.delete(sortOut, true);
		}
		
		FileInputFormat.setInputPaths(job2, outPath);
		FileOutputFormat.setOutputPath(job2, sortOut);
		job2.setNumReduceTasks(1);
		boolean waitForCompletion = job2.waitForCompletion(true);
		System.exit(waitForCompletion?0:1);
	}
}