MapReduce 序列化

MapReduce 序列化

概述

节点通过字节码传输

序列化内存->字节码

反序列化字节码->内存

自定义序列化

常用数据序列化类型

hadoop序列化采用简单校验使得存储空间少、传输速度快

int与IntWritable转化

//b是int类型

IntWritable outV = new IntWritable();

outV.set(b);

//a是IntWritable类型

int b = outV.get();

Text与String

//Text --> String

Text text = new Text();

String s = text.toString();

//String --> Text

Text.set(string);

序列化读写方法

序列化

String类型: writeUTF(str)

Int类型:writeInt(int)

Long类型:writeLong(long)

自定义bean对象实现序列化接口(Writable)

常用数据序列化类型并不能满足所有的需求，比如bean对象。

1.实现Writable接口

2.重写序列化方法write

3.重写反序列化方法readFields

4.反序列化，需要反射调用空参构造函数，所以必须有空参构造器

5.序列化的顺序和反序列化的顺序一致

可以想象成队列，先进的要先出去

6.要想把结果显示在文件中，需要重写toString()

默认传输过来的是地址值，可用'\t'分开，方便后续使用

7.如果自定义的bean放在key中传输，还需要实现Comparable接口，因为MapReduce框中的Shuffle过程要求对key必须能排序

序列化案例实操

需求

统计每个手机号消费的总上行流量、总下行流量、总流量

输入数据格式:

id 手机号网络ip 域名上行流量下行流量网络状态码

期望输出数据格式

格式中的空格可以通过重写toString()控制

需求分析

关注输入与输出

map阶段

输入的key:这一行的偏移量

输入的value:这一行的数据

输出的key:手机号

统计的是每个手机号的消费情况

输出的value:bean对象{上行流量、下行流量、总流量}

1.读取一行数据，切分字段（原始数据使用\t分割的）

1 13736230513 192.196.100.1 www.atguigu.com 2481 24681 200

2.根据输出格式，抽取需要的数据手机号、上行流量、下行流量

3.以手机号为key，bean对象为value输出

context.write(手机号,bean)

4.这里的bean对象是我们自定义的，本身是不可以序列化的。bean对象想要能够传输，必须实现序列化接口

可能Map在hadoop102，Reduce在hadoop103

Reduce阶段

1.累加上行流量和下行流量得到总流量

13736230513 2481 + 24681 = 27162

编写MapReduce程序

FlowBean类

bean对象{上行流量、下行流量、总流量}，作为中间的数据传输

1.这个对象可以序列化，实现writable接口

2.重写序列化和反序列化方法

3.重写空参构造，用于反射调用空参构造函数

4.重写toString方法用于打印输出

package ranan.mapreduce.writable;

import org.apache.hadoop.io.Writable;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

/**

 * 1.这个对象可以序列化，实现writable接口

 * 2.重写序列化和反序列化方法

 * 3.重写空参构造，用于反射调用空参构造函数

 * 4.重写toString方法用于打印输出

 */

public class FlowBean implements Writable {

    private long upFlow; //上行流量

    private long downFlow; //下行流量

    private long sumFlow; //总流量

    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;

    }

    //3.重写空参构造

    public FlowBean() {

    }

    //4.重写toString方法

    @Override

    public String toString() {

        //输出会调用此对象的此方法，所以按输出的格式来写

        return upFlow + "\t" + downFlow + "\t" + sumFlow;

    }

    //2.重写序列化方法

    @Override

    public void write(DataOutput dataOutput) throws IOException {

        //这里的数据都是Long类型，所以使用writeLong

       dataOutput.writeLong(upFlow);

       dataOutput.writeLong(downFlow);

       dataOutput.writeLong(sumFlow);

    }

    //2.重写反序列化方法

    @Override

    public void readFields(DataInput dataInput) throws IOException {

        //反序列化顺序需要和序列化顺序一致，想象队列

        this.upFlow=dataInput.readLong();

        this.downFlow=dataInput.readLong();

        this.sumFlow=dataInput.readLong();

    }

}

编写FlowMapper类

1.继承Mapper类，注意是org.apache.hadoop.mapreduce.Mapper

2.确定输入输出的key-value

输入key是偏移量LongWritable，value是这行内容Text，注意是org.apache.hadoop.io

输出key是手机号字符串Text，value是自定义变量FlowBean

3.重写map()方法

package ranan.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, Mapper<LongWritable, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {

        //1.获取一行信息，转换成String

        String line = value.toString();

        //1	13736230513	192.196.100.1	www.atguigu.com	2481	24681	200

        //2.切割"/t"

        String[] item = line.split("\t");

        //3.抓取要想的数据 手机号，上行流量，下行流量

        String phone = item[1];

        //上行流量与下行流量有残缺需要注意，有些行有域名有些行没有域名

        //从后往前数 都在倒数第二，第三个

        Long up = Long.parseLong(item[item.length - 3]);

        Long down = Long.parseLong(item[item.length - 2]);

        //4.封装

        outK.set(phone);

        outV.setUpFlow(up);

        outV.setDownFlow(down);

        outV.setSumFlow();

        //写出

        context.write(outK,outV);

    }

}

编写FlowReducer类

1.继承Reducer类，注意是import org.apache.hadoop.mapreduce.Reducer;

2.FlowReducer的输入就是Mapper的输出，本道题FlowReducer的输出就是输入的数据类型

3.重写reduce()方法

package ranan.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, Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {

        //遍历结合累加值

        long upTotal=0,dowmTotal=0;

        for (FlowBean value : values){

            upTotal+=value.getUpFlow();

            dowmTotal+=value.getDownFlow();

        }

        //封装输出outK与outV

        outV.setUpFlow(totalUp);

        outV.setDownFlow(dowmTotal);

        outV.setSumFlow();

        context.write(key,outV);

    }

}

编写FlowDriver驱动类

1.获取job

2.设置jar包路径

3.关联mapper和reducer

4.设置map输出的kv类型

5.设置最终输出的kV类型,有些程序没有reduce阶段，所以这里设置的是最终输出而不是reduce输出类型

6.设置输入路径和输出路径

7.提交job

输出路径一定要不存在

package ranan.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 FlowDrier {

    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        // 1 获取job

        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf); //单例模式实例化job

        //2.设置jar

        job.setJarByClass(FlowDrier.class);

        //3.关联mapper和reducer

        job.setMapperClass(FlowMapper.class);

        job.setReducerClass(FlowReducer.class);

        //4.设置mapper输出的key和value类型

        job.setMapOutputKeyClass(Text.class);

        job.setMapOutputValueClass(FlowBean.class);

        //5.设置最终数据输出的key和value类型

        job.setOutputKeyClass(Text.class);

        job.setOutputValueClass(FlowBean.class);

        //6.设置数据的输入路径和输出路径

		//还可以指明具体的文件D:\\hadoop_data\\input\\phone_data.txt

       FileInputFormat.setInputPaths(job,new Path("D:\\hadoop_data\\input"));

        FileOutputFormat.setOutputPath(job,new Path("D:\\hadoop_data\\output"));

        //7.提交job

        boolean result = job.waitForCompletion(true);

        System.exit(result ? 0 : 1);

    }

}

测试

本地测试

执行结束

巴特西

MapReduce02 序列化