长沙专业个人做网站哪家好,网站出现wordpress,安徽住房和城乡建设部网站官网,dw做企业网站一、场景描述
以统计号码的流量案例为基础#xff0c;进行开发。
流量统计结果 我们现在要对这个数据的总流量进行自定义排序。
二、代码实现
我们要对总流量进行排序#xff0c;就是对FlowBean中的sumFlow字段进行排序。 所以#xff0c;我们需要让FlowBean实现Writab…一、场景描述
以统计号码的流量案例为基础进行开发。
流量统计结果 我们现在要对这个数据的总流量进行自定义排序。
二、代码实现
我们要对总流量进行排序就是对FlowBean中的sumFlow字段进行排序。 所以我们需要让FlowBean实现WritableComparable接口并重写compareTo方法。
另外我们知道排序是在Shuffle过程进行的且是在环形缓冲区进行的排序 此处的排序采用快速排序算法针对key的索引进行排序按照字典顺序进行排序。
所以我们需要在mapper程序中把FlowBean设置成key这样Shuffle阶段会调用FlowBean的compareTo方法进行排序。
FlowBean.java
package com.atguigu.mapreduce.writableComparable;import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;/*** 1、定义类实现writable接口* 2、重写序列化和反序列化方法* 3、重写空参构造* 4、toString方法*/
public class FlowBean implements WritableComparableFlowBean {private long upFlow; // 上行流量private long downFlow; // 下行流量private long sumFlow; // 总流量// 空参构造public FlowBean() {}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 long getSumFlow() {return sumFlow;}public void setSumFlow(long sumFlow) {this.sumFlow sumFlow;}public void setSumFlow() {this.sumFlow this.upFlow this.downFlow;}Overridepublic void write(DataOutput out) throws IOException {out.writeLong(upFlow);out.writeLong(downFlow);out.writeLong(sumFlow);}Overridepublic void readFields(DataInput in) throws IOException {this.upFlow in.readLong();this.downFlow in.readLong();this.sumFlow in.readLong();}Overridepublic String toString() {return upFlow \t downFlow \t sumFlow;}Overridepublic int compareTo(FlowBean o) {// 总流量的倒序排序if (this.sumFlow o.sumFlow) {return -1;} else if (this.sumFlow o.sumFlow) {return 1;} else {// 按照上行流量的正序排if (this.upFlow o.upFlow) {return 1;} else if (this.upFlow o.upFlow) {return -1;} else {return 0;}}}
}FlowMapper.java
package com.atguigu.mapreduce.writableComparable;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;public class FlowMapper extends MapperLongWritable, Text, FlowBean, Text {private FlowBean outK new FlowBean();private Text outV new Text();Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {// 获取一行String line value.toString();// 切割String[] split line.split(\t);// 封装outV.set(split[0]);outK.setUpFlow(Long.parseLong(split[1]));outK.setDownFlow(Long.parseLong(split[2]));outK.setSumFlow();// 写出context.write(outK, outV);}
}
三、测试
可以看出实现了排序效果。 同时我们可以在这个基础上实现分区 这样就实现了分区排序。
分区教程参考Hadoop3MapReduce中的Partition原理及自定义Partition
