代碼實現——MapReduce實現Hadoop序列化

簡單介紹

1、什么是序列化

序列化：把內存中的對象，轉換成字節序列(或其他數據傳輸協議)以便于存儲到磁盤(持久化)和網絡傳輸。
反序列化：將收到字節序列(或其他數據傳輸協議)或者是磁盤的持久化數據，轉換成內存中的對象。

2、為什么要序列化

對象的序列化(Serialization)用于將對象編碼成一個字節流，以及從字節流中重新構建對象。"將一個對象編碼成一個字節流"稱為序列化該對象(SeTializing);相反的處理過程稱為反序列化(Deserializing)。序列化有三種主要的用途：

作為一種持久化格式：一個對象被序列化以后，它的編碼可以被存儲到磁盤上，供以后反序列化用。
作為一種通信數據格式：序列化結果可以從一個正在運行的虛擬機，通過網絡被傳遞到另一個虛擬機上。
作為一種拷貝、克隆(clone)機制：將對象序列化到內存的緩存區中。然后通過反序列化，可以得到一個對已存對象進行深拷貝的新對象。

在分布式數據處理中，主要使用上面提到的前兩種功能：數據持久化和通信數據格式

需求

統計每一個手機號耗費的總上行流量、下行流量、總流量（txt文檔在/Users/lizhengi/test/input/目錄下）

1       13736230513     192.196.2.1     www.shouhu.com  2481    24681   200
2       13846544121     192.196.2.2                     264     0       200
3       13956435636     192.196.2.3                     132     1512    200
4       13966251146     192.168.2.1                     240     0       404
5       18271575951     192.168.2.2     www.shouhu.com  1527    2106    200
6       18240717138     192.168.2.3     www.hao123.com  4116    1432    200
7       13590439668     192.168.2.4                     1116    954     200
8       15910133277     192.168.2.5     www.hao123.com  3156    2936    200
9       13729199489     192.168.2.6                     240     0       200
10      13630577991     192.168.2.7     www.shouhu.com  6960    690     200
11      15043685818     192.168.2.8     www.baidu.com   3659    3538    200
12      15959002129     192.168.2.9     www.hao123.com  1938    180     500
13      13560439638     192.168.2.10                    918     4938    200
14      13470253144     192.168.2.11                    180     180     200
15      13682846555     192.168.2.12    www.qq.com      1938    2910    200
16      13992314666     192.168.2.13    www.gaga.com    3008    3720    200
17      13509468723     192.168.2.14    www.qinghua.com 7335    110349  404
18      18390173782     192.168.2.15    www.sogou.com   9531    2412    200
19      13975057813     192.168.2.16    www.baidu.com   11058   48243   200
20      13768778790     192.168.2.17                    120     120     200
21      13568436656     192.168.2.18    www.alibaba.com 2481    24681   200
22      13568436656     192.168.2.19                    1116    954     200

實現過程

1、新建Maven工程，pom.xml依賴如下

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.lizhengi</groupId><artifactId>Hadoop-API</artifactId><version>1.0-SNAPSHOT</version><dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>RELEASE</version></dependency><dependency><groupId>org.apache.logging.log4j</groupId><artifactId>log4j-core</artifactId><version>2.8.2</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>3.2.1</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>3.2.1</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-hdfs</artifactId><version>3.2.1</version></dependency></dependencies></project>

2、src/main/resources目錄下，新建一個文件，命名為“log4j.properties”，添加內容如下

log4j.rootLogger=INFO, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
log4j.appender.logfile=org.apache.log4j.FileAppender
log4j.appender.logfile.File=target/spring.log
log4j.appender.logfile.layout=org.apache.log4j.PatternLayout
log4j.appender.logfile.layout.ConversionPattern=%d %p [%c] - %m%n

3、編寫Bean類-FlowBean

package com.lizhengi.flow;import org.apache.hadoop.io.Writable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;/*** @author lizhengi* @create 2020-07-20*/
// 1 實現writable接口
public class FlowBean implements Writable {private long upFlow;    //上行流量private long downFlow;  //下行流量private long sumFlow;   //總流量//2  反序列化時，需要反射調用空參構造函數，所以必須有public FlowBean() {}@Overridepublic String toString() {return upFlow + "\t" + downFlow + "\t" + sumFlow;}public void set(long upFlow, long downFlow) {this.upFlow = upFlow;this.downFlow = downFlow;this.sumFlow = upFlow + downFlow;}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;}//3  寫序列化方法public void write(DataOutput out) throws IOException {out.writeLong(upFlow);out.writeLong(downFlow);out.writeLong(sumFlow);}//4 反序列化方法//5 反序列化方法讀順序必須和寫序列化方法的寫順序必須一致public void readFields(DataInput in) throws IOException {upFlow = in.readLong();downFlow = in.readLong();sumFlow = in.readLong();}}

4、編寫Mapper類-FlowMapper

package com.lizhengi.flow;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;import java.io.IOException;/*** @author lizhengi* @create 2020-07-20*/
public class FlowMapper extends Mapper<LongWritable, Text, Text, FlowBean> {private Text phone = new Text();private FlowBean flow = new FlowBean();@Overrideprotected void map(LongWritable key, Text value, Context context)	throws IOException, InterruptedException {String[] fields = value.toString().split("\t");phone.set(fields[1]);long upFlow = Long.parseLong(fields[fields.length - 3]);long downFlow = Long.parseLong(fields[fields.length - 2]);flow.set(upFlow,downFlow);context.write(phone, flow);}
}

5、編寫Reducer類-FlowReducer

package com.lizhengi.flow;import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;/*** @author lizhengi* @create 2020-07-20*/
public class FlowReducer extends Reducer<Text, FlowBean, Text, FlowBean> {private FlowBean sunFlow = new FlowBean();@Overrideprotected void reduce(Text key, Iterable<FlowBean> values, Context context)throws IOException, InterruptedException {long sum_upFlow = 0;long sum_downFlow = 0;// 1 遍歷所用bean，將其中的上行流量，下行流量分別累加for (FlowBean value : values) {sum_upFlow += value.getUpFlow();sum_downFlow += value.getDownFlow();}// 2 封裝對象sunFlow.set(sum_upFlow, sum_downFlow);// 3 寫出context.write(key, sunFlow);}
}

6、編寫Drvier類-FlowDriver

package com.lizhengi.flow;/*** @author lizhengi* @create 2020-07-20*/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 FlowDriver {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {// 1 獲取job實例Job job = Job.getInstance(new Configuration());// 2.設置類路徑job.setJarByClass(FlowDriver.class);// 3 指定本業務job要使用的mapper/Reducer業務類job.setMapperClass(FlowMapper.class);job.setReducerClass(FlowReducer.class);// 4 指定mapper輸出數據的kv類型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(FlowBean.class);// 5 指定最終輸出的數據的kv類型job.setOutputKeyClass(Text.class);job.setOutputValueClass(FlowBean.class);// 6 指定job的輸入原始文件所在目錄FileInputFormat.setInputPaths(job, "/Users/marron27/test/input");FileOutputFormat.setOutputPath(job, new Path("/Users/marron27/test/output"));//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);}
}

結果展示

Carlota:output marron27$ pwd
/Users/marron27/test/output
Carlota:output marron27$ cat part-r-00000 
13470253144	180	180	360
13509468723	7335	110349	117684
13560439638	918	4938	5856
13568436656	3597	25635	29232
13590439668	1116	954	2070
13630577991	6960	690	7650
13682846555	1938	2910	4848
13729199489	240	0	240
13736230513	2481	24681	27162
13768778790	120	120	240
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
18240717138	4116	1432	5548
18271575951	1527	2106	3633
18390173782	9531	2412	11943