跳至主要內容

序列化

Mahe666

序列化概述

什么是序列化

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

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

为什么要序列化

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

为什么不用Java的序列化

​Java的序列化是一个重量级序列化框架 Serializable，一个对象被序列化后，会附带很多额外的信息（各种校验信息，Header，继承体系等），不便于在网络中高效传输。

所以， Hadoop自己开发了一套序列化机制（Writable）。

Hadoop序列化特点:

  1. 结构紧凑：高效使用存储空间。
  2. 快速速度快：读写数据的额外开销小。
  3. 可扩展：随着通信协议的升级而可升级
  4. 互操作：支持多语言的交互。例如使用Java语言进行Hadoop序列化，C或C++可以对该字节序列进行反序列化

自定义序列化步骤

找到 org.apache.hadoop.io 包下的 Writable 接口

也可以通过 IntWritable 等类，一层一层往上点

Writable 接口有两个方法，一个是 write()，用来序列化对象；另一个是 readFields() ，用来反序列化字节序列

在企业开发中往往常用的基本序列化类型不能满足所有需求，比如在Hadoop框架内部传递一个 bean 对象，那么该对象就需要实现序列化接口。

实现对象序列化需要如下7个步骤

  1. 必须实现 Writable 接口
  2. 反序列化时，需要反射调用无参构造函数，所以对象对应的类必须要有无参构造
  3. 重写 Writable接口 中的 write() 序列化方法
  4. 重写 Writable接口 中的 readFields() 反序列化方法
  5. 注意！反序列化的顺序和序列化的顺序完全一致
  6. 要想把结果显示在文件中，需要重写 toString() 方法，可用 \t 分开，方便后续用
  7. 如果需要将自定义的bean放在key中传输，则还需要实现 Comparable接口，因为 MapReduce 框架中的 Shuffle 过程要求对key必须能排序

其他的不用多说，主要说一下第7条。Comparable接口主要用于排序，排序的是输出的Key。例如先输出哪个Key，再输出哪个Key。

之前的WordCount示例中需要排序的是不同的单词，而这里自定义序列化的排序规则需要自己写。

想要同时实现 Writable接口 和 Comparable接口，可以去实现 org.apache.hadoop.io 包下的 WritableComparable 接口

电话流量统计 案例

输入文件：phone_data.txt

查看数据，共有七个字段，分别是：id、手机号码、网络ip、上行流量、下行流量、网络状态码

要求，统计每一个手机号码耗费的总上行流量、总下行流量和总流量

需求分析

期望输出：手机号码、总上行流量、总下行流量、总流量

统计每一个手机号码耗费的流量，也就是说需要将重复出现的手机号，在reduce阶段结束后，合并为一个Key，然后将后面的相同项数据合并

总上行流量 + 总下行流量 = 总流量

这里要先处理总上行流量，再处理总下行流量，最后相加得到总流量

而这里，重复的Key是手机号。而Value则是总上行流量、总下行流量、总流量三个值，任何一个已知的序列化类都无法清晰地同时体现出这三个属性的不同点

所以只能对Value，也就是将这三个值封装为bean

流程分析

  1. Map阶段
    1. 读取每一行数据，切分字符串
    2. 抽取手机号、上行流量、下行流量并计算当前行的总流量
    3. 以手机号为Key，bean对象为输出
    4. bean对象想要传输，就必须实现序列化接口
  2. Reduce阶段
    1. 累加上行流量和下行流量和每一行的总流量

Bean文件编写 FlowBean

import org.apache.hadoop.io.Writable;

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

/**
 * 命名为FlowBean的意思是：Flow是流量，流量的Bean类
 * 1. 定义类实现Writable接口
 * 2. 重写序列化和反序列化方法
 * 3. 重写无参构造方法
 * 4. 重写toString()方法
 */
public class FlowBean implements Writable {

    /**
     * 上行流量
     */
    private long upFlow;

    /**
     * 下行流量
     */
    private long downFlow;

    /**
     * 总流量
     */
    private long sumFlow;

    // get和set方法


    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;
    }

    /**
     * 重载setSumFlow()方法
     */
    public void setSumFlow() {
        this.sumFlow = this.upFlow + this.downFlow;
    }


    /**
     * 重写无参构造方法
     */
    public FlowBean() {
    }

    /**
     * 序列化方法，方法体中先序列化哪个，反序列化时就需要先接哪个
     * @param out <code>DataOuput</code> to serialize this object into.
     * @throws IOException
     */
    @Override
    public void write(DataOutput out) throws IOException {
        out.writeLong(upFlow);
        out.writeLong(downFlow);
        out.writeLong(sumFlow);
    }

    /**
     * 反序列化方法，接收值时，需要与序列化时的顺序完全一致
     * @param in <code>DataInput</code> to deseriablize this object from.
     * @throws IOException
     */
    @Override
    public void readFields(DataInput in) throws IOException {
        this.upFlow = in.readLong();
        this.downFlow = in.readLong();
        this.sumFlow = in.readLong();
    }

    /**
     * 方便写入数据
     * @return
     */
    @Override
    public String toString() {
        return upFlow + "\t" + downFlow + "\t" + sumFlow;
    }
}

Mapper文件编写 FlowMapper

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 outKey = new Text();
    private FlowBean outValue = new FlowBean();

    @Override
    protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {
        // 1. 获取一行数据
        String line = value.toString();

        // 2. 切割字符串
        String[] split = line.split("\t");

        // 3. 抓取想要的数据：手机号，上行流量和下行流量
        String phoneNumber = split[1];
        // 注意，数据源中部分的属性是空的，所以最简单的办法是从后往前数，不然很有可能取错值
        String up = split[split.length - 3];
        String down = split[split.length - 2];

        // 4. 封装
        outKey.set(phoneNumber);
        outValue.setUpFlow(Long.parseLong(up));
        outValue.setDownFlow(Long.parseLong(down));
        outValue.setSumFlow();

        // 5. 写出
        context.write(outKey, outValue);

    }
}

Reducer文件编写 FlowReducer

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

import java.io.IOException;

/**
 * 第一个泛型是map阶段输出的手机号
 * 第二个泛型是map阶段输出的bean对象，保存着需要合并处理的值
 * 第三个泛型是当前reduce阶段将要输出的Key，也是手机号。（map阶段输出的手机号会重复，reduce阶段不会）
 * 第四个泛型是当前reduce阶段将要输出的Value，是map阶段输出的相同手机号对应的值的合并
 */
public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> {

    private FlowBean outValue = new FlowBean();

    @Override
    protected void reduce(Text key, Iterable<FlowBean> values, Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {
        // 1. 遍历集合 累加值

        // 这里不能提升作用域，这里的目的是为了保证不同的手机号之间的初始总数为0
        long totalUp = 0;
        long totalDown = 0;

        for (FlowBean value : values) {
            totalUp += value.getUpFlow();
            totalDown += value.getDownFlow();
        }

        // 2. 封装输出outKey和outValue
        outValue.setUpFlow(totalUp);
        outValue.setDownFlow(totalDown);
        outValue.setSumFlow();

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

    }

}

Driver文件编写 FlowDriver

注意

注意！这里的路径是写死的，如果想要改活，去参考wordcount案例中的Driver代码

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, InterruptedException, ClassNotFoundException {
        // 1.获取job
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);

        // 2.设置jar包路径
        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("C:\\Users\\MaHe666\\Desktop\\phone_data.txt"));
        FileOutputFormat.setOutputPath(job, new Path("D:\\output"));

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

    }

}
上次编辑于:
贡献者: mahe666
进入本站的人默认认为马赫很帅
Copyright © 2024 Mahe666