Hadoop自定義序列化類

Hadoop自定義序列化類1.理解Hadoop序列化

2.了解序列化特點(diǎn)

3.掌握自定義序列化類1.

Hadoop序列化

2.序列化特點(diǎn)

3.自定義序列化類1、序列化（Serialization）是指把結(jié)構(gòu)化對(duì)象轉(zhuǎn)化為字節(jié)流。2、反序列化（Deserialization）是序列化的逆過程。即把字節(jié)流轉(zhuǎn)回結(jié)構(gòu)化對(duì)象。3、Java序列化（java.io.Serializable）Hadoop序列化作用：1、序列化在分布式環(huán)境的兩大作用：進(jìn)程間通信，永久存儲(chǔ)。2、Hadoop節(jié)點(diǎn)間通信。Hadoop的序列化格式：WritableHadoop序列化1、緊湊：高效使用存儲(chǔ)空間。2、快速：讀寫數(shù)據(jù)的額外開銷小3、可擴(kuò)展：可透明地讀取老格式的數(shù)據(jù)4、互操作：支持多語言的交互序列化格式特點(diǎn)publicclassTrafficApp{ publicstaticvoidmain(String[]args)throwsIOException,ClassNotFoundException,InterruptedException{ Jobjob=Job.getInstance(newConfiguration(),TrafficApp.class.getSimpleName()); job.setJarByClass(TrafficApp.class); FileInputFormat.setInputPaths(job,args[0]); job.setMapperClass(MyMapper.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(TrafficWritable.class); job.setReducerClass(MyReduce.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(TrafficWritable.class); FileOutputFormat.setOutputPath(job,newPath(args[1])); job.waitForCompletion(true); }自定義序列化類publicstaticclassMyMapperextendsMapper<LongWritable,Text,Text,TrafficWritable>{ Textk2=newText(); TrafficWritablev2=newTrafficWritable(); @Override protectedvoidmap(LongWritablekey,Textvalue, Mapper<LongWritable,Text,Text,TrafficWritable>.Contextcontext) throwsIOException,InterruptedException{ Stringline=value.toString(); String[]splited=line.split("\t"); k2.set(splited[1]); v2.set(splited[6],splited[7],splited[8],splited[9]); context.write(k2,v2); } }自定義序列化類publicstaticclassMyReduceextendsReducer<Text,TrafficWritable,Text,TrafficWritable>{ TrafficWritablev3=newTrafficWritable(); @Override protectedvoidreduce(Textk2,Iterable<TrafficWritable>v2s, Reducer<Text,TrafficWritable,Text,TrafficWritable>.Contextcontext) throwsIOException,InterruptedException{ longt1=0L; 自定義序列化類longt2=0L; longt3=0L; longt4=0L; for(TrafficWritablev2:v2s){ t1+=v2.t1; t2+=v2.t2; t3+=v2.t3; t4+=v2.t4; } v3.set(t1,t2,t3,t4); context.write(k2,v3); } }自定義序列化類longt2=0L; longt3=0L; longt4=0L; for(TrafficWritablev2:v2s){ t1+=v2.t1; t2+=v2.t2; t3+=v2.t3; t4+=v2.t4; } v3.set(t1,t2,t3,t4); context.write(k2,v3); } }自定義序列化類staticclassTrafficWritableimplementsWritable{ longt1; longt2; longt3; longt4; publicTrafficWritable(){} publicvoidset(longt1,longt2,longt3,longt4){ this.t1=t1; this.t2=t2; this.t3=t3; this.t4=t4; }自定義序列化類publicvoidset(Stringt1,Stringt2,Stringt3,Stringt4){ this.t1=Long.parseLong(t1); this.t2=Long.parseLong(t2); this.t3=Long.parseLong(t3); this.t4=Long.parseLong(t4); } publicvoidreadFields(DataInputin)throwsIOException{ this.t1=in.readLong(); this.t2=in.readLong(); this.t3=in.readLong(); this.t4=in.readLong(); }自定義序列化類publicvoidwrite(DataOutputout)throwsIOException{ out.writeLong(t1); out.writeLong(t2); out.writeLong(t3); out.write