MapReduce之输出结果排序

MapReduce之输出结果排序

public class Flow implements WritableComparable {

// 上下流量

private long upFlow;

// 下行流量

private long downFlow;

// 总流量

private long sumFlow;

@Override

public int compareTo(Flow o) {

// TODO Auto-generated method stub

return -(int)(this.sumFlow - o.getSumFlow());

}

public long getUpFlow() {

return upFlow;

}

public void setUpFlow(long upFlow) {

this.upFlow = upFlow;

}

public long getDownFlow() {

return downFlow;

}

public void setDownFlow(long downFlow) {

this.downFlow = downFlow;

}

public Flow(long upFlow, long downFlow) {

super();

this.upFlow = upFlow;

this.downFlow = downFlow;

this.sumFlow = upFlow + downFlow;

}

public Flow() {

super();

}

@Override

public void write(DataOutput out) throws IOException {

// TODO Auto-generated method stub

out.writeLong(upFlow);

out.writeLong(downFlow);

out.writeLong(sumFlow);

}

@Override

public String toString() {

return upFlow + “t” + downFlow + “t” + sumFlow;

}

@Override

public void readFields(DataInput in) throws IOException {

// TODO Auto-generated method stub

upFlow = in.readLong();

downFlow = in.readLong();

sumFlow = in.readLong();

}

public long getSumFlow() {

return sumFlow;

}

public void setSumFlow(long sumFlow) {

this.sumFlow = sumFlow;

}

}

2.Map阶段

---

public class FlowCountMap extends Mapper {

@Override

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

// 将一行数据转换为String

String line = value.toString();

// 切分字段

String[] fields = line.split("t");

// 取出手机号

String phoneNum = fields[0];

// 取出上行流量下行流量

long upFlow = Long.parseLong(fields[fields.length - 3]);

long downFlow = Long.parseLong(fields[fields.length - 2]);

Flow flow = new Flow(upFlow, downFlow);

context.write(flow, new Text(phoneNum));

}

}

3.Reduce阶段

---

public class FlowCountReducer extends Reducer{

@Override

protected void reduce(Flow flow, Iterable values, Context context)

throws IOException, InterruptedException {

String phone = values.iterator().next().toString();

// 输出结果

context.write(new Text(phone), flow);

}

}

4.启动类

---

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

Configuration conf = new Configuration(true);

conf.set(“mapreduce.framework.name”, “local”);

// 输出到HDFS文件系统中

// conf.set(“fs.defaultFS”, “hdfs://hadoop-node01:9000”);

// 输出到本地文件系统

conf.set(“fs.defaultFS”, “file:///”);

Job job = Job.getInstance(conf);