一、Hadoop数据序列化的数据类型
Java数据类型 => Hadoop数据类型
int IntWritable
float FloatWritable
long LongWritable
double DoubleWritable
String Text
boolean BooleanWritable
byte ByteWritable
map MapWritable
array ArrayWritable
二、Hadoop的序列化
1.什么是序列化?
在java中,序列化接口是Serializable,它下面又实现了很多的序列化接口,所以java的序列化是一个重量级的序列化框架,一个对象被java序列化之后会附带很多额外的信息(校验信息、header、继承体系等),不便于在网络中进行高效的传输,所以Hadoop开发了一套自己的序列化框架——Writable。
序列化就是把内存当中的对象,转化为字节序列以便于存储和网络传输;
反序列化是将收到的字节序列或硬盘当中的持续化数据,转换成内存中的对象。
2.序列化的理解方法(自己悟的,不对勿喷~~)
比如下面流量统计案例中,流量的封装类FlowBean实现了Writable接口,其中定义了变量upFlow、dwFlow、flowSum;
在Mapper和Reducer类中初始化封装类FlowBean时,内存会分配空间加载这些对象,而这些对象不便于在网络中高效的传输,这是封装类FlowBean中的序列化方法将这些对象转换为字节序列,方便了存储和传输;
当Mapper或Reducer需要将这些对象的字节序列写出到磁盘时,封装类FlowBean中的反序列化方法将字节序列转换为对象,然后写道磁盘中。
3.序列化特点
序列化与反序列化时分布式数据处理当中经常会出现的,比如hadoop通信是通过远程调用(rpc)实现的,这个过程就需要序列化。
特点:1)紧凑;
2)快速
3)可扩展
4)可互操作
三、Mapreduce的流量统计程序案例
1.代码
/** * @author: PrincessHug * @date: 2019/3/23, 23:38 * @Blog: https://www.cnblogs.com/HelloBigTable/ */ public class FlowBean implements Writable { private long upFlow; private long dwFlow; private long flowSum; public long getUpFlow() { return upFlow; } public void setUpFlow(long upFlow) { this.upFlow = upFlow; } public long getDwFlow() { return dwFlow; } public void setDwFlow(long dwFlow) { this.dwFlow = dwFlow; } public long getFlowSum() { return flowSum; } public void setFlowSum(long flowSum) { this.flowSum = flowSum; } public FlowBean() { } public FlowBean(long upFlow, long dwFlow) { this.upFlow = upFlow; this.dwFlow = dwFlow; this.flowSum = upFlow + dwFlow; } /** * 序列化 * @param out 输出流 * @throws IOException */ @Override public void write(DataOutput out) throws IOException { out.writeLong(upFlow); out.writeLong(dwFlow); out.writeLong(flowSum); } /** * 反序列化 * @param in * @throws IOException */ @Override public void readFields(DataInput in) throws IOException { upFlow = in.readLong(); dwFlow = in.readLong(); flowSum = in.readLong(); } @Override public String toString() { return upFlow + "\t" + dwFlow + "\t" + flowSum; } } public class FlowCountMapper extends Mapper<LongWritable, Text,Text,FlowBean> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //获取数据 String line = value.toString(); //切分数据 String[] fields = line.split("\t"); //封装数据 String phoneNum = fields[1]; long upFlow = Long.parseLong(fields[fields.length - 3]); long dwFlow = Long.parseLong(fields[fields.length - 2]); //发送数据 context.write(new Text(phoneNum),new FlowBean(upFlow,dwFlow)); } } public class FlowCountReducer extends Reducer<Text,FlowBean,Text,FlowBean> { @Override protected void reduce(Text key, Iterable<FlowBean> values, Context context) throws IOException, InterruptedException { //聚合数据 long upFlow_sum = 0; long dwFlow_sum = 0; for (FlowBean f:values){ upFlow_sum += f.getUpFlow(); dwFlow_sum += f.getDwFlow(); } //发送数据 context.write(key,new FlowBean(upFlow_sum,dwFlow_sum)); } } public class FlowPartitioner extends Partitioner<Text,FlowBean> {