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