大資料-Hadoop生態(12)-Hadoop序列化和原始碼追蹤
1.什麼是序列化
2.為什麼要序列化
3.為什麼不用Java的序列化
4.自定義bean物件實現序列化介面（Writable）
在企業開發中往往常用的基本序列化型別不能滿足所有需求，比如在Hadoop框架內部傳遞一個bean物件，那麼該物件就需要實現序列化介面。
具體實現bean物件序列化步驟如下7步:
1) 必須實現Writable介面
2) 反序列話時,需要反射呼叫無參構造方法,所以必須要有無參構造方法
3) 重寫序列化方法write()
4) 重寫反序列化方法readFields()
5) 反序列化的順序和序列化的順序務必一致
6) 要想把結果顯示在檔案中,需要重寫toString()方法
7) 如果將自定義的Bean放在key中傳輸,則必須要實現comparable介面.因為MapReduce的Shuffle過程要求對key必須能排序
注:Hadoop的序列化只儲存物件的指定屬性資料,而Java儲存的東西太多,顯得太重,在傳輸和磁碟儲存上,會造成資源的浪費和緊張
5. 序列化案例實操
如下檔案,統計每一個手機號耗費的總上行流量、下行流量、總流量
1 13736230513 192.196.100.1 www.baidu.com 2481 24681 200
2 13846544121 192.196.100.2 264 0 200
3 13956435636 192.196.100.3 132 1512 200
4 13966251146 192.168.100.1 240 0 404
5 18271575951 192.168.100.2 www.baidu.com 1527 2106 200
6 84188413 192.168.100.3 www.baidu.com 4116 1432 200
7 13590439668 192.168.100.4 1116 954 200
8 15910133277 192.168.100.5 www.hao123.com 3156 2936 200
9 13729199489 192.168.100.6 240 0 200
10 13630577991 192.168.100.7 www.shouhu.com 6960 690 200
11 15043685818 192.168.100.8 www.baidu.com 3659 3538 200
12 15959002129 192.168.100.9 www.baidu.com 1938 180 500
13 13560439638 192.168.100.10 918 4938 200
14 13470253144 192.168.100.11 180 180 200
15 13682846555 192.168.100.12 www.qq.com 1938 2910 200
16 13992314666 192.168.100.13 www.gaga.com 3008 3720 200
17 13509468723 192.168.100.14 www.qinghua.com 7335 110349 404
18 18390173782 192.168.100.15 www.sogou.com 9531 2412 200
19 13975057813 192.168.100.16 www.baidu.com 11058 48243 200
FlowBean
package com.nty.writable; import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; /** * author nty * date time 2018-12-08 13:39 */ //實現Writable介面 public class Flow implements Writable { private Long upflow; //上行流量 privateLong downflow; //下行流量 private Long total; //總流量 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 getTotal() { return total; } public void setTotal(Long total) { this.total = total; } /** * 快速賦值 * @param upflow * @param downflow */ public void setFlow(long upflow, long downflow){ this.upflow = upflow; this.downflow = downflow; this.total = upflow + downflow; } //序列化方法 public void write(DataOutput out) throws IOException { out.writeLong(upflow); out.writeLong(downflow); out.writeLong(total); } //反序列化方法,讀取資料的順序和序列化寫值得順序一致 public void readFields(DataInput in) throws IOException { upflow = in.readLong(); downflow = in.readLong(); total = in.readLong(); } @Override public String toString() { return upflow + "\t" + downflow + "\t" + total; } }
Mapper類
package com.nty.writable; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Mapper; import java.io.IOException; /** * author nty * date time 2018-12-10 16:44 */ //輸入的K,V型別為偏移量和本行內容,輸出的K,V型別為本行的手機號和Folw物件 public class FlowMapper extends Mapper<LongWritable, Text, Text, Flow> { private Flow flow = new Flow(); private Text phone = new Text(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { String[] fields = value.toString().split("\t"); //設定輸入的K為手機號 phone.set(fields[1]); //設定Flow,用倒取的原因是因為,有些行資料沒有訪問地址 flow.setFlow( Long.parseLong(fields[fields.length-3]), Long.parseLong(fields[fields.length-2]) ); //將map後的結果輸出給context context.write(phone,flow); } }
Reducer類
package com.nty.writable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Reducer; import java.io.IOException; /** * author nty * date time 2018-12-10 16:44 */ //Reducer的輸入K,V型別為Mapper的輸出型別,即Text,Flow,輸出K,V型別為手機號和Flow,即Text,Flow public class FlowReducer extends Reducer<Text, Flow, Text, Flow> { private Flow flow = new Flow(); @Override protected void reduce(Text key, Iterable<Flow> values, Context context) throws IOException, InterruptedException {
long upflow = 0;
long downflow = 0;
//遍歷values,計算總的下行流量和上行流量 for (Flow flow : values) { upflow += flow.getUpflow(); downflow += flow.getDownflow(); } //物件賦值 flow.setFlow(upflow, downflow); //寫出 context.write(key, flow); } }
Driver類
package com.nty.writable; 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; /** * author nty * date time 2018-12-10 16:44 */ public class FlowDriver { public static void main(String[] args) throws Exception { //獲取配置資訊和job Configuration configuration = new Configuration(); Job job = Job.getInstance(configuration); //設定jar類 job.setJarByClass(FlowDriver.class); //設定Mapper和Reducer job.setMapperClass(FlowMapper.class); job.setReducerClass(FlowReducer.class); //設定Mapper的輸出型別 job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(Flow.class); //設定Reduce的輸出型別 job.setOutputKeyClass(Text.class); job.setOutputValueClass(Flow.class); //設定檔案的輸入輸出路徑 FileInputFormat.setInputPaths(job, new Path("d:\\Hadoop_test")); FileOutputFormat.setOutputPath(job, new Path("d:\\Hadoop_test_output")); //提交 boolean res = job.waitForCompletion(true); System.exit(res ? 0 :1); } }
最終輸出結果
13470253144 180 180 360
13509468723 7335 110349 117684
13560439638 918 4938 5856
13590439668 1116 954 2070
13630577991 6960 690 7650
13682846555 1938 2910 4848
13729199489 240 0 240
13736230513 2481 24681 27162
13846544121 264 0 264
13956435636 132 1512 1644
13966251146 240 0 240
13975057813 11058 48243 59301
13992314666 3008 3720 6728
15043685818 3659 3538 7197
15910133277 3156 2936 6092
15959002129 1938 180 2118
18271575951 1527 2106 3633
18390173782 9531 2412 11943
84188413 4116 1432 5548
6.序列化執行過程(原始碼)
map方法將flow物件write出去,框架接收到flow物件,進行序列化
step into
這是key的序列化方法
再跳出看value的序列化
step into
最後value進入我們自己重寫的序列化方法
