• MapReduce序列化【用户流量使用统计】

    目录

    什么是序列化和反序列化？

    序列化

    反序列化

    为什么要序列化？

    序列化的主要应用场景

    MapReduce实现序列化

    自定义bean对象实现Writable接口

    1.实现Writable接口

    2.无参构造

    3.重写序列化方法

    4.重写反序列化方法

    5.顺序一致

    6.重写toString

    7.实现Comparable接口

    MapReduce自定义序列化案例 

    案例

    解决思路

    Map阶段

    Reduce阶段

    Bean

    Coding

    1、编写Bean

    2、编写Mapper类

    3、编写Reducer类

    4、编写Runner类 

    运行结果

    断点设置技巧

    什么是序列化和反序列化？

    序列化

    序列化是将对象的状态信息转化为可以存储或传输的形式的过程，通常指将对象在内存中的状态信息转换为可以被存储在外部介质上的二进制流或其他格式的数据，以便在需要时可以重新读取和还原对象的状态信息。

    反序列化

    反序列化则是将存储或传输的数据重新装配成对象的过程。

    为什么要序列化？

    因为MapReduce是一个分布式计算框架，需要将数据在各个节点之间传输。而网络传输必须是二进制数据，所以不同节点之间的数据传输就需要将数据转换为二进制流进行传输，因此需要进行序列化。

    序列化的主要应用场景

    1. 对象的持久化：将对象保存到磁盘或数据库中，以便在需要时可以重新读取和还原对象的状态信息。
    2. 远程方法调用（RPC）：将对象转换为可以在网络上传输的格式，以便在不同的进程或计算机之间进行远程通信。

    3. 分布式计算：将对象转换为可以在分布式计算环境中进行传输和计算的格式，以便在不同的计算节点之间进行数据传输和计算。

    MapReduce实现序列化

    在实际开发中，基本的序列化类型往往不能满足实际开发需求，比如在Hadoop内部传递一个bean对象，Hadoop的基本序列化类型是没有这种类型的，所以这就需要我们自己去构造该序列化类型。

    自定义bean对象实现Writable接口

    Writable接口是Hadoop序列化框架的核心接口，用户可以通过实现该接口来实现自定义的序列化类型。Writable接口的实现类包括IntWritable、DoubleWritable、Text等，我们可以通过继承Writable接口来实现自定义的序列化类。

    1.实现Writable接口

    2.无参构造

    反序列化需要反射调用无参构造函数，所以必须有无参构造

    3.重写序列化方法

    4.重写反序列化方法

    5.顺序一致

    序列化和反序列化的顺序必须完全一致，也就是说序列化的顺序为（a,b,c），那么反序列化的顺序也应该为（a,b,c）

    6.重写toString

    如果需要把结果显示在文件中，需要重写toString，不然对象输出就是一个地址值

    7.实现Comparable接口

    如果需要将我们自定义的bean放在key中传输，就必须重写Comparable接口，因为MapReduce框架中的Shuffle过程要求key必须能够排序。

    MapReduce自定义序列化案例 

    案例

    统计每一个用户耗费的总上行流量、总下行流量、总流量。

    输入案例：

    1. 1363157985066	13726230503	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200
    2. 1363157995052 	13826544101	5C-0E-8B-C7-F1-E0:CMCC	120.197.40.4			4	0	264	0	200
    3. 1363157991076 	13926435656	20-10-7A-28-CC-0A:CMCC	120.196.100.99			2	4	132	1512	200
    4. 1363154400022 	13926251106	5C-0E-8B-8B-B1-50:CMCC	120.197.40.4			4	0	240	0	200
    5. 1363157993044 	18211575961	94-71-AC-CD-E6-18:CMCC-EASY	120.196.100.99	iface.qiyi.com	视频网站	15	12	1527	2106	200
    6. 1363157995074 	84138413	5C-0E-8B-8C-E8-20:7DaysInn	120.197.40.4	122.72.52.12		20	16	4116	1432	200
    7. 1363157993055 	13560439658	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			18	15	1116	954	200
    8. 1363157995033 	15920133257	5C-0E-8B-C7-BA-20:CMCC	120.197.40.4	sug.so.360.cn	信息安全	20	20	3156	2936	200
    9. 1363157983019 	13719199419	68-A1-B7-03-07-B1:CMCC-EASY	120.196.100.82			4	0	240	0	200
    10. 1363157984041 	13660577991	5C-0E-8B-92-5C-20:CMCC-EASY	120.197.40.4	s19.cnzz.com	站点统计	24	9	6960	690	200
    11. 1363157973098 	15013685858	5C-0E-8B-C7-F7-90:CMCC	120.197.40.4	rank.ie.sogou.com	搜索引擎	28	27	3659	3538	200
    12. 1363157986029 	15989002119	E8-99-C4-4E-93-E0:CMCC-EASY	120.196.100.99	www.umeng.com	站点统计	3	3	1938	180	200
    13. 1363157992093 	13560439658	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			15	9	918	4938	200
    14. 1363157986041 	13480253104	5C-0E-8B-C7-FC-80:CMCC-EASY	120.197.40.4			3	3	180	180	200
    15. 1363157984040 	13602846565	5C-0E-8B-8B-B6-00:CMCC	120.197.40.4	2052.flash2-http.qq.com	综合门户	15	12	1938	2910	200
    16. 1363157995093 	13922314466	00-FD-07-A2-EC-BA:CMCC	120.196.100.82	img.qfc.cn		12	12	3008	3720	200
    17. 1363157982040 	13502468823	5C-0A-5B-6A-0B-D4:CMCC-EASY	120.196.100.99	y0.ifengimg.com	综合门户	57	102	7335	110349	200
    18. 1363157986072 	18320173382	84-25-DB-4F-10-1A:CMCC-EASY	120.196.100.99	input.shouji.sogou.com	搜索引擎	21	18	9531	2412	200
    19. 1363157990043 	13925057413	00-1F-64-E1-E6-9A:CMCC	120.196.100.55	t3.baidu.com	搜索引擎	69	63	11058	48243	200
    20. 1363157988072 	13760778710	00-FD-07-A4-7B-08:CMCC	120.196.100.82			2	2	120	120	200
    21. 1363157985066 	13726238888	00-FD-07-A4-72-B8:CMCC	120.196.100.82	i02.c.aliimg.com		24	27	2481	24681	200
    22. 1363157993055 	13560436666	C4-17-FE-BA-DE-D9:CMCC	120.196.100.99			18	15	1116	954	200

    输入格式：

            时间戳、电话号码、基站的物理地址、访问网址的ip、网站域名、数据包、接包数、上行/传流量、下行/载流量、响应码

    输出格式：

            手机号码 上行流量 下行流量 总流量

    解决思路

    Map阶段

    1. 读取一行数据，切分字段
    2. 获取我们需要的值（手机号、上行流和下行流量）
    3. 以手机号为key，bean对象为value输出（即context.write(手机号,bean)）

    Map

    1. KEYIN：map阶段key是文本偏移量，不需要设置。
    2. VALUEIN：同样value是文本中一行的数据，我们不需要设置。
    3. KEYOUT：我们希望对相同的手机号的流量进行累加，所以KEYOUT应该是手机号。
    4. VALUEOUT：map阶段我们希望输出一个手机号（KEY）对应三个值（上行流量、下行流量和总流量）。

    Reduce阶段

    累加上行流量和下行总流量得到总流量（合并相同手机号的上行流量和下行流量，即<手机号，bean1+bean2+...>）

    1. KEYIN：reduce阶段KEYIN就是map阶段的输出KEYOUT，上面设计好了我们就不需要设置了。
    2. VALUEIN：同样VALUEIN就是map阶段的输出VALUEOUT，我们不需要设置。
    3. KEYOUT：输出手机号。
    4. VALUEOUT：输出该手机号对应的bean对象（需要重写toString）。

    Bean

    我们这里的Bean是作为输出的Value，所以不需要继承Comparable接口，仅仅需要注意是就是重写toString方法。

    Coding

    1、编写Bean

    • 我们这里的Bean不需要继承Comparable接口，因为它不作为Key，我们这里的Key是手机号，是一个字符串，它是Text序列化类型，在Hadoop中，它已经继承了Comparable接口。
    • 下面的Bean中，我们重载了setSumFlow方法，因为sumFlow并不是原始数据中存在的，而是我们我们通过获取upFlow和downFlow计算和得来的。
    • 我们重写了toString方法来满足输出格式的要求。
    1. import org.apache.hadoop.io.Writable;
    2.  
    3. import java.io.DataInput;
    4. import java.io.DataOutput;
    5. import java.io.IOException;
    6.  
    7. /**
    8.  * 1.实现Writable接口
    9.  * 2.重写序列化接口和反序列化接口
    10.  * 3.重写无参构造
    11.  * 4.重写toString方法
    12.  */
    13. public class FlowBean implements Writable {
    14.     private long upFlow;    //上行流量
    15.     private long downFlow;    //下行流量
    16.     private long sumFlow;    //总流量
    17.  
    18.     //3.无参构造
    19.     public FlowBean(){
    20.     }
    21.  
    22.     //2.1序列化方法
    23.     @Override
    24.     public void write(DataOutput dataOutput) throws IOException {
    25.         //序列化顺序无所谓,但是必须和反序列化顺序一致
    26.         dataOutput.writeLong(upFlow);
    27.         dataOutput.writeLong(downFlow);
    28.         dataOutput.writeLong(sumFlow);
    29.  
    30.     }
    31.  
    32.     //2.2反序列化方法
    33.     @Override
    34.     public void readFields(DataInput dataInput) throws IOException {
    35.         this.upFlow = dataInput.readLong();
    36.         this.downFlow = dataInput.readLong();
    37.         this.sumFlow = dataInput.readLong();
    38.     }
    39.  
    40.     public long getUpFlow() {
    41.         return upFlow;
    42.     }
    43.  
    44.     public void setUpFlow(long upFlow) {
    45.         this.upFlow = upFlow;
    46.     }
    47.  
    48.     public long getDownFlow() {
    49.         return downFlow;
    50.     }
    51.  
    52.     public void setDownFlow(long downFlow) {
    53.         this.downFlow = downFlow;
    54.     }
    55.  
    56.     public long getSumFlow() {
    57.         return sumFlow;
    58.     }
    59.  
    60.     public void setSumFlow(long sumFlow) {
    61.         this.sumFlow = sumFlow;
    62.     }
    63.     //重载setSumFlow方法
    64.     public void setSumFlow() {
    65.         this.sumFlow = this.upFlow+this.downFlow;
    66.     }
    67.  
    68.     @Override
    69.     public String toString() {
    70.         return upFlow + "\t" + downFlow + "\t" + sumFlow;
    71.     }
    72. }

    2、编写Mapper类

    1. import org.apache.hadoop.io.LongWritable;
    2. import org.apache.hadoop.io.Text;
    3. import org.apache.hadoop.mapreduce.Mapper;
    4. import org.apache.hadoop.util.StringUtils;
    5.  
    6. import java.io.IOException;
    7.  
    8. public class FlowMapper extends Mapper {
    9.     //尽可能节省内存,不要每次读取一行就新建对象
    10.     private String phone;   //手机号
    11.     private long upFlow;  //上行流量
    12.     private long downFlow;    //下行流量
    13.  
    14.     private Text outKey = new Text();
    15.     private FlowBean outValue = new FlowBean();
    16.  
    17.  
    18.     @Override
    19.     protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    20.  
    21.         //1.获取一行
    22.         //1363157991076 	13926435656	20-10-7A-28-CC-0A:CMCC	120.196.100.99			2	4	132	1512	200
    23.         String line = value.toString();
    24.  
    25.         //2.切割
    26.         String[] words =  StringUtils.split(line,'\t');
    27.  
    28.         //3.获取想要的数据
    29.         phone = words[1];
    30.         upFlow = Long.parseLong(words[words.length-3]);
    31.         downFlow = Long.parseLong(words[words.length-2]);
    32.  
    33.         //4.封装
    34.         outKey.set(phone);
    35.         outValue.setUpFlow(upFlow);
    36.         outValue.setDownFlow(downFlow);
    37.         outValue.setSumFlow();
    38.  
    39.         //5.写出
    40.         context.write(outKey,outValue);
    41.     }
    42.  
    43. }

    3、编写Reducer类

    1. import org.apache.hadoop.io.Text;
    2. import org.apache.hadoop.mapreduce.Reducer;
    3.  
    4. import java.io.IOException;
    5.  
    6. public class FlowReducer extends Reducer {
    7.  
    8.     private FlowBean outValue = new FlowBean();
    9.  
    10.     //reduce方法每次只计算相同的key,所以totalUp和totalDown必须放在reduce方法内部,否则会把所有key的上行流量和下行流量加在一起
    11.     @Override
    12.     protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException {
    13.  
    14.         //1.遍历集合,累加值
    15.         long totalUp = 0;    //上行流量
    16.         long totalDown = 0;    //下行流量
    17.  
    18.  
    19.         for (FlowBean value : values) {
    20.             totalUp += value.getUpFlow();
    21.             totalDown += value.getDownFlow();
    22.         }
    23.  
    24.         //2.封装outKey,outValue
    25.         outValue.setUpFlow(totalUp);
    26.         outValue.setDownFlow(totalDown);
    27.         outValue.setSumFlow();
    28.  
    29.         //3. 写出
    30.         context.write(key,outValue);
    31.     }
    32. }

    4、编写Runner类 

    1. import org.apache.hadoop.conf.Configuration;
    2. import org.apache.hadoop.conf.Configured;
    3. import org.apache.hadoop.fs.Path;
    4. import org.apache.hadoop.io.Text;
    5. import org.apache.hadoop.mapreduce.Job;
    6. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    7. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    8. import org.apache.hadoop.util.Tool;
    9. import org.apache.hadoop.util.ToolRunner;
    10.  
    11. public class FlowRunner extends Configured implements Tool {
    12.  
    13.     public static void main(String[] args) throws Exception {
    14.         ToolRunner.run(new Configuration(),new FlowRunner(),args);
    15.     }
    16.  
    17.     @Override
    18.     public int run(String[] args) throws Exception {
    19.         //1.获取job
    20.         Configuration conf = new Configuration();
    21.         Job job = Job.getInstance(conf, "flow compu");
    22.  
    23.         //2.配置jar包路径
    24.         job.setJarByClass(FlowRunner.class);
    25.  
    26.         //3.关联mapper和reducer
    27.         job.setMapperClass(FlowMapper.class);
    28.         job.setReducerClass(FlowReducer.class);
    29.  
    30.         //4.设置map、reduce输出的k、v类型
    31.         job.setMapOutputKeyClass(Text.class);
    32.         job.setMapOutputValueClass(FlowBean.class);
    33.         job.setOutputKeyClass(Text.class);
    34.         job.setOutputValueClass(FlowBean.class);
    35.  
    36.         //5.设置数据输入的路径
    37.         FileInputFormat.setInputPaths(job,new Path("D:\\MapReduce_Data_Test\\writable\\input1"));
    38.  
    39.         //6.设置输出路径-输出目录不可存在
    40.         FileOutputFormat.setOutputPath(job,new Path("D:\\MapReduce_Data_Test\\writable\\output1"));
    41.  
    42.         //7.提交job
    43.         return job.waitForCompletion(true) ? 0 : 1;//verbose：是否监控并打印job的信息
    44.     }
    45. }

    运行结果

    计算正确 ！

    断点设置技巧

     调试MapReduce程序的时候，我们一般把断点设置在map和reduce方法内部。

  • 相关阅读:
    设计模式-原型模式
    【机器学习】PyTorch-MNIST-手写字识别
    目标框选之单阶段与两阶段目标检测区别
    突破编程_C++_面试（STL 编程 vector ）
    Spring Data Jpa 实现批量插入或更新
    Day51｜动态规划part12：309.最佳买卖股票时机含冷冻期、714.买卖股票的最佳时机含手续费
    如何做好银行统一报送系统UI设计
    Python安装pycrypto出错处理方法
    Android手机投屏神器scrcpy
    登录框Tricks
  • 原文地址：https://blog.csdn.net/m0_64261982/article/details/130643504