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）資料準備
phone:
1363157985066 13726230503 00-FD-07-A4-72-B8:CMCC 120.196.100.82 i02.c.aliimg.com 24 27 2481 24681 200
1363157995052 13826544101 5C-0E-8B-C7-F1-E0:CMCC 120.197.40.4 4 0 264 0 200
1363157991076 13926435656 20-10-7A-28-CC-0A:CMCC 120.196.100.99 2 4 132 1512 200
1363154400022 13926251106 5C-0E-8B-8B-B1-50:CMCC 120.197.40.4 4 0 240 0 200
1363157993044 18211575961 94-71-AC-CD-E6-18:CMCC-EASY 120.196.100.99 iface.qiyi.com 視訊網站 15 12 1527 2106 200
1363157995074 84138413 5C-0E-8B-8C-E8-20:7DaysInn 120.197.40.4 122.72.52.12 20 16 4116 1432 200
1363157993055 13560439658 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 18 15 1116 954 200
1363157995033 15920133257 5C-0E-8B-C7-BA-20:CMCC 120.197.40.4 sug.so.360.cn 資訊安全 20 20 3156 2936 200
1363157983019 13719199419 68-A1-B7-03-07-B1:CMCC-EASY 120.196.100.82 4 0 240 0 200
1363157984041 13660577991 5C-0E-8B-92-5C-20:CMCC-EASY 120.197.40.4 s19.cnzz.com 站點統計 24 9 6960 690 200
1363157973098 15013685858 5C-0E-8B-C7-F7-90:CMCC 120.197.40.4 rank.ie.sogou.com 搜尋引擎 28 27 3659 3538 200
1363157986029 15989002119 E8-99-C4-4E-93-E0:CMCC-EASY 120.196.100.99 www.umeng.com 站點統計 3 3 1938 180 200
1363157992093 13560439658 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 15 9 918 4938 200
1363157986041 13480253104 5C-0E-8B-C7-FC-80:CMCC-EASY 120.197.40.4 3 3 180 180 200
1363157984040 13602846565 5C-0E-8B-8B-B6-00:CMCC 120.197.40.4 2052.flash2-http.qq.com 綜合門戶 15 12 1938 2910 200
1363157995093 13922314466 00-FD-07-A2-EC-BA:CMCC 120.196.100.82 img.qfc.cn 12 12 3008 3720 200
1363157982040 13502468823 5C-0A-5B-6A-0B-D4:CMCC-EASY 120.196.100.99 y0.ifengimg.com 綜合門戶 57 102 7335 110349 200
1363157986072 18320173382 84-25-DB-4F-10-1A:CMCC-EASY 120.196.100.99 input.shouji.sogou.com 搜尋引擎 21 18 9531 2412 200
1363157990043 13925057413 00-1F-64-E1-E6-9A:CMCC 120.196.100.55 t3.baidu.com 搜尋引擎 69 63 11058 48243 200
1363157988072 13760778710 00-FD-07-A4-7B-08:CMCC 120.196.100.82 2 2 120 120 200
1363157985066 13560436666 00-FD-07-A4-72-B8:CMCC 120.196.100.82 i02.c.aliimg.com 24 27 2481 24681 200
1363157993055 13560436666 C4-17-FE-BA-DE-D9:CMCC 120.196.100.99 18 15 1116 954 200
輸入資料格式：
| 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); } } |
本教程由尚矽谷教育大資料研究院出品，如需轉載請注明來源。
![GitHub連夜封殺！這份阿裡 10W 字内部 Java 字面試手冊到底有多強？[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)