🏠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("程序运行结束!!");
}
}
}
程序运行后,产生的结果文件内容如下:
作者这水平有限,有不足之处欢迎留言指正!!