品牌建设传播网站公司福田网站建设有限公司

（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);        # 反序列化

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

如：

@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()，否则显示出来的是个内存地址值。

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

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

（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 实现序列化和反序列化方法,注意顺序一定要保持一致@Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeLong(upFlow);dataOutput.writeLong(downFlow);dataOutput.writeLong(sumFlow);}@Overridepublic void readFields(DataInput dataInput) throws IOException {this.upFlow = dataInput.readLong();this.downFlow = dataInput.readLong();this.sumFlow = dataInput.readLong();}//5 重写ToString@Overridepublic String toString() {return upFlow + "\t" + downFlow + "\t" + sumFlow;}
}

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();@Overrideprotected 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 outVoutK.set(phone);outV.setUpFlow(Long.parseLong(up));outV.setDownFlow(Long.parseLong(down));outV.setSumFlow();//5 写出outK outVcontext.write(outK, outV);}
}

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();@Overrideprotected 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 封装outKVoutV.setUpFlow(totalUp);outV.setDownFlow(totalDown);outV.setSumFlow();//3 写出outK outVcontext.write(key,outV);}
}

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和Reducerjob.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 提交Jobboolean b = job.waitForCompletion(true);System.exit(b ? 0 : 1);}
}

参考文献

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