MapReduce自定义排序WriteComparable

不念不忘少年蓝@ 2023-10-04 16:34 65阅读 0赞

#### MapReduce自定义排序 ####

MapReduce会按数据的Key进行默认的排序，如果想要进行定制排序，可以实现WritableComparable接口，重写compareTo方法

**注意，以上的办法是作用于MapReduce阶段的Key值进行排序，如果要对Value进行排序，则这里的写法不会起作用，按照Value的排序，在下面给出**

实现WritableComparable接口之后需要实现三个方法，分别是序列化方法，反序列化方法和compareTo方法

**需求，按照时间戳的正序进行排序**

要求在AISDataBean类中对其属性unixTime按照从小到大进行排序

AISDataBean类得编写：

package com.gis507.test.AISDataSort;
    
    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、属性有：speed，unixTime，lon_d，lat_d，(String)draught(类型),Length,Width,CallSign,Course,Rot
     */
    public class AISDataBean implements Writable {
        
    
        private double speed;
        private long unixTime;
        private double lon_d;
        private double lat_d;
        private String draught;
        private String length;
        private String width;
        private String callsign;
        private double course;
        private double rot;
    
        public AISDataBean(double speed, long unixTime, double lon_d, double lat_d, String draught, String length, String width, String callsign, double course, double rot) {
        
            this.speed = speed;
            this.unixTime = unixTime;
            this.lon_d = lon_d;
            this.lat_d = lat_d;
            this.draught = draught;
            this.length = length;
            this.width = width;
            this.callsign = callsign;
    //        this.heading = heading;
            this.course = course;
            this.rot = rot;
        }
    
        public AISDataBean() {
        
        }
    
        public double getSpeed() {
        
            return speed;
        }
    
        public void setSpeed(double speed) {
        
            this.speed = speed;
        }
    
        public long getUnixTime() {
        
            return unixTime;
        }
    
        public void setUnixTime(long unixTime) {
        
            this.unixTime = unixTime;
        }
    
        public double getLon_d() {
        
            return lon_d;
        }
    
        public void setLon_d(double lon_d) {
        
            this.lon_d = lon_d;
        }
    
        public double getLat_d() {
        
            return lat_d;
        }
    
        public void setLat_d(double lat_d) {
        
            this.lat_d = lat_d;
        }
    
        public String getDraught() {
        
            return draught;
        }
    
        public void setDraught(String draught) {
        
            this.draught = draught;
        }
    
        public String getLength() {
        
            return length;
        }
    
        public void setLength(String length) {
        
            this.length = length;
        }
    
        public String getWidth() {
        
            return width;
        }
    
        public void setWidth(String width) {
        
            this.width = width;
        }
    
        public String getCallsign() {
        
            return callsign;
        }
    
        public void setCallsign(String callsign) {
        
            this.callsign = callsign;
        }
    
        public double getCourse() {
        
            return course;
        }
    
        public void setCourse(double course) {
        
            this.course = course;
        }
    
        public double getRot() {
        
            return rot;
        }
    
        public void setRot(double rot) {
        
            this.rot = rot;
        }
    
        @Override
        public String toString() {
        
            return speed + "," + unixTime + "," + lon_d + "," + lat_d + "," + draught + ","
                    + length + "," + width + "," + callsign + "," + course + "," + rot ;
        }
    
    
        @Override
        public void write(DataOutput dataOutput) throws IOException {
        
    
            //speed，unixTime，lon_d，lat_d，(String)draught(类型),Length,Width,CallSign,Course,Rot
            dataOutput.writeDouble(speed);
            dataOutput.writeLong(unixTime);
            dataOutput.writeDouble(lon_d);
            dataOutput.writeDouble(lat_d);
            dataOutput.writeUTF(draught);
            dataOutput.writeUTF(length);
            dataOutput.writeUTF(width);
            dataOutput.writeUTF(callsign);
            dataOutput.writeDouble(course);
            dataOutput.writeDouble(rot);
    
        }
    
        @Override
        public void readFields(DataInput dataInput) throws IOException {
        
    
            this.speed = dataInput.readDouble();
            this.unixTime = dataInput.readLong();
            this.lon_d = dataInput.readDouble();
            this.lat_d = dataInput.readDouble();
            this.draught = dataInput.readUTF();
            this.length = dataInput.readUTF();
            this.width = dataInput.readUTF();
            this.callsign = dataInput.readUTF();
            this.course = dataInput.readDouble();
            this.rot = dataInput.readDouble();
    
        }
    
        @Override
        public int compareTo(AISDataBean o) {
        
    
            if (this.getUnixTime() > o.getUnixTime()){
        
                return 1;
            }else{
        
                return -1;
            }
        }
    }

Mapper和Reducer和Driver就正常按照逻辑写就行

**注意，这里的排序需要把AISDataBean当做key进行处理，如果要对value进行排序，通常的方法是：**

**1、在map阶段将AISDataBean当做是key，在Reducer的输出阶段将key和value互换位置，这里就要注意在Driver中设置输出的key和value类型时设置正确**

**2、除了上述方法，还可以在reducer阶段对value进行排序**

@Override
        protected void reduce(Text key, Iterable<AISDataBean> values, Context context) throws IOException, InterruptedException {
        
    
            //实例化一个列表，用于存放数据
            List<AISDataBean> aisList = new ArrayList<>(1000);
    
            //遍历values
            for (AISDataBean value : values) {
        
    
                //将value添加到列表
                aisList.add(new 		AISDataBean(value.getSpeed(),value.getUnixTime(),value.getLon_d(),value.getLat_d(),value.getDraught(),value.getLength(),value.getWidth(),value.getCallsign(),value.getCourse(),value.getRot()));   
                
                log.info(key + "," + value.getUnixTime());
            }
    
            //使用lambda表达式对aisList按照时间进行排序
    //        aisList.stream().sorted(Comparator.comparingLong(AISDataBean::getUnixTime));
    
            //出了上处的写法外，还可以使用lambda表达式这样排序
            aisList.sort((AISDataBean a1,AISDataBean a2) -> a1.compareTo(a2));
    		//遍历列表
            for (AISDataBean aisDataBean : aisList) {
        
    			//写出
                context.write(key,aisDataBean);
            }
        }

**注意：在使用的compareTo是在Bean中定义之后使用的，否则就会使用java提供的默认compareTo方法**