Hadoop序列化
序列化概述
序列化就是把内存中的对象、转换成字节系列（或者其他数据传输协议）以便于存储到磁盘（持久化）和网络传输。
反序列化就是将收到字节序列（或其他数据传输协议）或者是磁盘的持久化数据，转换成内存中的对象。
1、JAVA序列化和hadoop序列化
java序列化：java序列化时一个重量级序列化框架，一个对象对序列化后，会附带很多额外的信息（包括校验信息，Header、继承体系），不便于在网络中高效传输。
Hadoop序列化特点
- 紧凑：高效使用存储空间
- 快速：读写数据的额外空间小
- 可拓展：随着通信协议的升级而升级
- 互操作：支持多语言的交互
hadoop的序列化 Writable
1)自定的类需要实现Writable接口
2）提高无参数的构造器（反序列化时通过反射的方式调用无参数构造器来创建对象）
3）重写write方法实现序列化的过程（自定义序列化的内容）
4）重写readFields方法实现反序列化过程
5）如果自定义的类需要作为输出的key或者value来使用的化，一般建议重写ToString方法，因为hadoop会默认调用对象的toString方法进行输出
Haoop序列化实例
在MapReduce之wordcount实例中已经对源码进行分析，这里不对源码继续分析
1、新建一个类进行封装手机号
FlowBean.java
package Wtritable;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
/*
* 用于封装一个手机号的 上行流量 下行流量 总流量
* 因为flowBean要作为value在mr过程中传入，且会落盘，因此需要支持序列化和反序列化
*
* */
public class FlowBean implements Writable {
private Long upFlow;// 设置上行流量
private Long downFlow;//设置下行流量
private Long sumFlow;// 设置总流量
public FlowBean(){}
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.getUpFlow()+this.getDownFlow();
}
// 序列化方法
public void write(DataOutput dataOutput) throws IOException {
dataOutput.writeLong(upFlow);
dataOutput.writeLong(downFlow);
dataOutput.writeLong(sumFlow);
}
// 反序列化方法
/* 注意：反序列融化读取数据孙粗需要与反序列化写入数据的顺序一致
* */
public void readFields(DataInput dataInput) throws IOException {
this.upFlow= dataInput.readLong();
this.downFlow=dataInput.readLong();
this.sumFlow=dataInput.readLong();
}
@Override
public String toString() {
return this.upFlow+"\t"+this.downFlow+'\t'+this.sumFlow;
}
}
2、重写Mapper类
FlowMapper.java
package Wtritable;
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> {
// 定义写出的key
private Text outk=new Text();
// 定义写出的value
private FlowBean outv=new FlowBean();
@Override
protected void map(LongWritable key, Text vale,Context context) throws IOException, InterruptedException {
//1、将读取的一行数据转回String
// 数据格式 12345567788 234 342324 200
String line=vale.toString();
//2、切割数据
String[] flowMsg=line.split("\t");
//3、封装数据
outk.set(flowMsg[1]);
// 4、封装value
outv.setUpFlow(Long.parseLong(flowMsg[flowMsg.length-3]));
outv.setDownFlow(Long.parseLong(flowMsg[flowMsg.length-2]));
outv.setSumFlow();
//5、写出
context.write(outk,outv);
}
}
3、重写Reudcer类
FlowReudecr.java
package Wtritable;
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> {
//写出的value
FlowBean outv=new FlowBean();
@Override
protected void reduce(Text key, Iterable<FlowBean> values, Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {
long totalUpFlow=0;
long totalDownFlow=0;
long totalSumFlow=0;
//1、迭代values，将相同的key和values汇总到一起
for(FlowBean bean:values){
totalUpFlow+=bean.getUpFlow();
totalDownFlow+=bean.getDownFlow();
totalSumFlow+=bean.getSumFlow();
}
//2、封装value
outv.setUpFlow(totalUpFlow);
outv.setDownFlow(totalDownFlow);
outv.setSumFlow();
//3、写出
context.write(key,outv);
}
}
4、执行代码
FlowDriver.java内容
package Wtritable;
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、创建配置环境
Configuration conf=new Configuration();
// 2、创建Job环境
Job job=Job.getInstance(conf);
//3、关联驱动类
job.setJarByClass(FlowDriver.class);
//4、关联Mapper和Reducer类
job.setMapperClass(FlowMapper.class);
job.setReducerClass(FlowReducer.class);
//5、设置map输出的k和v类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(FlowBean.class);
// 6、设置最终输出的k和v的类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(FlowBean.class);
//7、设置输入和输出路径
FileInputFormat.setInputPaths(job,new Path("D:\\踩着上帝的小丑\\大数据测试数据\\a"));
FileOutputFormat.setOutputPath(job,new Path("D:\\踩着上帝的小丑\\大数据测试数据\\b")); //自己的路径
// 8、提交数据
job.waitForCompletion(true);
}
}