[大数据]大数据—Hadoop（九）_ MapReduce_02、序列化

1、序列化概述

1.1 需求

把服务器Hadoop102放在内存中的数据：ss和cls移动到 Hadoop103 上

1.2 实现方式

把内存中的ss和cls持久化到磁盘(字节码)，拷贝到103，再加载到内存

1.3 概念

• 序列化
  • 内存中的数据落盘
• 反序列化
  • 磁盘中的数据加载到内存

1.4 好处

解决了一个系统的内存对象，无法直接传到另一个系统的内存对象的问题

1.5 java序列化

• 疑问
  • java中有自带的 Serializable 接口，为什么不使用呢
• 原因
  • java的序列化比较重，包含 数据+各种校验信息/继承体系/头信息
  • 对于hadoop来说，只需要 数据+简单校验

1.6 好处

结构紧凑、存储空间少、传输快

2、自定义bean对象实现序列化接口

2.1 问题

hadoop提供的10个常用序列化类型不能满足所有要求

2.2 步骤

a、实现writable接口
如果要对一个bean对象实现服务器之间的传输，需要实现序列化接口，就像之前WordCount中的k,v，都是基本序列化类型

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

public FlowBean() {
	super();
}

c、重写序列化方法

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

d、重写反序列化方法

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

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

• 都是up、down、sum

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

• 如果不重写，默认打印的是地址值
• 打印地址里的数据需要使用toString()

g、如果需要将自定义的bean放在key中传输，则还需要实现Comparable接口

• Map<key,val,key,val>，其中，第二个key需要实现Comparable接口，因为shuffle阶段会对key进行排序
• 详见后面排序案例。

3、序列化案例实操

3.1 需求分析

统计每一个手机号耗费的总上行流量、总下行流量、总流量
（提示：总流量 = 总上行流量+总下行流量）

输入数据

期望输出格式

map的参数

<偏移量，一行数据，key，value>

• 思考：key 和 value 值是什么
  • key：手机号
    • 需要将相同手机号值进行累加
  • value：JavaBean（FlowBean）
• JavaBean需要哪些数据
  • 手机号
  • 上行流量
  • 下行流量
  • 总流量后期计算得出
  • 其他无关字段不取

Map阶段

• 读取一行数据，切分字段；
• 抽取手机号，上行流量，下行流量；
• 以手机号为key，bean对象为value输出，即context.write(手机号,bean)。

Reduce阶段

• 累加上行流量+下行流量得到总流量

3.2 编写步骤

3.2.1 编写流量统计的bean对象

public class FlowBean implements Writable {

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

// 空参构造
public FlowBean() {
}

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

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

public long getUpFlow() {
    return upFlow;
}

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

public long getDowmFlow() {
    return dowmFlow;
}

public void setDowmFlow(long dowmFlow) {
    this.dowmFlow = dowmFlow;
}

public long getSumFlow() {
    return sumFlow;
}

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

public void setSumFlow() {
    this.sumFlow = this.dowmFlow + this.upFlow;
}

@Override
public String toString() {
    // 最终展示效果
    return  upFlow +"\t"+ dowmFlow +"\t" + sumFlow;
  }
}

a、实现writable接口

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

c、重写序列化方法

d、重写反序列化方法

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

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

• 如果不重写，默认打印的是地址值
• 打印地址里的数据需要使用toString()

g、如果需要将自定义的bean放在key中传输，则还需要实现Comparable接口

• Map<key,val,key,val>，其中，第二个key需要实现Comparable接口，因为shuffle阶段会对key进行排序

3.2.2 编写Mapper类

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 获取一行
    // 1	13736230513	192.196.100.1	www.atguigu.com	2481	24681	200
    String line = value.toString();

    // 2 切割
    // 1	13736230513	192.196.100.1	www.atguigu.com	2481	24681	200
    String[] word = line.split("\t");

    // 3 抓取想要的数据
    // 1	13736230513	192.196.100.1	www.atguigu.com	2481	24681	200
    // 手机号 、 上行流量 、 下行流量
    String phone = word[1];
    // 正着数可能会有字段缺失，需要反着数

    String upFlow = word[word.length - 3];
    String downFlow = word[word.length - 2];

    // 4 封装
    outK.set(phone);
    outV.setUpFlow(Long.parseLong(upFlow));
    outV.setDowmFlow(Long.parseLong(downFlow));
    outV.setSumFlow(); // 会自动求和

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

3.2.3 编写Reducer类

public class FlowReduce 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 {

    // 1 遍历集合累加值
    long sumUp = 0;
    long sumDown = 0;

    for (FlowBean value : values) {
        sumUp += value.getUpFlow();
        sumDown += value.getDowmFlow();
    }

    // 2 封装
    outV.setUpFlow(sumUp);
    outV.setDowmFlow(sumDown);
    outV.setSumFlow();

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

3.2.4 编写Driver类

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、指定Jar包所在的本地路径 通过全类名反射Jar包在什么位置
    job.setJarByClass(FlowDriver.class);

    // 3、关联Mapper、Reducer
    job.setMapperClass(FlowMapper.class);
    job.setReducerClass(FlowReduce.class);

    // 4、指定Mapper输出数据的k,v类型
    job.setMapOutputKeyClass(Text.class);
    job.setMapOutputValueClass(FlowBean.class);

    // 5、指定最终输出数据的k,v类型
    job.setOutputKeyClass(Text.class);
    job.setOutputValueClass(FlowBean.class);

    // 6、指定输入文件所在目录
    FileInputFormat.setInputPaths(job,new Path("D:\\input\\inputflow"));

    // 7、指定输出文件所在目录
    FileOutputFormat.setOutputPath(job,new Path("D:\\hadoop\\output12"));

    // 8、提交作业
    boolean result = job.waitForCompletion(true);

    //成功返回0
    System.exit(result?0:1);
  }
}

3.3 最终输出