Hadoop3教程（八）：MapReduce中的序列化概述

（79）MR序列化概述

什么是序列化，什么是反序列化？

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

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

为什么要序列化呢？

• 因为存活在内存里的对象，关机断电之后就没有了，要持久化保存的话，必须先序列化；

• 本地内存里的对象，只能供本地进程使用，如果想发送到另外一台计算机上使用，也必须先序列化。

那两台节点之间的内存数据传输，具体可以怎么做呢。

需要先序列化节点A中需要传输的内存数据，然后将序列化的结果传输到节点B中，然后节点B进行一个加载（反序列化）到内存，就实现了不同节点间，内存到内存的数据传输。

为什么不用java自带的序列化，而是Hadoop自己有一套序列化呢？

原因很简单，java的序列化中，待传输数据块后面都是跟了一大堆校验信息的。这对Hadoop来讲，有些过于繁重了，不便于在网络中高效传输，Hadoop里可能并不需要这么多的校验位，它只需要做简单校验就可以了。

基于这种需求，Hadoop就自己搞了一套序列化。主要是为了轻量

Hadoop的这套序列化，有什么好处呢？

• 结构紧凑；
• 存储空间占用相对少；
• 传输快；
• 互操作性；多种语言都可以反序列化（竟然有这个使用需求么还。。。）

（80）自定义序列化步骤

一般来讲，Hadoop里提供的那几种序列化类型，往往不能满足企业的要求，这时候企业就需要自定义一个bean对象，用于在Hadoop内部传递。

如果要自定义一个序列化对象的话，需要实现Writable接口，并重写以下方法：

void write(DataOutput out);                # 序列化
void readFields(DataInput in);        # 反序列化
1
2

注意，序列化时元素的顺序要跟反序列化的顺序完全一致。（这个很好理解，相当于位置参数嘛）

如：

@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();
}
1
2
3
4
5
6
7
8
9
10
11
12
13

同时，如果想把结果显示在文件里（或者打印出来），都需要重写toString()，否则显示出来的是个内存地址值。

最后，如果想把自定义的bean放在key中传输，还需要实现Comparable接口，因为Map阶段需要对数据做shuffle，这意味着数据的key必须是能排序的。

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

（81）序列化案例需求分析

需求案例：统计每个手机号耗费的总上行流量、总下行流量和总流量。

输入数据是每个手机号对每个网站的流量消耗情况。

输出数据是每个手机号的总上行流量、总下行流量和总流量。

需求设计的重点在于，明确map阶段输入输出的KV类型，reduce阶段输入输出的KV类型。

其中，map阶段输入的KV类型不需要操心，K相当于就是行号，V就是每行的内容；

而map阶段输出的KV跟reduce阶段输入的KV是一样的。

结合本次需求，考虑到要聚合的是手机号，所以map输出的K就应该设置成手机号，而value就只能设置成一个bean对象，包含了该条数据中的上行流量字段、下行流量字段，以及加和得到的总流量。

以以上形式，输入到reduce。

这里需要注意，bean对象如果想在不同节点（从map的节点传到reduce的节点）传输，就必须实现序列化接口。

（82）序列化案例代码

直接原样贴一下教程的代码，这块仅做了解，我也并没有实操，主要是考虑结合代码可能更好理解原理，所以还是在这里直接复制了。

1）编写自定义Bean对象：

package com.atguigu.mapreduce.writable;

import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

//1 继承Writable接口
public class FlowBean implements Writable {

    private long upFlow; //上行流量
    private long downFlow; //下行流量
    private long sumFlow; //总流量

    //2 提供无参构造
    public FlowBean() {
    }

    //3 提供三个参数的getter和setter方法
    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;
    }

    //4 实现序列化和反序列化方法,注意顺序一定要保持一致
    @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();
    }

    //5 重写ToString
    @Override
    public String toString() {
        return upFlow + "\t" + downFlow + "\t" + sumFlow;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68

2）编写Mapper类：

package com.atguigu.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<LongWritable, Text, Text, FlowBean> {
    private Text outK = new Text();
    private FlowBean outV = new FlowBean();

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {

        //1 获取一行数据,转成字符串
        String line = value.toString();

        //2 切割数据
        String[] split = line.split("\t");

        //3 抓取我们需要的数据:手机号,上行流量,下行流量
        String phone = split[1];
        String up = split[split.length - 3];
        String down = split[split.length - 2];

        //4 封装outK outV
        outK.set(phone);
        outV.setUpFlow(Long.parseLong(up));
        outV.setDownFlow(Long.parseLong(down));
        outV.setSumFlow();

        //5 写出outK outV
        context.write(outK, outV);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

3）编写Reducer类：

package com.atguigu.mapreduce.writable;

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

public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> {
    private FlowBean outV = new FlowBean();
    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException {

        long totalUp = 0;
        long totalDown = 0;

        //1 遍历values,将其中的上行流量,下行流量分别累加
        for (FlowBean flowBean : values) {
            totalUp += flowBean.getUpFlow();
            totalDown += flowBean.getDownFlow();
        }

        //2 封装outKV
        outV.setUpFlow(totalUp);
        outV.setDownFlow(totalDown);
        outV.setSumFlow();

        //3 写出outK outV
        context.write(key,outV);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

4）编写Driver驱动类：

package com.atguigu.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 conf = new Configuration();
        Job job = Job.getInstance(conf);

        //2 关联本Driver类
        job.setJarByClass(FlowDriver.class);

        //3 关联Mapper和Reducer
        job.setMapperClass(FlowMapper.class);
        job.setReducerClass(FlowReducer.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:\\inputflow"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\flowoutput"));
        
//7 提交Job
        boolean b = job.waitForCompletion(true);
        System.exit(b ? 0 : 1);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

参考文献

1. 【尚硅谷大数据Hadoop教程，hadoop3.x搭建到集群调优，百万播放】