尚硅谷大数据技术之Hadoop(MapReduce)第2章 Hadoop序列化

2.1 序列化概述

1) 什么是序列化

序列化就是把内存中的对象,转换成字节序列(或其他数据传输协议)以便于存储(持久化)和网络传输。 

反序列化就是将收到字节序列(或其他数据传输协议)或者是硬盘的持久化数据,转换成内存中的对象。

2) 为什么要序列化

        一般来说,“活的”对象只生存在内存里,关机断电就没有了。而且“活的”对象只能由本地的进程使用,不能被发送到网络上的另外一台计算机。 然而序列化可以存储“活的”对象,可以将“活的”对象发送到远程计算机。

3) 为什么不用Java的序列化

        Java的序列化是一个重量级序列化框架(Serializable),一个对象被序列化后,会附带很多额外的信息(各种校验信息,header,继承体系等),不便于在网络中高效传输。所以,hadoop自己开发了一套序列化机制(Writable),特点如下:

(1)紧凑:紧凑的格式能让我们充分利用网络带宽,而带宽是数据中心最稀缺的资

(2)快速:进程通信形成了分布式系统的骨架,所以需要尽量减少序列化和反序列化的性能开销,这是基本的;

(3)可扩展:协议为了满足新的需求变化,所以控制客户端和服务器过程中,需要直接引进相应的协议,这些是新协议,原序列化方式能支持新的协议报文;

(4)互操作:能支持不同语言写的客户端和服务端进行交互; 

2.2 常用数据序列化类型

常用的数据类型对应的hadoop数据序列化类型

Java类型

Hadoop Writable类型

boolean

BooleanWritable

byte

ByteWritable

int

IntWritable

float

FloatWritable

long

LongWritable

double

DoubleWritable

string

Text

map

MapWritable

array

ArrayWritable

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

1)自定义bean对象要想序列化传输,必须实现序列化接口,需要注意以下7项。

(1)必须实现Writable接口

(2)反序列化时,需要反射调用空参构造函数,所以必须有空参构造

public FlowBean() {

super();

}

(3)重写序列化方法

@Override

public void write(DataOutput out) throws IOException {

out.writeLong(upFlow);

out.writeLong(downFlow);

out.writeLong(sumFlow);

}

(4)重写反序列化方法

@Override

public void readFields(DataInput in) throws IOException {

upFlow = in.readLong();

downFlow = in.readLong();

sumFlow = in.readLong();

}

(5)注意反序列化的顺序和序列化的顺序完全一致

(6)要想把结果显示在文件中,需要重写toString(),可用”\t”分开,方便后续用。

(7)如果需要将自定义的bean放在key中传输,则还需要实现comparable接口,因为mapreduce框中的shuffle过程一定会对key进行排序。

@Override

public int compareTo(FlowBean o) {

// 倒序排列,从大到小

return this.sumFlow > o.getSumFlow() ? -1 : 1;

}

2.4 序列化案例实操

1)需求:

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

2)数据准备

http://blog.sina.com.cn/s/blog_c30a9e680102xvii.html

输入数据格式:

1363157993055 13560436666 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 18 15 1116 954 200

手机号码 上行流量 下行流量

 

输出数据格式

1356·0436666 1116       954 2070

手机号码 上行流量        下行流量 总流量

3)分析

基本思路:

Map阶段:

(1)读取一行数据,切分字段

(2)抽取手机号、上行流量、下行流量

(3)以手机号为key,bean对象为value输出,即context.write(手机号,bean);

Reduce阶段:

(1)累加上行流量和下行流量得到总流量。

(2)实现自定义的bean来封装流量信息,并将bean作为map输出的key来传输

(3)MR程序在处理数据的过程中会对数据排序(map输出的kv对传输到reduce之前,会排序),排序的依据是map输出的key

所以,我们如果要实现自己需要的排序规则,则可以考虑将排序因素放到key中,让key实现接口:WritableComparable。

然后重写key的compareTo方法。

4)编写mapreduce程序

(1)编写流量统计的bean对象

package com.atguigu.mapreduce.flowsum;

import java.io.DataInput;

import java.io.DataOutput;

import java.io.IOException;

import org.apache.hadoop.io.Writable;

 

// 1 实现writable接口

public class FlowBean implements Writable{

 

private long upFlow ;

private long downFlow;

private long sumFlow;

//2  反序列化时,需要反射调用空参构造函数,所以必须有

public FlowBean() {

super();

}

 

public FlowBean(long upFlow, long downFlow) {

super();

this.upFlow = upFlow;

this.downFlow = downFlow;

this.sumFlow = upFlow + downFlow;

}

//3  写序列化方法

@Override

public void write(DataOutput out) throws IOException {

out.writeLong(upFlow);

out.writeLong(downFlow);

out.writeLong(sumFlow);

}

//4 反序列化方法

//5 反序列化方法读顺序必须和写序列化方法的写顺序必须一致

@Override

public void readFields(DataInput in) throws IOException {

this.upFlow  = in.readLong();

this.downFlow = in.readLong();

this.sumFlow = in.readLong();

}

 

// 6 编写toString方法,方便后续打印到文本

@Override

public String toString() {

return upFlow + “\t” + downFlow + “\t” + 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;

}

}

(2)编写mapper

package com.atguigu.mapreduce.flowsum;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Mapper;

 

public class FlowCountMapper extends Mapper<LongWritable, Text, Text, FlowBean>{

FlowBean v = new FlowBean();

Text k = new Text();

@Override

protected void map(LongWritable key, Text value, Context context)

throws IOException, InterruptedException {

// 1 获取一行

String line = value.toString();

// 2 切割字段

String[] fields = line.split(“\t”);

// 3 封装对象

// 取出手机号码

String phoneNum = fields[1];

// 取出上行流量和下行流量

long upFlow = Long.parseLong(fields[fields.length – 3]);

long downFlow = Long.parseLong(fields[fields.length – 2]);

v.set(downFlow, upFlow);

// 4 写出

context.write(new Text(phoneNum), new FlowBean(upFlow, downFlow));

}

}

(3)编写reducer

package com.atguigu.mapreduce.flowsum;

import java.io.IOException;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;

 

public class FlowCountReducer extends Reducer<Text, FlowBean, Text, FlowBean> {

 

@Override

protected void reduce(Text key, Iterable<FlowBean> values, Context context)

throws IOException, InterruptedException {

 

long sum_upFlow = 0;

long sum_downFlow = 0;

 

// 1 遍历所用bean,将其中的上行流量,下行流量分别累加

for (FlowBean flowBean : values) {

sum_upFlow += flowBean.getSumFlow();

sum_downFlow += flowBean.getDownFlow();

}

 

// 2 封装对象

FlowBean resultBean = new FlowBean(sum_upFlow, sum_downFlow);

// 3 写出

context.write(key, resultBean);

}

}

(4)编写驱动

package com.atguigu.mapreduce.flowsum;

import java.io.IOException;

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;

 

public class FlowsumDriver {

 

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

// 1 获取配置信息,或者job对象实例

Configuration configuration = new Configuration();

Job job = Job.getInstance(configuration);

 

// 6 指定本程序的jar包所在的本地路径

job.setJarByClass(FlowsumDriver.class);

 

// 2 指定本业务job要使用的mapper/Reducer业务类

job.setMapperClass(FlowCountMapper.class);

job.setReducerClass(FlowCountReducer.class);

 

// 3 指定mapper输出数据的kv类型

job.setMapOutputKeyClass(Text.class);

job.setMapOutputValueClass(FlowBean.class);

 

// 4 指定最终输出的数据的kv类型

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(FlowBean.class);

// 5 指定job的输入原始文件所在目录

FileInputFormat.setInputPaths(job, new Path(args[0]));

FileOutputFormat.setOutputPath(job, new Path(args[1]));

 

// 7 将job中配置的相关参数,以及job所用的java类所在的jar包, 提交给yarn去运行

boolean result = job.waitForCompletion(true);

System.exit(result ? 0 : 1);

}

}


上一篇:
下一篇:
关于尚硅谷
教育理念
名师团队
学员心声
资源下载
视频下载
资料下载
工具下载
加入我们
招聘岗位
岗位介绍
招贤纳师
联系我们
电话:010-56253825
邮箱:info@atguigu.com
地址:北京市昌平区宏福科技园综合楼6层(北京校区)

 深圳市宝安区西部硅谷大厦B座C区一层(深圳校区)

上海市松江区谷阳北路166号大江商厦6层(上海校区)