MapReduce- 数据的排序处理

MapReduce- 数据的排序处理

package com.huhu.day02;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

/**

 * 6 9

 * 3 8

 * 4 8

 * 1 0

 * 3 0

 * 8 8

 * 6 7

 * 第一列升序，第二列降序

 * @author huhu_k

 *

 */

public class Number implements WritableComparable<Number> {

	private int first;

	private int second;

	// private int third;

	public Number() {

		super();

	}

	public Number(int first, int second) {

		super();

		this.first = first;

		this.second = second;

	}

	public int getFirst() {

		return first;

	}

	public void setFirst(int first) {

		this.first = first;

	}

	public int getSecond() {

		return second;

	}

	public void setSecond(int second) {

		this.second = second;

	}

	@Override

	public int hashCode() {

		final int prime = 31;

		int result = 1;

		result = prime * result + first;

		result = prime * result + second;

		return result;

	}

	@Override

	public boolean equals(Object obj) {

		if (this == obj)

			return true;

		if (obj == null)

			return false;

		if (getClass() != obj.getClass())

			return false;

		Number other = (Number) obj;

		if (first != other.first)

			return false;

		if (second != other.second)

			return false;

		return true;

	}

	@Override

	public String toString() {

		return "Number [first=" + first + ", second=" + second + "]";

	}

	@Override

	public void readFields(DataInput in) throws IOException {

		this.first = in.readInt();

		this.second = in.readInt();

	}

	@Override

	public void write(DataOutput out) throws IOException {

		out.writeInt(this.first);

		out.writeInt(this.second);

	}

	@Override

	public int compareTo(Number o) {

		if (this.first== o.first) {

			//第二行数据降序

			return o.second - this.second;

		}

		//第一行升序

		return this.first - o.first;

	}

}

package com.huhu.day02;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;

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;

import org.apache.hadoop.util.GenericOptionsParser;

import org.apache.hadoop.util.Tool;

import org.apache.hadoop.util.ToolRunner;

public class NumericSorting extends ToolRunner implements Tool {

	public static class MyMapper extends Mapper<LongWritable, Text, Number, NullWritable> {

		@Override

		protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

			String[] line = value.toString().split(" ");

			Number number = null;

			if (line.length == 2) {

				number = new Number(Integer.parseInt(line[0]), Integer.parseInt(line[1]));

			}

			context.write(number, NullWritable.get());

		}

	}

	public static class MyReduce extends Reducer<Number, NullWritable, Number, Text> {

		@Override

		protected void reduce(Number key, Iterable<NullWritable> values, Context context)

				throws IOException, InterruptedException {

			for (NullWritable n : values) {

				context.write(key, new Text("---"));

			}

		}

	}

	@Override

	public Configuration getConf() {

		return new Configuration();

	}

	@Override

	public void setConf(Configuration arg0) {

	}

	@Override

	public int run(String[] other) throws Exception {

		Job job = Job.getInstance(getConf(), "NumbericSorting");

		job.setJarByClass(NumericSorting.class);

		job.setMapperClass(MyMapper.class);

		job.setMapOutputKeyClass(Number.class);

		job.setMapOutputValueClass(NullWritable.class);

		job.setReducerClass(MyReduce.class);

		job.setOutputKeyClass(Number.class);

		job.setOutputValueClass(Text.class);

		FileInputFormat.addInputPath(job, new Path(other[0]));

		FileOutputFormat.setOutputPath(job, new Path(other[1]));

		return job.waitForCompletion(true) ? 0 : 1;

	}

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

		Configuration conf = new Configuration();

		String[] other = new GenericOptionsParser(conf, args).getRemainingArgs();

		if (other.length != 2) {

			System.out.println("your input args number is fail,you need input <in> and <out>");

			System.exit(0);

		}

		ToolRunner.run(conf, new NumericSorting(), other);

	}

}

运行结果：