🏠Hadoop序列化
👉序列化
序列化就是把内存中的对象，转换成字节序列(或其他数据传输协议)以便于存储(持久化)和网络传输。反序列化就是将收到字节序列(或其他数据传输协议)或者是硬盘的持久化数据，转换成内存中的对象。
👉为什么要序列化
一般来说，“活的”对象只生存在内存里，关机断电就没有了。而且“活的”对象只能由本地的进程使用，不能被发送到网络上的另外一台计算机。然而序列化可以存储“活的”对象，可以将“活的”对象发送到远程计算机。
👉为什么不用Java的序列化
Java的序列化是一个重量级序列化框架(Serializable)，一个对象被序列化后，会附带很多额外的信息(各种校验信息，Header，继承体系等)，不便于在网络中高效传输。所以，Hadoop自己开发了一套序列化机制(Writable)。
Hadoop序列化特点：
(1) 紧凑：高效使用存储空间 (2) 快速：读写数据的额外开销小 (3) 可扩展：随着通信协议的升级而可升级 (4) 互操作：支持多语言的交互
👉常用序列化类型
常用数据类型对应的Hadoop数据序列化类型如下：
| Java 类型 | Hadoop Writable 类型 |
| boolean | BooleanWritable |
| byte | ByteWritable |
| int | IntWritable |
| float | FloatWritable |
| long | LongWritable |
| double | DoubleWritable |
| String | Text |
| map | MapWritable |
| array | ArrayWritable |
👉自定义 bean 对象实现序列化
自定义 bean 要想实现序列化传输，必须实现 org.apache.hadoop.io.Writable 接口并且覆写 toString 方法以便将结果显示在文件中。
本文通过一个案例来演示自定义序列化类实现多指标统计。
🏠案例
本文任务的输入文件如下：
1,13736230513,10,30 2,13736230513,11,55 3,13956435636,22,66 4,13956435636,44,49 5,13966251146,15,50 6,13788413164,1432,101 8,13788413164,56,200 9,13788413164,400,210 10,13788413164,60,200 11,13966251146,69,35 12,13966251146,19,500
输入文件利用逗号分割字段内容，从左到右的字段依次代表序号、手机号码、上行流量和下行流量。
任务的目标是统计各个手机号码的总上行流量、总下行流量、最大上行流量以及最大下行流量。
本文主要用于演示，故采用本地文件系统。
👉代码
pom文件和日志输出配置以及环境相关问题可参考作者的另一篇博文《MapReduce编程模型——在idea里面邂逅CDH MapReduce》，其他代码如下
自定义序列化类：
import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; public class DataBean implements Writable { // 上行流量 private long up; // 下行流量 private long down; // 最大上行流量 long maxUp; // 最大下行流量 long maxDown; @Override public void write(DataOutput out) throws IOException { out.writeLong(up); out.writeLong(down); out.writeLong(maxUp); out.writeLong(maxDown); } @Override public void readFields(DataInput in) throws IOException { up = in.readLong(); down = in.readLong(); } @Override public String toString() { return this.up + "\t" + this.down + "\t" + this.maxUp + "\t" + this.maxDown; } public long getUp() { return up; } public void setUp(long up) { this.up = up; } public long getDown() { return down; } public void setDown(long down) { this.down = down; } public long getMaxUp() { return maxUp; } public void setMaxUp(long maxUp) { this.maxUp = maxUp; } public long getMaxDown() { return maxDown; } public void setMaxDown(long maxDown) { this.maxDown = maxDown; }
Mapper类：
import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; public class MyMapper extends Mapper<LongWritable, Text, Text, DataBean> { @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { // 分词 String[] report = value.toString().split(","); DataBean flow = new DataBean(); flow.setUp(Long.parseLong(report[2])); flow.setDown(Long.parseLong(report[3])); // 电话号码 Text phone = new Text(); phone.set(report[1]); context.write(phone, flow); } }
Reducer类如下
import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; public class MyReducer extends Reducer<Text, DataBean, Text, DataBean> { @Override protected void reduce(Text key, Iterable<DataBean> values, Context context) throws IOException, InterruptedException { // 总上行流量 long totalUp = 0; // 总下行流量 long totalDown = 0; // 最大上行流量 long maxUp = 0; // 最大下行流量 long maxDown = 0; // 统计 for (DataBean flow : values) { totalUp += flow.getUp(); totalDown += flow.getDown(); if (flow.getUp() > maxUp){ maxUp = flow.getUp(); } if (flow.getDown() > maxDown){ maxDown = flow.getDown(); } } // 最终结果 DataBean result = new DataBean(); result.setUp(totalUp); result.setDown(totalDown); result.setMaxUp(maxUp); result.setMaxDown(maxDown); context.write(key, result); } }
主程序如下：
import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FileSystem; 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 App { public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException { // 任务配置 Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "flow"); job.setMapperClass(MyMapper.class); job.setReducerClass(MyReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(DataBean.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(DataBean.class); job.setJarByClass(App.class); // 输入输出路径 Path inputPath = new Path("D:\\mrdemo\\input"); Path outputPath = new Path("D:\\mrdemo\\output"); // 输出目录若存在，则进行删除 FileSystem fileSystem = FileSystem.get(conf); if(fileSystem.exists(outputPath)){ fileSystem.delete(outputPath, true); } // 给任务设置输入输出路径 FileInputFormat.addInputPath(job, inputPath); FileOutputFormat.setOutputPath(job, outputPath); // 提交任务 boolean flag = job.waitForCompletion(true); if (flag){ System.out.println("程序运行结束！！"); } } }
程序运行后，产生的结果文件内容如下：
