大数据之hadoop中的序列化

                序列化就是把内存中的对象，转换成字节序列（或其他数据传输协议）以便于存储到磁 盘（持久化）和网络传输。反序列化就是将收到字节序列（或其他数据传输协议）或者是磁盘的持久化数据，转换 成内存中的对象。

                紧凑（高效使用存储空间）；快速（读写数据较为轻便）；互操作（支持多语言交互）

                自我理解：定义序列化接口就是为了将多个参数进行封装统一使用，方面后续的使用。

实现bean对象序列化步骤如下：

        必须实现Writable接口（implements）

        反序列化时，需要反射调用空参构造函数，必须要有空参构造

public FlowBean() {
super();
}

        重写序列化方法

@Override
public void write(DataOutput out) throws IOException {
    out.writeLong(upFlow);
    out.writeLong(downFlow);
    out.writeLong(sumFlow);
}

        重写反序列化方法

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

        注意反序列化的顺序和序列化的顺序完全一致

        要想把结果显示在文件中，需要重写 toString()，可用"t"分开，方便后续用

        如果需要将自定义的 bean 放在 key 中传输，则还需要实现 Comparable 接口，因为 MapReduce 框中的 Shuffle 过程要求对 key 必须能排序。

@Override
public int compareTo(FlowBean o) {
// 倒序排列，从大到小
return this.sumFlow > o.getSumFlow() ? -1 : 1;
}

               首先准备一个数据文本（分别为索引，电话号码，ip，网址，上行流量，下行流量，状态码）。主要需求是计算一个手机号码的上行流量，下行流量和总流量。由于hadoop的map只能输出一个类型数值，因此创建一个FlowBean将上行流量，下行流量和总流量进行封装。

        接着编写一个FlowBean对象

package cn.itjdb.mapreduce.writable;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;


public class FlowBean implements Writable {
    //hadoop的序列化就是将多个参数封装在一起
    private Long upFlow; //上行流量
    private Long downFlow; //下行流量
    private Long sumFlow; //总流量

    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;  //直接进行总流量计算
    }

    //重写空构造方法
    public FlowBean() {
    }

    //重写序列化方法
    @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();
    }

    //重写toString方法


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

         编写Mapper类

package cn.itjdb.mapreduce.writable;

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 Mapper {
    //对输出的对象进行实例化
    private Text outK=new Text();
    private FlowBean outV=new FlowBean();
    //重写map方法
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //获取一行数据
        String line=value.toString();

        //将获取的一行数据进行切割
        String[] s = line.split("t");  //原数据的数据是一个tab键进行分离的

        //获取要进行处理的数据
        String phone=s[1];
        String upFlow=s[s.length-3];
        String downFlow=s[s.length-2];
        //不能s[-3]这样提取数据
        //String upFlow=s[-3];
        //String downFlow=s[-2];
        System.out.println(phone);

        outK.set(phone);  //将String类型的phone转换为hadoop中的Text类型
        //将获取的数据进行封装到FlowBean中
        outV.setUpFlow(Long.parseLong(upFlow));  //若直接填upFlow会显示错误，要先将其转化为Long类型,再进行封装
        outV.setDownFlow(Long.parseLong(downFlow));
        outV.setSumFlow();//可以不写吗？,不可以，这是计算一行的总流量，而在map中计算的是同一个手机号的总流量

        //写出
        context.write(outK,outV);


    }
}

         编写Reduce类

package cn.itjdb.mapreduce.writable;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;


public class FlowReduce extends Reducer {
    private FlowBean outV=new FlowBean();
    //重写reduce方法
    @Override
    protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {

        //遍历集合累加值
        long totalUp=0;
        long totalDown=0;
        for (FlowBean value : values) {
            totalUp+=value.getUpFlow();
            totalDown+=value.getDownFlow();
        }
        System.out.println("cc:"+key);
        //封装outK（已存在，不需要在定义），outV（全新的FlowBean）
        outV.setUpFlow(totalUp);
        outV.setDownFlow(totalDown);
        outV.setSumFlow();

        //写出
        context.write(key,outV);
    }
}

        编写Driver类

package cn.itjdb.mapreduce.writable;


import org.apache.hadoop.conf.Configuration;
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 java.io.IOException;

public class FlowDriver {
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //1、获取job
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        //2、获取jar包路径
        job.setJarByClass(FlowDriver.class);

        //3、关联mapper和reduce
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReduce.class);

        //4、设置map输出kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(FlowBean.class);

        //5、设置最终输出的kv类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(FlowBean.class);

        //6、设置输入路径和输出路径
        FileInputFormat.setInputPaths(job,new Path("D:\input\inputflow"));
        FileOutputFormat.setOutputPath(job,new Path("D:\input\outputflow"));

        //7、提交job
        boolean result=job.waitForCompletion(true);
        System.exit(result? 0:1);
    }
}

        本地模式：就可以在D:\input\outputflow路径下查看结果。