序列化

Hadoop : MapReduce 的序列化接口

序列化接口书写流程

1. 创建Bean类实现Writable接口
2. 创建无参构造
3. 重写write，readFields方法
4. 重写toString方法

创建FlowBean

public class FlowBean implements Writable {


    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.downFlow + this.upFlow;
    }
    @Override
    public void write(DataOutput dataOutput) throws IOException {
        dataOutput.writeLong(upFlow);
        dataOutput.writeLong(downFlow);
        dataOutput.writeLong(sumFlow);
    }

    @Override
    public void readFields(DataInput dataInput) throws IOException {
        this.upFlow = dataInput.readLong();
        this.downFlow = dataInput.readLong();
        this.sumFlow = dataInput.readLong();
    }

    @Override
    public String toString() {
        return upFlow + "\t" + downFlow + "\t" + sumFlow;
    }
}

创建FlowMapper

public class FlowMapper extends Mapper<Long, Text,Text,FlowBean> {
    private Text outK = new Text();
    private FlowBean outV = new FlowBean();
    @Override
    protected void map(Long key, Text value, Mapper<Long, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {
        String line = value.toString();
        String []split = line.split("\t");
        String phone = split[1];
        String up = split[split.length-3];
        String down =split[split.length-2];
        outK.set(phone);
        outV.setUpFlow(Long.parseLong(up));
        outV.setDownFlow(Long.parseLong(down));
        outV.setSumFlow();
        context.write(outK,outV);
    }
}

创建FlowReducer

public class FlowReducer extends Reducer<Text,FlowBean,Text,FlowBean> {
    private FlowBean outV = new FlowBean();
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {
        long totalUp=0;
        long totalDown=0;
        for(FlowBean value :values)
        {
            totalUp+= value.getUpFlow();
            totalDown+=value.getDownFlow();
        }
        outV.setUpFlow(totalUp);
        outV.setDownFlow(totalDown);
        outV.setSumFlow();
        context.write(key,outV);
    }
}

创建FlowDriver

public class FlowDriver {


    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);
        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]));
        Boolean result = job.waitForCompletion(true);
        System.exit(result?0:1);
    }


}