實現(xiàn)bean對象序列化步驟

? ? ? ? 自定義bean對象實現(xiàn)序列化接口。

1）必須實現(xiàn)Writable接口

2）反序列化時，需要反射調用空參構造函數(shù)，所以必須有空參構造

public FlowBean(){super();
}

3）重寫序列化方法

@Override
public void write(DataOutput out) throws IOException{out.writeLong(upFlow);out.writeLong(downFlow);out.wirteLong(sumFlow);
}

4）重寫反序列化方法

@override
public void readFields(DataInput in) throws IOException{upFlow = in.readLong();downFlow = in.readLong();sumFlow = in.readLong();
}

5）注意反序列化順序和序列化順序要完全一致

6）要想把結果顯示在文件中，需要重寫toString()方法，可用\t分開，方便后續(xù)使用

7）如果需要將自定義的bean放在key中傳輸，則還需要實現(xiàn)Comparable接口，因為MapReduce框中的shuffle過程要求對key必須能排序

@Override
public int compareTo(FlowBean o){return this.sumFlow > o.getSumFlow() ? -1 : 1;
}

代碼案例（統(tǒng)計流量）

FlowBean

public class FlowBean implements Writable {private long upFlow;private long downFlow;private long sumFlow;public FlowBean() {}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;}public void setSumFlow() {this.sumFlow = this.upFlow + this.downFlow;}@Overridepublic void write(DataOutput dataOutput) throws IOException {dataOutput.writeLong(upFlow);dataOutput.writeLong(downFlow);dataOutput.writeLong(sumFlow);}@Overridepublic void readFields(DataInput dataInput) throws IOException {this.upFlow = dataInput.readLong();this.downFlow = dataInput.readLong();this.sumFlow = dataInput.readLong();}@Overridepublic String toString() {return upFlow + "\t" + downFlow + "\t" + sumFlow ;}
}

FlowMapper

public class FlowMapper extends Mapper<LongWritable, Text,Text,FlowBean> {private Text outK = new Text();private FlowBean outV = new FlowBean();@Overrideprotected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, FlowBean>.Context context) throws IOException, InterruptedException {String line = value.toString();String[] split = line.split(" ");String phone = split[1];String upFlow = split[2];String downFlow = split[3];outK.set(phone);outV.setUpFlow(Long.parseLong(upFlow));outV.setDownFlow(Long.parseLong(downFlow));// outV.setSumFlow(Long.parseLong(upFlow) + Long.parseLong(downFlow));outV.setSumFlow();context.write(outK,outV);}
}

FlowReducer

public class FlowReducer extends Reducer<Text,FlowBean,Text,FlowBean> {private FlowBean outV = new FlowBean();@Overrideprotected void reduce(Text key, Iterable<FlowBean> values, Reducer<Text, FlowBean, Text, FlowBean>.Context context) throws IOException, InterruptedException {long totalUpFlow = 0;long totalDownFlow = 0;for (FlowBean value : values) {totalUpFlow += value.getUpFlow();totalDownFlow += value.getDownFlow();}outV.setUpFlow(totalUpFlow);outV.setDownFlow(totalDownFlow);outV.setSumFlow();context.write(key,outV);}
}

FlowDriver

public class FlowDriver{public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {Configuration conf = new Configuration();Job job = Job.getInstance(conf);job.setJarByClass(FlowDriver.class);job.setMapperClass(FlowMapper.class);job.setReducerClass(FlowReducer.class);job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);FileInputFormat.setInputPaths(job,new Path("Data/input/TestFlowBean"));FileOutputFormat.setOutputPath(job,new Path("Data/output/TestFlow3"));Boolean result = job.waitForCompletion(true);System.exit(result ? 0:1);}
}