MapReduce求平均值

一：背景

求平均数是MapReduce比较常见的算法，求平均数的算法也比较简单，一种思路是Map端读取数据，Reduce端汇总并且统计记录数，然后作商即可。

二：技术实现

#需求：现有成绩单如下，求出每个同学的平均成绩

[java] view plaincopy

1. 小民  语文  80
2. 小民  数学  98
3. 小民  英语  89
4. 小芳  语文  88
5. 小芳  数学  99
6. 小芳  英语  90

实现代码：

[java] view plaincopy

1. public class AverageTest {
2. // 定义输入路径
3. private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/average_file";
4. // 定义输出路径
5. private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";
6. public static void main(String[] args) {
7. try {
8. // 创建配置信息
9. Configuration conf = new Configuration();
10. // 创建文件系统
11. FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);
12. // 如果输出目录存在，我们就删除
13. if (fileSystem.exists(new Path(OUT_PATH))) {
14. fileSystem.delete(new Path(OUT_PATH), true);
15. }
16. // 创建任务
17. Job job = new Job(conf, AverageTest.class.getName());
18. //1.1   设置输入目录和设置输入数据格式化的类
19. FileInputFormat.setInputPaths(job, INPUT_PATH);
20. job.setInputFormatClass(TextInputFormat.class);
21. //1.2   设置自定义Mapper类和设置map函数输出数据的key和value的类型
22. job.setMapperClass(AverageMapper.class);
23. job.setMapOutputKeyClass(Text.class);
24. job.setMapOutputValueClass(Text.class);
25. //1.3   设置分区和reduce数量(reduce的数量，和分区的数量对应，因为分区为一个，所以reduce的数量也是一个)
26. job.setPartitionerClass(HashPartitioner.class);
27. job.setNumReduceTasks(1);
28. //1.4   排序
29. //1.5   归约
30. //2.1   Shuffle把数据从Map端拷贝到Reduce端。
31. //2.2   指定Reducer类和输出key和value的类型
32. job.setReducerClass(AverageReducer.class);
33. job.setOutputKeyClass(Text.class);
34. job.setOutputValueClass(FloatWritable.class);
35. //2.3   指定输出的路径和设置输出的格式化类
36. FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));
37. job.setOutputFormatClass(TextOutputFormat.class);
38. // 提交作业 退出
39. System.exit(job.waitForCompletion(true) ? 0 : 1);
40. } catch (Exception e) {
41. e.printStackTrace();
42. }
43. }
44. public static class AverageMapper extends Mapper<LongWritable, Text, Text, Text>{
45. //设置输出的key和value
46. private Text outKey = new Text();
47. private Text outValue = new Text();
48. @Override
49. protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
50. //获取输入的行
51. String line = value.toString();
52. //取出无效记录
53. if (line == null || line.equals("")){
54. return ;
55. }
56. //对数据进行切分
57. String[] splits = line.split("\t");
58. //截取姓名和成绩
59. String name = splits[0];
60. String score = splits[2];
61. //设置输出的Key和value
62. outKey.set(name);
63. outValue.set(score);
64. //将结果写出去
65. context.write(outKey, outValue);
66. }
67. }
68. public static class AverageReducer extends Reducer<Text, Text, Text, FloatWritable>{
69. //定义写出去的Key和value
70. private Text name = new Text();
71. private FloatWritable avg = new FloatWritable();
72. @Override
73. protected void reduce(Text key, Iterable<Text> value, Reducer<Text, Text, Text, FloatWritable>.Context context) throws IOException, InterruptedException {
74. //定义科目数量
75. int courseCount = 0;
76. //定义中成绩
77. int sum = 0;
78. //定义平均分
79. float average = 0;
80. //遍历集合求总成绩
81. for (Text val : value){
82. sum += Integer.parseInt(val.toString());
83. courseCount ++;
84. }
85. //求平均成绩
86. average = sum / courseCount;
87. //设置写出去的名字和成绩
88. name.set(key);
89. avg.set(average);
90. //把结果写出去
91. context.write(name, avg);
92. }
93. }
94. }

程序运行的结果：