Hadoop自定义排序

package myhadoop.mr.sort;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;import org.apache.hadoop.io.WritableComparable;public class InfoBean implements WritableComparable {
private String account;//帐号
private double income;//收入，这里我们不考虑多么精准，因我我们用到的数据都是整数
private double outcome;//支出
private double surplus;//结余

/**
* Text类中专门有个set方法让我们给它赋值，我们模仿Text类，也定义一个set方法来初始化InfoBean
* @param account 帐号
* @param income 收入
* @param outcome 支出
*/
public void set(String account,double income,double outcome){
this.account=account;
this.income=income;
this.outcome=outcome;
this.surplus=income-outcome;
}
/**
* 序列化Bean
*/
public void write(DataOutput out) throws IOException {
out.writeUTF(account);
out.writeDouble(income);
out.writeDouble(outcome);
out.writeDouble(surplus);
}

/**
* 反序列化Bean
*/
public void readFields(DataInput in) throws IOException {
this.account=in.readUTF();
this.income=in.readDouble();
this.outcome=in.readDouble();
this.surplus=in.readDouble();
}
/**
* 自定义比较方法，返回到int值是参数Bean与当前Bean的比较结果
* 如果传过来到Bean的收入值小于当前Bean的收入值，那么应该返回-1;
* 表示传过来的Bean应该排在当前Bean的后面。
* 如果传过来的Bean与当前Bean的收入值一样，那么就再比较支出的值，
* 如果传过来的Bean的支出值小于当前Bean的支出值的话，应该返回1，
* 表示传过来的Bean应该排在当前Bean的前面。
*/
public int compareTo(InfoBean o) {
if(this.income==o.getIncome()){
return this.getOutcome()>o.getOutcome() 1:-1;
}else{
return this.getIncome()>o.getIncome() -1:1;
}
}

/**
* 重写toString方法
*/
@Override
public String toString() {
return this.income+"\t"+this.outcome+"\t"+this.surplus;
}public String getAccount() {
return account;
}public void setAccount(String account) {
this.account = account;
}public double getIncome() {
return income;
}public void setIncome(double income) {
this.income = income;
}public double getOutcome() {
return outcome;
}public void setOutcome(double outcome) {
this.outcome = outcome;
}public double getSurplus() {
return surplus;
}public void setSurplus(double surplus) {
this.surplus = surplus;
}

}
接下来我们写第一个MapReduce，第一个MapReduce用来求和，SumStep类的代码如下：
package myhadoop.mr.sort;
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.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 SumStep {
public static void main(String[] args) throws Exception {
Configuration conf=new Configuration();
Job job=Job.getInstance(conf);
job.setJarByClass(SumStep.class);
job.setMapperClass(SumMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(InfoBean.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));

job.setReducerClass(SumReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(InfoBean.class);
FileOutputFormat.setOutputPath(job, new Path(args[1]));

job.waitForCompletion(true);
}

public static class SumMapper extends Mapper{
private Text k=new Text();
private InfoBean v=new InfoBean();
/**
* 重写map方法
*/
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
//首先获取一行的数据
String line=value.toString();
//由于我们的数据是以空格分割的，因此我们以空格进行分割
String[] fields=line.split("\t");
//获取帐号
String account=fields[0];
//获取收入的值
double income=Double.parseDouble(fields[1]);
//获取支出的值
double outcome=Double.parseDouble(fields[2]);
//给k赋值
k.set(account);
//给v赋值
v.set(account, income, outcome);
//输出
context.write(k, v);
}
}

public static class SumReducer extends Reducer{
private InfoBean v=new InfoBean();
/**
* 重写Reduce方法
*/
@Override
protected void reduce(Text key, Iterable values,Context context)
throws IOException, InterruptedException {
//每个账户总的收入
double sum_income=0;
//每个账户总的支出
double sum_outcome=0;
for(InfoBean bean:values){
sum_income+=bean.getIncome();
sum_outcome+=bean.getOutcome();
}
//由于key就是账户，因此这里不用传account进去也没问题
v.set("", sum_income, sum_outcome);
//输出
context.write(key, v);
}
}
}
接下来我们写第二个MapReduce，第二个MapReduce用来排序，我们起名为SortStep，如下所示：
package myhadoop.mr.sort;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;public class SortStep {

public static void main(String[] args) throws Exception {
Configuration conf=new Configuration();
Job job=Job.getInstance(conf);
job.setJarByClass(SortStep.class);

job.setMapperClass(SortMapper.class);
job.setMapOutputKeyClass(InfoBean.class);
job.setMapOutputValueClass(NullWritable.class);
FileInputFormat.setInputPaths(job, new Path(args[0]));

job.setReducerClass(SortReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(InfoBean.class);
FileOutputFormat.setOutputPath(job, new Path(args[1]));

job.waitForCompletion(true);
}

public static class SortMapper extends Mapper{
private InfoBean k=new InfoBean();
@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
//首先获取一行的数据
String line=value.toString();
//由于我们第一个MapReduce处理完之后toString方法是以\t为分割写的，因此我们这里分割数据应该用\t来分割
String[] fields=line.split("\t");
//获取帐号
String account=fields[0];
//获取收入的值
double income=Double.parseDouble(fields[1]);
//获取支出的值
double outcome=Double.parseDouble(fields[2]);
//给v赋值
k.set(account, income, outcome);
//输出，把Bean当作key它自己就具有排序功能，因此我们只需要Bean就可以了，value的值我们可以用NullWritable来代替
context.write(k, NullWritable.get());
}
}

public static class SortReducer extends Reducer{
private Text k=new Text();
@Override
protected void reduce(InfoBean bean, Iterable values,Context context)
throws IOException, InterruptedException {
String account=bean.getAccount();
k.set(account);
context.write(k, bean);
}
}
}