热门标签
热门文章
- 1git与github区别与简介_git和github的区别
- 2Flutter、Electron、Tauri框架对比及Electron+Vue项目实战_electron tauri
- 3常见排序算法及其C语言实现
- 4操作系统学习笔记（七）：连续内存分配——内存碎片、动态分配_windows 查看是否内存碎片
- 5【Python】Python 去除图片水印_python 去水印
- 6Word2vec：skip-gram模型+Negative Sampling(负采样)代码实现_skip-gram代码
- 7rtthread系统 - 普中stm32开发板 - 数码管驱动_stm32数码管的配置
- 8一文读懂【Git 工作流】_git日志规范
- 9iOS 屏蔽系统更新描述文件更新！快把烦人的系统更新提示关掉！_otablock描述文件
- 10访问网站显示“不安全”怎么办_网站显示不安全_此网站无法提供安全连接
当前位置: article > 正文
MapReduce--实现手机流量分析_mapreduce手机流量统计
作者：小舞很执着 | 2024-07-02 20:18:37
赞
踩
mapreduce手机流量统计
1、需求
-
数据
-
需求一：统计每个手机号的上行数据包总和，下行数据包总和，上行总流量之和，下行总流量之和
-
需求二：将需求一的结果按照上行流量总和降序排序
-
需求三：将需求一的结果重新分区
- 135开头的写入一个分区
- 136开头的写入一个分区
- 137开头的写入一个分区
- 其他的在一个分区
2、分析：逻辑：设计
（1）需求一
step1：不论是写SQL还是写MR，一般都先把结果的格式先列出来
手机号 上行包总和 下行包总和 上行流量总和 下行流量总和
- 1
step2：有没有分组或者排序：决定Map输出的Key是什么
- 判断是否经过shuffle以及谁作为Map输出的key，只有Key才能利用 MR的分组和排序
- 只要出现：不同、各个、每个这些词后面的字段就是分组的字段
- 按照手机号进行分组：Map阶段输出的Key就定了，就是手机号
- 只有将手机号作为key，shuffle才会对手机号进行分组
- 没有排序：不用关心
- 定：用手机号作为Map输出的key
step3：Map输出的Value应该是什么？
- 根据结果中的列来倒推，Map输出的Value必须用什么才能得到结果
- 如果想要得到step1中的结果，还必须得到每条数据的上行包、下行包、上行流量和下行流量四个数据
- value：将这四个字段作为Value
step4：验证
-
Input
- 会将数据进行分片
- 将每一条数据变成一个Keyvalue对
-
Map
- 输入：value就是每一条数据
- 输出：
- key：手机号：第2个字段,
- value：上行包、下行包、上行流量、下行流量：第7/8/9/10四个字段
构建JavaBean
-
Shuffle
- 输入及输出
key value
110 1 1 10 10
110 2 2 20 20
119 1 1 3 3
120 1 2 3 4
|
按照key分组
|
相同手机号的所有的包和流量信息都在一个迭代器中
|
key value:JavaBean
110 {1 1 10 10,2 2 20 20}
119 1 1 3 3
120 1 2 3 4
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- Reduce
key value:JavaBean
110 3 3 30 30
119 1 1 3 3
120 1 2 3 4
- 1
- 2
- 3
- 4
- Output
13480253104 3 3 180 180 13502468823 57 102 7335 110349 13560439658 33 24 2034 5892 13600217502 37 266 2257 203704 13602846565 15 12 1938 2910 13660577991 24 9 6960 690 13719199419 4 0 240 0 13726230503 24 27 2481 24681 13760778710 2 2 120 120 13823070001 6 3 360 180 13826544101 4 0 264 0 13922314466 12 12 3008 3720 13925057413 69 63 11058 48243 13926251106 4 0 240 0 13926435656 2 4 132 1512 15013685858 28 27 3659 3538 15920133257 20 20 3156 2936 15989002119 3 3 1938 180 18211575961 15 12 1527 2106 18320173382 21 18 9531 2412 84138413 20 16 4116 1432
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
需求一是按照key，就是手机号排序的
（2）需求二
- 将需求一的结果按照上行流量总和降序排序
step1：结果长什么样？
- 与需求一的结果格式一致，但是显示按照上流流量总和进行降序排序的
step2：有没有分组或者排序？
- 没有分组
- 有排序：按照上行的流量总和进行降序排序
- key：整个需求一的结果五列作为Key，自定义数据类型
- 重写comparaTo方法：按照上行总流量降序排序
step3：Map输出的Value是什么？
- Value给定NullWritable即可
step4：验证过程
- Input：读需求一的结果文件
- Map
- 输出Key：五列，自定义数据类型
输出Value：NullWritable
- 输出Key：五列，自定义数据类型
- Shuffle
- 排序：在自定义数据类型中定义对这个类型的key排序时按照上行总流量降序排序
- Reduce
- 直接输出
- Output
（3）需求三
step1：结果长什么样
- part-r-00000
13502468823 57 102 7335 110349
13560439658 33 24 2034 5892
- 1
- 2
- part-r-00001
13600217502 37 266 2257 203704
13602846565 15 12 1938 2910
13660577991 24 9 6960 690
- 1
- 2
- 3
- part-r-00002
13719199419 4 0 240 0
13726230503 24 27 2481 24681
13760778710 2 2 120 120
- 1
- 2
- 3
- part-r-00003
13480253104 3 3 180 180
13823070001 6 3 360 180
13826544101 4 0 264 0
13922314466 12 12 3008 3720
13925057413 69 63 11058 48243
13926251106 4 0 240 0
13926435656 2 4 132 1512
15013685858 28 27 3659 3538
15920133257 20 20 3156 2936
15989002119 3 3 1938 180
18211575961 15 12 1527 2106
18320173382 21 18 9531 2412
84138413 20 16 4116 1432
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- Reduce有4个
- 自定义分区方式
Step2：Map输出的Key是什么
- 不重要了
- 需求中既没有分组也没有排序
- 实现重新分区
- 将手机号作为Key
Step3：Map输出的Value是什么？
- 其他4列作为Value
Step4：验证
- Input：读取需求1的结果
- Map：
- Key：手机号
- Value：这个手机号的其他信息，4列
- Shuffle
- 分区：根据手机号的开头做分区
- 分组：
- 排序：
- Reduce
- 直接输出即可
3、实现：
需求一实现
- 自定义数据类型
package bigdata.hanjiaxiaozhi.cn.mr.flow; import org.apache.hadoop.io.Writable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; /** * @ClassName FlowBean01 * @Description TODO Flow1程序中的自定义数据类型 * 用于封装：上行包、下行包、上行流量、下行流量 * @Date 2020/6/1 11:16 * @Create By hanjiaxiaozhi */ public class FlowBean01 implements Writable { //先定义属性 private long upPack; private long downPack; private long upFlow; private long downFlow; public FlowBean01(){ } //get and set public long getUpPack() { return upPack; } public void setUpPack(long upPack) { this.upPack = upPack; } public long getDownPack() { return downPack; } public void setDownPack(long downPack) { this.downPack = downPack; } 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; } //序列化 @Override public void write(DataOutput out) throws IOException { out.writeLong(this.upPack); out.writeLong(this.downPack); out.writeLong(this.upFlow); out.writeLong(this.downFlow); } //反序列化 @Override public void readFields(DataInput in) throws IOException { this.upPack = in.readLong(); this.downPack = in.readLong(); this.upFlow = in.readLong(); this.downFlow = in.readLong(); } //toString方法，output阶段会调用该方法写入文件 @Override public String toString() { return upPack+"\t"+downPack+"\t"+upFlow+"\t"+downFlow; } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- MR代码
package bigdata.hanjiaxiaozhi.cn.mr.flow; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import java.io.IOException; /** * @ClassName MRDriver * @Description TODO 这是MapReduce程序实现流量分析统计的需求一：统计每个手机号码的上行和下行总和 * @Date 2020/5/30 10:34 * @Create By hanjiaxiaozhi */ public class MRFlow1 extends Configured implements Tool { /** * 用于将Job的代码封装 * @param args * @return * @throws Exception */ @Override public int run(String[] args) throws Exception { //todo:1-构建一个Job Job job = Job.getInstance(this.getConf(),"flow1");//构建Job对象，调用父类的getconf获取属性的配置 job.setJarByClass(MRFlow1.class);//指定可以运行的类型 //todo:2-配置这个Job //input // job.setInputFormatClass(TextInputFormat.class);//设置输入的类的类型，默认就是TextInputFormat Path inputPath = new Path("datas/flowCase/data_flow.dat");//用程序的第一个参数做为第一个输入路径 //设置的路径可以给目录，也可以给定文件，如果给定目录，会将目录中所有文件作为输入，但是目录中不能包含子目录 TextInputFormat.setInputPaths(job,inputPath);//为当前job设置输入的路径 //map job.setMapperClass(MRMapper.class);//设置Mapper的类，需要调用对应的map方法 job.setMapOutputKeyClass(Text.class);//设置Mapper输出的key类型 job.setMapOutputValueClass(FlowBean01.class);//设置Mapper输出的value类型 //shuffle // job.setPartitionerClass(HashPartitioner.class);//自定义分区 // job.setGroupingComparatorClass(null);//自定义分组的方式 // job.setSortComparatorClass(null);//自定义排序的方式 //reduce job.setReducerClass(MRReducer.class);//设置Reduce的类，需要调用对应的reduce方法 job.setOutputKeyClass(Text.class);//设置Reduce输出的Key类型 job.setOutputValueClass(FlowBean01.class);//设置Reduce输出的Value类型 job.setNumReduceTasks(1);//设置ReduceTask的个数，默认为1 //output：输出目录默认不能提前存在 // job.setOutputFormatClass(TextOutputFormat.class);//设置输出的类，默认我诶TextOutputFormat Path outputPath = new Path("datas/output/flow/flow1/output");//用程序的第三个参数作为输出 //解决输出目录提前存在，不能运行的问题，提前将目前删掉 //构建一个HDFS的文件系统 FileSystem hdfs = FileSystem.get(this.getConf()); //判断输出目录是否存在，如果存在就删除 if(hdfs.exists(outputPath)){ hdfs.delete(outputPath,true); } TextOutputFormat.setOutputPath(job,outputPath);//为当前Job设置输出的路径 //todo:3-提交运行Job return job.waitForCompletion(true) ? 0:-1; } /** * 程序的入口，调用run方法 * @param args */ public static void main(String[] args) throws Exception { //构建一个Configuration对象，用于管理这个程序所有配置，工作会定义很多自己的配置 Configuration conf = new Configuration(); //t通过Toolruner的run方法调用当前类的run方法 int status = ToolRunner.run(conf, new MRFlow1(), args); //退出程序 System.exit(status); } /** * @ClassName MRMapper * @Description TODO 这是MapReduce模板的Map类 * 输入的KV类型：由inputformat决定，默认是TextInputFormat * 输出的KV类型：由map方法中谁作为key，谁作为Value决定 */ public static class MRMapper extends Mapper<LongWritable, Text, Text,FlowBean01> { private Text outputKey = new Text(); private FlowBean01 outputValue = new FlowBean01(); /** * 一般用于实现对数据前期的过滤、转换的操作 * * 输出新的KeyValue * * Key：手机号 * * Value：这条数据中的上行包、下行包、上行流量、下行流量 */ @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //将每一行的数据进行分割 String[] items = value.toString().split("\t"); //进行合法性判断 if(items.length >= 11){ //通过计数器，记录有多少条合法的 context.getCounter("UserGroup","legal record").increment(1L); //将手机号作为Key this.outputKey.set(items[1]); //将上行包、下行包、上行流量、下行流量作为value this.outputValue.setUpPack(Long.parseLong(items[6])); this.outputValue.setDownPack(Long.parseLong(items[7])); this.outputValue.setUpFlow(Long.parseLong(items[8])); this.outputValue.setDownFlow(Long.parseLong(items[9])); //输出 context.write(this.outputKey,this.outputValue); }else{ //通过计数器记录有多少个不合法的,计数器的结果可以在运行的日志中显示,这条代码执行一次就自增1 context.getCounter("UserGroup","illegal record").increment(1L); //如果没有11个字段，记录总共有多少个，然后直接返回，取下一条 return; } } } /** * @ClassName MRReducer * @Description TODO MapReduce模板的Reducer的类 * 输入的KV类型：由Map的输出决定，保持一致 * 输出的KV类型：由reduce方法中谁作为key，谁作为Value决定 */ public static class MRReducer extends Reducer<Text,FlowBean01,Text,FlowBean01> { private FlowBean01 outputValue = new FlowBean01(); /** * 实现reduce处理的逻辑：用于实现分组后的聚合 */ @Override protected void reduce(Text key, Iterable<FlowBean01> values, Context context) throws IOException, InterruptedException { long sumUpPack = 0; long sumDownPack = 0; long sumUpFlow = 0; long sumDownFlow = 0; //迭代取出这个手机号所有的上网记录中的上行和下行的信息，累加即可 for (FlowBean01 value : values) { sumUpPack += value.getUpPack(); sumDownPack += value.getDownPack(); sumUpFlow += value.getUpFlow(); sumDownFlow += value.getDownFlow(); } //将最后计算出的总的数据作为Value this.outputValue.setUpPack(sumUpPack); this.outputValue.setDownPack(sumDownPack); this.outputValue.setUpFlow(sumUpFlow); this.outputValue.setDownFlow(sumDownFlow); //输出结果 context.write(key,this.outputValue); } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
- 143
- 144
- 145
- 146
- 147
- 148
- 149
- 150
- 151
- 152
- 153
- 154
- 155
- 156
- 157
- 158
- 159
- 160
- 161
- 162
- 163
- 164
- 165
- 166
- 167
- 168
- 169
- 170
- 171
- 172
- 173
- 174
- 175
- 176
需求二实现：
- 自定义数据类型
package bigdata.hanjiaxiaozhi.cn.mr.flow; import org.apache.hadoop.io.Writable; import org.apache.hadoop.io.WritableComparable; import java.io.DataInput; import java.io.DataOutput; import java.io.IOException; /** * @ClassName FlowBean01 * @Description TODO 用于封装Flow01的结果 * 用于封装：手机号、上行总包、下行总包、上行总流量、下行总流量 * @Date 2020/6/1 11:16 * @Create By hanjiaxiaozhi */ public class FlowBean02 implements WritableComparable<FlowBean02> { //先定义属性 private String phone; private long upPack; private long downPack; private long upFlow; private long downFlow; public FlowBean02(){ } //get and set public String getPhone() { return phone; } public void setPhone(String phone) { this.phone = phone; } public long getUpPack() { return upPack; } public void setUpPack(long upPack) { this.upPack = upPack; } public long getDownPack() { return downPack; } public void setDownPack(long downPack) { this.downPack = downPack; } 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; } //序列化 @Override public void write(DataOutput out) throws IOException { out.writeUTF(this.phone); out.writeLong(this.upPack); out.writeLong(this.downPack); out.writeLong(this.upFlow); out.writeLong(this.downFlow); } //反序列化 @Override public void readFields(DataInput in) throws IOException { this.phone = in.readUTF(); this.upPack = in.readLong(); this.downPack = in.readLong(); this.upFlow = in.readLong(); this.downFlow = in.readLong(); } //toString方法，output阶段会调用该方法写入文件 @Override public String toString() { return phone+"\t"+upPack+"\t"+downPack+"\t"+upFlow+"\t"+downFlow; } /** * 对于这个类型在Shuffle中的排序，按照上行流量降序 * @param o * @return */ @Override public int compareTo(FlowBean02 o) { //只按照上行总流量降序排序即可 return -Long.valueOf(this.getUpFlow()).compareTo(Long.valueOf(o.getUpFlow())); } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- MR实现
package bigdata.hanjiaxiaozhi.cn.mr.flow; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.conf.Configured; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.io.LongWritable; import org.apache.hadoop.io.NullWritable; import org.apache.hadoop.io.Text; import org.apache.hadoop.mapreduce.Job; import org.apache.hadoop.mapreduce.Mapper; import org.apache.hadoop.mapreduce.Reducer; import org.apache.hadoop.mapreduce.lib.input.TextInputFormat; import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; import java.io.IOException; /** * @ClassName MRDriver * @Description TODO 基于MRFlow1的结果，对数据按照上行流量总和降序排序 * @Date 2020/5/30 10:34 * @Create By hanjiaxiaozhi */ public class MRFlow2 extends Configured implements Tool { /** * 用于将Job的代码封装 * @param args * @return * @throws Exception */ @Override public int run(String[] args) throws Exception { //todo:1-构建一个Job Job job = Job.getInstance(this.getConf(),"flow2");//构建Job对象，调用父类的getconf获取属性的配置 job.setJarByClass(MRFlow2.class);//指定可以运行的类型 //todo:2-配置这个Job //input // job.setInputFormatClass(TextInputFormat.class);//设置输入的类的类型，默认就是TextInputFormat Path inputPath = new Path("datas/output/flow/flow1/output/part-r-00000");//用程序的第一个参数做为第一个输入路径 //设置的路径可以给目录，也可以给定文件，如果给定目录，会将目录中所有文件作为输入，但是目录中不能包含子目录 TextInputFormat.setInputPaths(job,inputPath);//为当前job设置输入的路径 //map job.setMapperClass(MRMapper.class);//设置Mapper的类，需要调用对应的map方法 job.setMapOutputKeyClass(FlowBean02.class);//设置Mapper输出的key类型 job.setMapOutputValueClass(NullWritable.class);//设置Mapper输出的value类型 //shuffle // job.setPartitionerClass(HashPartitioner.class);//自定义分区 // job.setGroupingComparatorClass(null);//自定义分组的方式 // job.setSortComparatorClass(null);//自定义排序的方式 //reduce job.setReducerClass(MRReducer.class);//设置Reduce的类，需要调用对应的reduce方法 job.setOutputKeyClass(FlowBean02.class);//设置Reduce输出的Key类型 job.setOutputValueClass(NullWritable.class);//设置Reduce输出的Value类型 job.setNumReduceTasks(1);//设置ReduceTask的个数，默认为1 //output：输出目录默认不能提前存在 // job.setOutputFormatClass(TextOutputFormat.class);//设置输出的类，默认我诶TextOutputFormat Path outputPath = new Path("datas/output/flow/flow2/output");//用程序的第三个参数作为输出 //解决输出目录提前存在，不能运行的问题，提前将目前删掉 //构建一个HDFS的文件系统 FileSystem hdfs = FileSystem.get(this.getConf()); //判断输出目录是否存在，如果存在就删除 if(hdfs.exists(outputPath)){ hdfs.delete(outputPath,true); } TextOutputFormat.setOutputPath(job,outputPath);//为当前Job设置输出的路径 //todo:3-提交运行Job return job.waitForCompletion(true) ? 0:-1; } /** * 程序的入口，调用run方法 * @param args */ public static void main(String[] args) throws Exception { //构建一个Configuration对象，用于管理这个程序所有配置，工作会定义很多自己的配置 Configuration conf = new Configuration(); //t通过Toolruner的run方法调用当前类的run方法 int status = ToolRunner.run(conf, new MRFlow2(), args); //退出程序 System.exit(status); } /** * @ClassName MRMapper * @Description TODO 这是MapReduce模板的Map类 * 输入的KV类型：由inputformat决定，默认是TextInputFormat * 输出的KV类型：由map方法中谁作为key，谁作为Value决定 */ public static class MRMapper extends Mapper<LongWritable, Text, FlowBean02,NullWritable> { private FlowBean02 outputKey = new FlowBean02(); private NullWritable outputValue = NullWritable.get(); @Override protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException { //将文件中的五列封装为Key String[] items = value.toString().split("\t"); //将五列封装到Key中 this.outputKey.setPhone(items[0]); this.outputKey.setUpPack(Long.parseLong(items[1])); this.outputKey.setDownPack(Long.parseLong(items[2])); this.outputKey.setUpFlow(Long.parseLong(items[3])); this.outputKey.setDownFlow(Long.parseLong(items[4])); //输出 context.write(this.outputKey,this.outputValue); } } /** * @ClassName MRReducer * @Description TODO MapReduce模板的Reducer的类 * 输入的KV类型：由Map的输出决定，保持一致 * 输出的KV类型：由reduce方法中谁作为key，谁作为Value决定 */ public static class MRReducer extends Reducer<FlowBean02,NullWritable,FlowBean02,NullWritable> { @Override protected void reduce(FlowBean02 key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException { /** * 直接输出即可 */ for (NullWritable value : values) { context.write(key,value); } } } }
- 1
- 2
- 3
- 4
- 5
- 6
- 7
- 8
- 9
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
- 100
- 101
- 102
- 103
- 104
- 105
- 106
- 107
- 108
- 109
- 110
- 111
- 112
- 113
- 114
- 115
- 116
- 117
- 118
- 119
- 120
- 121
- 122
- 123
- 124
- 125
- 126
- 127
- 128
- 129
- 130
- 131
- 132
- 133
- 134
- 135
- 136
- 137
- 138
- 139
- 140
- 141
- 142
声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/小舞很执着/article/detail/780787
推荐阅读
相关标签
