0070-mapreduce自定义输出类

ゝ一世哀愁。 2023-07-14 11:00 6阅读 0赞

### 文章目录 ###

*   *   *  1. 需求背景
         *  2. 场景
         *  3. 具体实现
         *   *  3.1 自定义输出类
             *  3.2 Mapper类
             *  3.3 Reducer类
             *  3.4 job类-标准写法
         *  4. 总结

### 1. 需求背景 ###

mapper-reducer程序只能输入键值对，如果需要输出多个字段信息，必须对键或值进行自定义封装。

### 2. 场景 ###

// 手机号码 上行流量 下行流量 其他字段
    13576119010 11 22 aa1 bb1
    13676119011 12 23 aa2 bb2
    13776119012 13 24 aa3 bb3
    13876119013 14 25 aa4 bb4
    13976119014 15 26 aa5 bb5
    14076119015 16 27 aa6 bb6
    14176119016 17 28 aa7 bb7

需求统计手机号码和手机号码对应的上行流量，下行流量和总流量。

### 3. 具体实现 ###

#### 3.1 自定义输出类 ####

package com.honor.flowcount;
    
    import lombok.Data;
    import org.apache.hadoop.io.Writable;
    
    import java.io.DataInput;
    import java.io.DataOutput;
    import java.io.IOException;
    
    /** * create by rongyao * at 2/12/20 * 继承Writable，这是hadoop自定义的序列化接口，轻便不会序列化父类结构 */
    @Data
    public class FlowBean implements Writable { 
        private String number;
        private long upFlow;
        private long downFlow;
        private long sumFlow;
    
        public FlowBean(String number, long upFlow, long downFlow) { 
            this.number = number;
            this.upFlow = upFlow;
            this.downFlow = downFlow;
            this.sumFlow = upFlow + downFlow;
        }
    
        public FlowBean() { 
        }
    
        @Override
        public void write(DataOutput out) throws IOException { 
            out.writeUTF(number);
            out.writeLong(upFlow);
            out.writeLong(downFlow);
            out.writeLong(sumFlow);
        }
    
        @Override
        public void readFields(DataInput in) throws IOException { 
            this.number = in.readUTF();
            this.upFlow = in.readLong();
            this.downFlow = in.readLong();
            this.sumFlow = in.readLong();
        }
    
        @Override
        public String toString() { 
            return upFlow + " " + downFlow + " " + sumFlow;
        }
    }

#### 3.2 Mapper类 ####

package com.honor.flowcount;
    
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    /** * create by rongyao * at 2/12/20 */
    public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean> { 
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { 
            String line = value.toString();
            String[] strings = line.split("\\s");
            String number = strings[0];
            long upFlow = Long.parseLong(strings[1]);
            long downFlow = Long.parseLong(strings[2]);
            FlowBean flowBean = new FlowBean(number, upFlow, downFlow);
            context.write(new Text(number), flowBean);
        }
    }

#### 3.3 Reducer类 ####

package com.honor.flowcount;
    
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    /** * create by rongyao * at 2/12/20 */
    public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> { 
        @Override
        protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException { 
            long upFlowCounter = 0;
            long downFlowCounter = 0;
            for (FlowBean flowBean : values) { 
                upFlowCounter += flowBean.getUpFlow();
                downFlowCounter += flowBean.getDownFlow();
            }
            context.write(key, new FlowBean(key.toString(), upFlowCounter, downFlowCounter));
        }
    }

#### 3.4 job类-标准写法 ####

package com.honor.flowcount;
    
    import org.apache.commons.math3.optim.nonlinear.vector.JacobianMultivariateVectorOptimizer;
    import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    /** * create by rongyao * at 2/12/20 */
    public class FlowRunner extends Configured implements Tool { 
        @Override
        public int run(String[] args) throws Exception { 
            Configuration conf = new Configuration();
            Job job = Job.getInstance(conf);
            job.setJarByClass(this.getClass());
    
            job.setMapperClass(FlowMapper.class);
            job.setReducerClass(FlowReducer.class);
    
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(FlowBean.class);
    
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(FlowBean.class);
    
            FileInputFormat.setInputPaths(job, new Path(args[0]));
            FileOutputFormat.setOutputPath(job, new Path(args[1]));
            return job.waitForCompletion(true) ? 1 : -1;
        }
    
        public static void main(String[] args) throws Exception { 
            ToolRunner.run(new Configuration(), new FlowRunner(), args);
        }
    }

### 4. 总结 ###

可以自定义输出类以后，就解决了键值对的限制，可以输出任意信息。