mapreduce wc单词计数 自定义分区 自定义排序实现

package com.ellis.mr3; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; import org.apache.hadoop.io.Writable; import org.apache.hadoop.io.WritableComparable; /** * 自定义排序需要根据上一次结果进行排序，上一次结果的reduce输入为一个文件，不然只能做到局部排序。 * * @author wxz * */ public class FlowBean implements WritableComparable<FlowBean>{ private long upFlow; private long dFlow; private long sumFlow; public FlowBean(){} public FlowBean(long upFlow, long dFlow) { this.upFlow = upFlow; this.dFlow = dFlow; this.sumFlow = upFlow + dFlow; } public void set(long upFlow, long dFlow,long sumFlow) { this.upFlow = upFlow; this.dFlow = dFlow; this.sumFlow = sumFlow; } /** * 序列化 * @param out * @throws IOException */ @Override public void write(DataOutput out) throws IOException { out.writeLong(upFlow); out.writeLong(dFlow); out.writeLong(sumFlow); } /** * 反序列化 * @param in * @throws IOException */ @Override public void readFields(DataInput in) throws IOException { upFlow = in.readLong(); dFlow = in.readLong(); sumFlow = in.readLong(); } /** * 自定义排序，根据总流量排序 */ @Override public int compareTo(FlowBean o) { return this.sumFlow > o.getSumFlow() ? 1 : -1; } public long getUpFlow() { return upFlow; } public void setUpFlow(long upFlow) { this.upFlow = upFlow; } public long getdFlow() { return dFlow; } public void setdFlow(long dFlow) { this.dFlow = dFlow; } public long getSumFlow() { return sumFlow; } public void setSumFlow(long sumFlow) { this.sumFlow = sumFlow; } @Override public String toString() { return "" + upFlow + "\t" + dFlow + "\t" + sumFlow; } }