package cn.aparke.mr.order.topn.grouping;

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

//实现思路三个关键点
//1.用bean做key,相同的的orderId排在一起，id小的排在前面，大的排在后面
//2.重写数据分发规则Partitioner 让orderId相同的被分给同一个reduce task
//3.重写Grouping Comparator只要orderId相同就 会被看成同一组进行一次reduce聚合
public class OrderTopn {

	public static class OrderTopnMapper extends Mapper<LongWritable, Text, OrderBean, NullWritable> {
		OrderBean orderBean = new OrderBean();
		NullWritable v = NullWritable.get();

		@Override
		protected void map(LongWritable key, Text value,
				Mapper<LongWritable, Text, OrderBean, NullWritable>.Context context)
				throws IOException, InterruptedException {
			String[] filds = value.toString().split(",");
			orderBean.set(filds[0], filds[1], filds[2], Float.parseFloat(filds[3]), Integer.parseInt(filds[4]));
			String k = filds[0];

			context.write(orderBean, v);
		}
	}

	public static class OrderTopnReducer extends Reducer<OrderBean, NullWritable, OrderBean, NullWritable> {

		@Override
		protected void reduce(OrderBean key, Iterable<NullWritable> values,
				Reducer<OrderBean, NullWritable, OrderBean, NullWritable>.Context context)
				throws IOException, InterruptedException {
			
			int topn = context.getConfiguration().getInt("top.n", 3);
			int i = 0;
			for (NullWritable v : values) {
				context.write(key, v);
				if(++i==topn) return;
			}

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

		
		Configuration conf = new Configuration(); // 默认只加载core-default.xml core-site.xml
		conf.setInt("top.n",3);
		
		Job job = Job.getInstance(conf);

		job.setJarByClass(OrderTopn.class);

		job.setMapperClass(OrderTopnMapper.class);
		job.setReducerClass(OrderTopnReducer.class);
		
		job.setPartitionerClass(OrderIdPartitioner.class);
		job.setGroupingComparatorClass(OrderIdGroupingComparator.class);
		
		job.setNumReduceTasks(3);

		job.setMapOutputKeyClass(OrderBean.class);
		job.setMapOutputValueClass(NullWritable.class);
		
		job.setOutputKeyClass(OrderBean.class);
		job.setOutputValueClass(NullWritable.class);
		
		Path inPath  =new Path("D:\\testdata\\order\\input");
		Path outPath  = new Path("D:\\testdata\\order\\outputGrouping");
		FileSystem fs = outPath.getFileSystem(conf);
		if(fs.exists(outPath))
			fs.delete(outPath,true);

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

		job.waitForCompletion(true);
	}
	
}

package cn.aparke.mr.order.topn.grouping;

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

import org.apache.hadoop.io.WritableComparable;

public class OrderBean implements WritableComparable<OrderBean> {

	private String orderId;
	private String userId;
	private String pdtName;
	private float price;
	private int number;
	private float amountFee;// 销售总金额

	// 此方法用来设置对象的参数
	public void set(String orderId, String userId, String pdtName, float price, int number) {
		this.orderId = orderId;
		this.userId = userId;
		this.pdtName = pdtName;
		this.price = price;
		this.number = number;
		this.amountFee = number * price;// 销售金额等于数量*单价
	}

	public String getOrderId() {
		return orderId;
	}

	public void setOrderId(String orderId) {
		this.orderId = orderId;
	}

	public String getUserId() {
		return userId;
	}

	public void setUserId(String userId) {
		this.userId = userId;
	}

	public String getPdtName() {
		return pdtName;
	}

	public void setPdtName(String pdtName) {
		this.pdtName = pdtName;
	}

	public float getPrice() {
		return price;
	}

	public void setPrice(float price) {
		this.price = price;
	}

	public int getNumber() {
		return number;
	}

	public void setNumber(int number) {
		this.number = number;
	}

	public float getAmountFee() {
		return amountFee;
	}

	public void setAmountFee(float amountFee) {
		this.amountFee = amountFee;
	}

	@Override
	public String toString() {
		return this.orderId + "," + this.userId + "," + this.pdtName + "," + this.price + "," + this.number + ","
				+ this.amountFee;

	}

	// 序列化
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeUTF(this.orderId);
		out.writeUTF(this.userId);
		out.writeUTF(this.pdtName);
		out.writeFloat(this.price);
		out.writeInt(this.number);
	}

	// 反序列化
	@Override
	public void readFields(DataInput in) throws IOException {
		this.orderId = in.readUTF();
		this.userId = in.readUTF();
		this.pdtName = in.readUTF();
		this.price = in.readFloat();
		this.number = in.readInt();
		this.amountFee = this.price * this.number;// 序列化中虽然没有序列化出，但是反序列化可以通过逻辑实现进去
	}

	// 比较规则，先比较orderId，次之比较总金额
	@Override
	public int compareTo(OrderBean o) {
		//三元表达式实现比较规则
		return this.orderId.compareTo(o.getOrderId())==0?Float.compare(o.getAmountFee(), this.getAmountFee()):this.orderId.compareTo(o.getOrderId());
		
	}
}

package cn.aparke.mr.order.topn.grouping;

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

public class OrderIdGroupingComparator extends WritableComparator{

	//指定反序列化那个类
	//反序列化为一个对象类型，此时只知道对象为WritableComparable 不知道是OrderBean 需要 告诉它比的对象是什么类型 ，需要用构造函数来实现
	//调用父类构造函数
public OrderIdGroupingComparator() {//通过构造加载
	super(OrderBean.class,true);//用反射，加载要分组的类  true为是否要创建示例 选择true
}

@Override
public int compare(WritableComparable a, WritableComparable b) {
	
	
	//首先保证一点：让ordId相同的数据发送给相同的reduce task  修改partitioner 分发规则 ，只要orderid相同就返回同一个reduce Task
	//设法让reduce task 程序认为 orderId相同的key（orderBean），属于同一组， 修改Grouping Comparator的逻辑实现
	OrderBean orderBean_a=(OrderBean)a;
	OrderBean orderBean_b=(OrderBean)b;
	//使订单id相同的连续的排在一起
	return orderBean_a.getOrderId().compareTo(orderBean_b.getOrderId());
}
}

package cn.aparke.mr.order.topn.grouping;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Partitioner;

public class OrderIdPartitioner extends Partitioner<OrderBean, NullWritable> {

	@Override
	public int getPartition(OrderBean key, NullWritable value, int numPartitions) {

		// 按照订单中的orderId来分发数据，控制相同的key被分到同一个reduceTask，否则就只能设置一个分区
		return (key.getOrderId().hashCode() & Integer.MAX_VALUE) % numPartitions;
	}
}