序列化和反序列化

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

反序列化就是将收到字节序列（或其他数据传输协议）或者是硬盘的持久化数据，转换成内存中的对象。

主要作用是将MR中产生的数据以序列化类型在网络中、不同的电脑中进行数据传递

引入序列化的原因

一般来说，"活的" 对象只生存在内存里，关机断电就没有了。而且"活的"对象只能由本地的进程使用，不能被发送到网络上的另外一台计算机。然而序列化可以存储"活的" 对象，可以将"活的"对象发送到远程计算机，方便分布式处理

Java序列化和大数据序列化的区别

Java的序列化是一个重量级序列化框架（Serializable），一个对象被序列化后，会附带很多额外的信息（各种校验信息，header，继承体系等），不便于在网络中高效传输。所以，hadoop自己开发了一套序列化机制（Writable），精简、高效。

Hadoop序列化特点:

1. 紧凑:高效使用存储空间。
2. 快速:读写数据的额外开销小。
3. 可扩展:随着通信协议的升级而可升级
4. 互操作:支持多语言的交互

常用数据序列化类型

Java类型

Hadoop Writable类型

boolean

BooleanWritable

byte

ByteWritable

int

IntWritable

float

FloatWritable

long

LongWritable

double

DoubleWritable

string

Text

map

MapWritable

array

ArrayWritable

案例实操

计算每一个手机号总的上行流量、总的下行流量、总流量

在HDFS上的 /school 目录下有 phone_data.txt 文件，该文件中记录有手机号的上行流量、下行流量

其中：上行流量：倒数第三列；下行流量：倒数第二列

1. Map阶段：

   读取到每一行数据，将每一行中的数据的手机号、上行流量、下行流量获取出来。

   map阶段在去输出中间键值对数据的时候，应该以手机号为key，以上行流量和下行流量为value，发送给reduce

2. Reduce阶段：

   reduce根据手机号（key）将这个手机对应的所有上行流量和下行流量获取到，累加即可

   其中通过实现自定义的bean来封装流量信息，并将bean作为map输出的value来传输

DataCountMapper.java

public class DataCountMapper extends Mapper {
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
String line = value.toString();
String[] fields = line.split("\t");
FlowBean flowBean = new FlowBean();
String phone = fields[1];
int upFlow = Integer.parseInt(fields[fields.length - 3]);
int downFlow = Integer.parseInt(fields[fields.length - 2]);
flowBean.setPhone(phone);
flowBean.setUpFlow(upFlow);
flowBean.setDownFlow(downFlow);
/**
* 将数据以手机号为key，flowBean对象为value写出到reduce
*/
context.write(new Text(phone), flowBean);
}
}

DataCountReducer.java

public class DataCountReducer extends Reducer {
@Override
protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
Iterator iterator = values.iterator();
int upSum = 0;
int downSum = 0;
while (iterator.hasNext()) {
FlowBean bean = iterator.next();
upSum += bean.getUpFlow();
downSum += bean.getDownFlow();
}
int sum = upSum + downSum;
FlowBean flowBean = new FlowBean();
flowBean.setPhone(key.toString());
flowBean.setUpFlow(upSum);
flowBean.setDownFlow(downSum);
flowBean.setSumFlow(sum);
context.write(NullWritable.get(), flowBean);
}
}

DataCountDriver.java

public class DataCountDriver {
public static void main(String[] args) throws Exception {
// 获取job
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
// 设置jar包
job.setJarByClass(DataCountDriver.class);
// 关联Mapper和Reducer
job.setMapperClass(DataCountMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(FlowBean.class);

    job.setReducerClass(DataCountReducer.class);  
    job.setOutputKeyClass(NullWritable.class);  
    job.setOutputValueClass(FlowBean.class);  
    // 管理文件的输入和文件的输出  
    FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.218.55:9000/school/phone\_data.txt"));  
    Path output = new Path("hdfs://192.168.218.55:9000/test/phone\_data");  
    FileSystem fs = FileSystem.get(new URI("hdfs://192.168.218.55:9000"), conf, "root");  
    if (fs.exists(output)) {  
        fs.delete(output, true);  
    }  
    FileOutputFormat.setOutputPath(job, output);  
    // 提交运行  
    boolean result = job.waitForCompletion(true);  
    System.exit(result ? 0 : 1);  
}  

}

FlowBean.java

public class FlowBean implements Writable {
/**
* JavaBean对象主要目的是为了封装手机号的上行流量和下行流量，然后在Map阶段当做key-value键值对的value输出到Reduce阶段
*/
private String phone;
private int upFlow;
private int downFlow;
private int sumFlow;

public String getPhone() {  
    return phone;  
}

public void setPhone(String phone) {  
    this.phone = phone;  
}

public int getUpFlow() {  
    return upFlow;  
}

public void setUpFlow(int upFlow) {  
    this.upFlow = upFlow;  
}

public int getDownFlow() {  
    return downFlow;  
}

public void setDownFlow(int downFlow) {  
    this.downFlow = downFlow;  
}

public int getSumFlow() {  
    return sumFlow;  
}

public void setSumFlow(int sumFlow) {  
    this.sumFlow = sumFlow;  
}

@Override  
public String toString() {  
    return "FlowBean{" +  
            "phone='" + phone + '\\'' +  
            ", upFlow=" + upFlow +  
            ", downFlow=" + downFlow +  
            ", sumFlow=" + sumFlow +  
            '}';  
}

/\*\*  
 \* 序列化方法：将Java对象的属性值怎么序列化写出  
 \* @param dataOutput  
 \* @throws IOException  
 \*/  
@Override  
public void write(DataOutput dataOutput) throws IOException {  
    // 将一个String类型的属性序列化写出成二进制数据  
    dataOutput.writeUTF(this.phone);  
    // 将一个String类型的属性序列化写出成二进制数据  
    dataOutput.writeInt(upFlow);  
    dataOutput.writeInt(downFlow);  
    dataOutput.writeInt(sumFlow);  
}

/\*\*  
 \* 反序列化方法：怎么将二进制代码转成JavaBean对象属性的值  
 \*    反序列化的时候，读取二进制数据时，不能随便读  
 \*    序列化写出时先写出哪个属性的值，就先读哪个属性值  
 \* @param dataInput  
 \* @throws IOException  
 \*/  
@Override  
public void readFields(DataInput dataInput) throws IOException {  
    this.phone = dataInput.readUTF();  
    this.upFlow = dataInput.readInt();  
    this.downFlow = dataInput.readInt();  
    this.sumFlow = dataInput.readInt();  
}  

}

Notes:

FlowBean是无法直接当做MR程序的key-value键值对的，除非JavaBean对象是hadoop的一个序列化对象

1. 让自定义的JavaBean对象实现一个借口：Writable（Hadoop的序列化接口，实现了这个接口，这个JavaBean对象就可以实现序列化）

2. 重写接口中的两个方法：①序列化的方法---怎么将对象序列化成二进制； ②反序列化的方法---怎么将二进制转成JavaBean对象

如果我们想用一个自定义的JavaBean对象去充当MR程序的key-value键值对的输入和输出，那么JavaBean对象必须实现Hadoop的序列化机制 ：

1. 实现接口Writable

2. 重写write方法---序列化方法，将Java对象的属性值序列化写出

3. 重写readFields方法---反序列化方法，将二进制数据反序列化成JavaBean对象的属性值

   要求：反序列化的顺序必须是write方法写出数据的顺序

4. 如果我们reduce阶段也是输出的JavaBean对象，那么在文件当中数据的格式就是JavaBean对象的toString()方法