hadoop序列化

序列化概念:　

　　序列化（Serialization）是指把结构化对象转化为字节流。

　　反序列化（Deserialization）是序列化的逆过程。即把字节流转回结构化对象。

　　Java序列化（java.io.Serializable）

Hadoop序列化的特点

　　紧凑：高效使用存储空间。

　　快速：读写数据的额外开销小

　　可扩展：可透明地读取老格式的数据

　　互操作：支持多语言的交互

Hadoop的序列化格式：

　　实现Writable接口.

hadoop序列化的作用:

　　序列化在分布式环境的两大作用：进程间通信，永久存储。

　　Hadoop节点间通信。

hadoop在运行时,为什么要使用自定义类型LongWritable,Text?java自带的类型不够使用吗?

　　java自身是有序列化和反序列化机制的,序列化和反序列化是从我们虚拟机进程中,把数据写出到进程之外,写出到外边之后,就离开了程序的范畴,我们在运行时就没法使用了.那么如果我们要把那些数据再读回来的时候我们必须还原成原来的样子,只有还原成原来的样子,我们的程序才可以去运行它.序列化就是把我们的程序或者是代码写到虚拟机之外,有可能进入到另外的虚拟机,这个也是正常的,写出到一个输出文件中,或者是写出到磁盘中,或者写到网络流中,我们的java是支持这种序列化和反序列化的,并且java实现起来特别简单,只需要实现java.io.Serializable这个接口就可以了,这个接口是一个标记接口,加上即可.

我们的数据写出去和读进来其实是很复杂的操作,为什么我们只需要一个接口就完事了?

　　因为这个接口是我们jdk默认自带的,它的这个序列化的工作虚拟机已经内置支持了,我们写代码的时候只需要告诉虚拟机我要实现序列化和反序列化即可,那么虚拟机在运行的时候,就会自动做这件事.不是说没有去做,而是说我们的虚拟机都已经内置了这种虚拟化这种功能.java的这种序列化和反序列化有什么特点?java这个东西是面向对象的,是有继承关系的,我们在写出的时候,就要保证类之间的继承体系,就要全部就写出去,而读进来的话也是需要全部都读进来,就会浪费很多的资源,小数据没关系,但是我们的hadoop处理的是海量数据,hadoop就不满意java自身带的这种序列化机制,所以就开发自己的一套序列化机制所以hadoop不愿用它.
　　hadoop开发的序列化机制的东西有一个接口Writable,和java自身带的有所不同,因为这个不是jdk内置的,需要我们去实现的.

　　　　write():序列化到一个流DataOutput中,还有一个叫readFiles():把数据从外面反序列化回来.
　　hadoop在数据序列化的时候,写出去的是基本数据类型,没有复杂的继承关系,所以开销是小的.读数据也是一些基本的数据类型.hadoop实现的序列化机制要求只对一些基本类型要进行序列化.
　　传递的是一些音视频信息使用BytesWritable来进行存储.

Hadoop自定义类型的实现:

class KpiWritable implements Writable{
	long upPackNum ;//上行数据包数
	long downPackNum ;//下行数据包数
	long upPayLoad ;//上行总流量
	long downPayLoad ;//下行总流量
	@Override
	public void write(DataOutput out) throws IOException {
		out.writeLong(upPackNum);
		out.writeLong(downPackNum);
		out.writeLong(upPayLoad);
		out.writeLong(downPayLoad);
	}
	//需要注意 按照什么顺序写出去,就按照什么顺序读进来,以为我们的数据写出去之后,是一个流,流是一个一维的.
	//就是从这个方向到那个方向.
	@Override
	public void readFields(DataInput in) throws IOException {
		this.upPackNum = in.readLong();
		this.downPackNum = in.readLong();
		this.upPayLoad = in.readLong();
		this.downPayLoad = in.readLong();
	}
	public KpiWritable() {
	}
	public KpiWritable(long upPackNum, long downPackNum, long upPayLoad,
			long downPayLoad) {
		super();
		set(upPackNum, downPackNum, upPayLoad, downPayLoad);
	}
	public void set(long upPackNum, long downPackNum, long upPayLoad,
			long downPayLoad) {
		this.upPackNum = upPackNum;
		this.downPackNum = downPackNum;
		this.upPayLoad = upPayLoad;
		this.downPayLoad = downPayLoad;
	}
	@Override
	public String toString() {
		return upPackNum + "	"+downPackNum + "	"+upPayLoad+"	"+downPayLoad;
	}
}

自定义序列化类型应用:

　　　　 public static final String INPUT_PATH = "hdfs://hadoop1:9000/kpi";
	public static final String OUT_PATH = "hdfs://hadoop1:9000/kpi_out";
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH),conf);
		if(fileSystem.isDirectory(new Path(OUT_PATH))){
			fileSystem.delete(new Path(OUT_PATH));
		}
		Job job = new Job(conf, KpiApp2.class.getSimpleName());
		job.setJarByClass(KpiApp2.class);//指定代码打包成jar包运行
		FileInputFormat.setInputPaths(job, new Path(INPUT_PATH));//指定数据的输入路径
		//对数据格式化 默认值就是TextInputFormat 可以不用写
		job.setMapperClass(MyMapper.class);//指定自定义map类
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(KpiWritable.class);
		
		//当reduce输出类型与map输出类型一致时,map输出类型可以不设置.使用reduce类型
		job.setReducerClass(MyReducer.class);//指定自定义reducer类
		job.setOutputKeyClass(Text.class);//指定reduce输出key类型
		job.setOutputValueClass(NullWritable.class);//指定reduce输出value类型
		FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));//指定输出路径
		job.waitForCompletion(true);//把作业提交给JobTracker 等待运行结束
		
	}
	/**
	 * 第一个,第二个泛型为LongWritable和Text由解析器TextInputFormat决定的.
	 * 输出类型一个是手机号,value是流量,没办法使用hadoop基本类型表示,只能自己定义类型.
	 */
	public static class MyMapper extends Mapper<LongWritable, Text, Text, KpiWritable>{
		//我们的map对应一行数据,如果我们有1T的数据,我们的Text执行多少次都不可计数,每次都创建一次就特别浪费时间
		//可以把我们的Text放到外边创建.  text对象中的数据不会被复用.
		Text k2 = new Text();
		@Override
		protected void map(LongWritable key, Text value,Context context)
				throws IOException, InterruptedException {
			String line = value.toString();//value就是输入的每一行
			String[] splited = line.split("	");//制表符分割
			String mobileNumber = splited[1];//手机号
			k2.set(mobileNumber);
			KpiWritable v2 = new KpiWritable(Long.parseLong(splited[6]), Long.parseLong(splited[7]), Long.parseLong(splited[8]), Long.parseLong(splited[9]));
			//我们在写出数据的时候,将map是写出到context中去的,我们在context.write()的时候,已经把数据处理掉了.
			context.write(k2, v2);
		}
	}
	public static class MyReducer extends Reducer<Text, KpiWritable, Text, NullWritable>{
		//经过分组之后,k2到reduce端就是唯一的.
		//v2s中包含的是相同手机号的KpiWritable对象
		//工作中关于mapreduce的算法并不复杂,代码多在数据的清洗.复杂的场景.
		KpiWritable v3 = new KpiWritable();
		@Override
		protected void reduce(Text k2, Iterable<KpiWritable> v2s,Context context)throws IOException, InterruptedException {
			long upPackNum = 0L ;//上行数据包数
			long downPackNum = 0L ;//下行数据包数
			long upPayLoad = 0L ;//上行总流量
			long downPayLoad = 0L ;//下行总流量
			for (KpiWritable kpiWritable : v2s) {
				upPackNum += kpiWritable.upPackNum ;
				downPackNum += kpiWritable.downPackNum ;
				upPayLoad += kpiWritable.upPayLoad ;
				downPayLoad += kpiWritable.downPayLoad ;
			}
			v3.set(upPackNum, downPackNum, upPayLoad, downPayLoad);
			context.write(new Text(k2.toString()+"	"+upPackNum + "	"+downPackNum + "	"+upPayLoad+"	"+downPayLoad), null);
		}
	}