MapReduce实战：统计不同工作年限的薪资水平

1.薪资数据集

我们要写一个薪资统计程序，统计数据来自于互联网招聘hadoop岗位的招聘网站，这些数据是按照记录方式存储的，因此非常适合使用 MapReduce 程序来统计。

2.数据格式

我们使用的数据来自互联网招聘网站，其中每一行是一条记录。

下面我们展示一行数据，其中重要的字段被突出显示。该行数据被分割成很多行以突出每个字段，但在实际文件中，这些字段被整合成一行

美团

3-5年经验 # 工作年限

15-30k # 薪资

北京

【够牛就来】hadoop高级工程

3.分析

在这里，map阶段的输入是原始数据。我们选择文本格式作为输入格式，将数据集的每一行作为文本输入。键是某一行起始位置相对于文件起始位置的偏移量，不过我们不需要这个信息，所以将其忽略。

我们的map函数很简单。由于我们只对工作年限和薪资感兴趣，所以只需要取出这两个字段数据。在本实战中，map 函数只是一个数据提取阶段，通过这种方式来提取数据，使 reducer 函数继续对它进行处理：即统计不同工作年限的薪资水平

为了全面了解 map 的工作方式，输入以下数据作为演示

美团 3-5年经验 15-30k 北京【够牛就来】hadoop高级工程...

北信源 3-5年经验 15-20k 北京 Java高级工程师（有Hadoo...

蘑菇街 3-5年经验 10-24k 杭州 hadoop开发工程师

晶赞科技 1-3年经验 10-30k 上海 hadoop研发工程师

秒针系统 3-5年经验 10-20k 北京 hadoop开发工程师

搜狐 1-3年经验 18-23k 北京大数据平台开发工程师（Hadoo...

执御 1-3年经验 8-14k 杭州 hadoop工程师

KK唱响 3-5年经验 15-30k 杭州高级hadoop开发工程师

晶赞科技 1-3年经验 12-30k 上海高级数据分析师（hadoop）

亿玛在线 3-5年经验 18-30k 北京 hadoop工程师

酷讯 1-3年经验 10-20k 北京 hadoop Engineer/...

游族网络 5-10年经验 20-35k 上海 hadoop研发工程师

易车公司 3-5年经验 15-30k 北京 hadoop工程师

爱图购 1-3年经验 8-15k 杭州 hadoop开发工程师

晶赞科技 3-5年经验 15-33k 上海 hadoop研发工程师

这些数据，以键/值对的方式作为map函数的输入，如下所示

0, 美团 3-5年经验 15-30k 北京【够牛就来】hadoop高级工程...

84, 北信源 3-5年经验 15-20k 北京 Java高级工程师（有Hadoo...

163, 蘑菇街 3-5年经验 10-24k 杭州 hadoop开发工程师

231 ,晶赞科技 1-3年经验 10-30k 上海 hadoop研发工程师

303, 秒针系统 3-5年经验 10-20k 北京 hadoop开发工程师

375, 搜狐 1-3年经验 18-23k 北京大数据平台开发工程师

461, 执御 1-3年经验 8-14k 杭州 hadoop工程师

521, KK唱响 3-5年经验 15-30k 杭州高级hadoop开发工程师

593, 晶赞科技 1-3年经验 12-30k 上海高级数据分析师（hadoop）

677, 亿玛在线 3-5年经验 18-30k 北京 hadoop工程师

774, 酷讯 1-3年经验 10-20k 北京 hadoop Engineer/...

838, 游族网络 5-10年经验 20-35k 上海 hadoop研发工程师

910, 易车公司 3-5年经验 15-30k 北京 hadoop工程师

976, 爱图购 1-3年经验 8-15k 杭州 hadoop开发工程师

1043,晶赞科技 3-5年经验 15-33k 上海 hadoop研发工程师

键(key)是文件中的偏移量，这里不需要这个信息，所以将其忽略。map 函数的功能仅限于提取工作经验和薪资，并将它们输出；map 函数的输出经由 MapReduce 框架处理后，最后发送到reduce函数。这个处理过程基于键来对键值对进行排序和分组。因此在这个示例中，reduce 函数看到的是如下输入：

1-3年经验,[12-30k, 10-30k, 8-14k, 18-23k, 10-20k, 8-15k]

3-5年经验,[15-33k, 15-30k, 18-30k, 15-30k, 10-20k, 10-24k, 15-20k, 15-30k]

5-10年经验,[20-35k]

每个工作年限后面紧跟着一系列薪资数据，reduce 函数现在要做的是遍历整个列表并统计薪资：

1-3年经验 8-30k

3-5年经验 10-33k

5-10年经验 20-35k

上面就是最终输出结果即统计不同工作年限的薪资水平

4.实现

上面已经分析完毕，下面我们就着手实现它。这里需要编写三块代码内容

1、map 函数、

2、reduce函数

3、一些用来运行作业的代码。

1、map 函数

下面我们来编写 Mapper 类，实现 map() 函数，提取工作年限和薪资

 1 /*
 2 
 3 * 提取数据
 4 
 5 */
 6 
 7 public static class SalaryMapper extends Mapper<LongWritable, Text, Text, Text>{
 8 
 9     public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
10 
11     {
12 
13         String[] valArr = value.toString().split("\s+");
14 
15         
16 
17         if(valArr.length > 2){
18 
19             /*
20 
21              * 美团 3-5年经验 15-30k 北京 【够牛就来】hadoop高级工程
22 
23              *
24 
25              * 查看原始数据，我们只需提取第二字段、第三个字段
26 
27              */
28 
29             // 工作年限
30 
31             String workAge = valArr[1];
32 
33             // 薪资
34 
35             String salary = valArr[2];
36 
37             
38 
39             context.write(new Text(workAge), new Text(salary));
40 
41         }
42 
43     }
44 
45 }

这个 Mapper 类是一个泛型类型，它有四个形参类型，分别指定 map 函数的输入键、输入值、输出键和输出值的类型。就本示例来说，输入键是一个长整数偏移量，输入值是一行文本，输出键是工作年限，输出值是薪资。Hadoop 本身提供了一套可优化网络序列化传输的基本类型，而不是使用 java 内嵌的类型。这些类型都在 org.apache.hadoop.io 包中。这里使用 LongWritable 类型（相当于 Java 的 Long 类型）、Text 类型（相当于 Java 中的 String 类型）

map() 方法的输入是一个键（key）和一个值（value），我们首先将 Text 类型的 value 转换成 Java 的 String 类型，之后使用 split方法分割数据，然后提取我们需要的值。map() 方法还提供了 Context 实例用于将输出内容进行写入。在这种情况下，我们将工作年限、薪资封装成Text对象，并将它们进行写入。

2、reduce函数

下面我们来编写 Reducer类，实现reduce函数，统计不同工作年限的薪资水平。

 1 /*
 2 
 3 * 统计数据
 4 
 5 */
 6 
 7 public static class SalaryReduce extends Reducer<Text, Text, Text, Text>{
 8 
 9     public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException{
10 
11         // 最低工薪
12 
13         int low = 0;
14 
15         // 最高薪资
16 
17         int hight = 0;
18 
19         // 计数功能
20 
21         int count = 0;
22 
23         
24 
25         for(Text value : values){
26 
27             String text = value.toString();
28 
29             /*
30 
31              * 15-30k
32 
33              * 薪资中的分割符是"-"
34 
35              */
36 
37             String[] strArr = text.split("-");
38 
39             
40 
41             int tlow = filterSalary(strArr[0]);
42 
43             int thight = filterSalary(strArr[1]);
44 
45             
46 
47             if(count == 0 || low > tlow){
48 
49                 low = tlow;
50 
51             }
52 
53             
54 
55             if(count == 0 || hight < thight){
56 
57                 hight = thight;
58 
59             }
60 
61             
62 
63             count ++;
64 
65         }
66 
67         
68 
69         context.write(key, new Text(low + "-" + hight + "k"));
70 
71     }
72 
73       
74 
75     private int filterSalary(String salary){
76 
77         String sal = Pattern.compile("[^0-9]").matcher(salary).replaceAll("");
78 
79         return Integer.parseInt(sal);
80 
81     }
82 
83 }

同样，reduce 函数也有四个形式参数类型用于指定输入和输出类型。reduce 函数的输入类型必须匹配 map 函数的输出类型：即 Text 类型和Text类型。在这种情况下，reduce 函数的输出类型也是 Text 和 Text 类型，分别是工作年限和薪资。在 map 的输出结果中，所有相同的工作年限（key）被分配到同一个reduce执行，这个薪资就是针对同一个工作年限（key），通过循环所有薪资（values）来求匹配所有薪资的薪资水平。

3、一些用来运行作业的代码

 1 public int run(String[] arg0) throws Exception {
 2 
 3     // 读取配置文件
 4 
 5     Configuration conf = new Configuration();
 6 
 7       
 8 
 9     // 如果输出目录存在，将其删除
10 
11     Path path = new Path(arg0[1]);
12 
13     FileSystem fileSystem = path.getFileSystem(conf);
14 
15     if(fileSystem.isDirectory(path)){
16 
17         fileSystem.delete(path, true);
18 
19     }
20 
21       
22 
23     // 创建Job对象
24 
25     Job job = new Job(conf,"salary");
26 
27     job.setJarByClass(Salary.class);
28 
29       
30 
31     // 设置输入路径、输出路径
32 
33     FileInputFormat.addInputPath(job, new Path(arg0[0]));
34 
35     FileOutputFormat.setOutputPath(job, new Path(arg0[1]));
36 
37       
38 
39     // 设置mapper、reduce
40 
41     job.setMapperClass(SalaryMapper.class);
42 
43     job.setReducerClass(SalaryReduce.class);
44 
45       
46 
47     // 设置mapper、reduce的输出类型
48 
49     job.setOutputKeyClass(Text.class);
50 
51     job.setOutputValueClass(Text.class);
52 
53       
54 
55     return job.waitForCompletion(true) ? 0 : 1;
56 
57 }

 1 /**
 2 
 3 * main 方法
 4 
 5 *
 6 
 7 * @param args
 8 
 9 * @throws Exception
10 
11 */
12 
13 public static void main(String[] args) throws Exception {
14 
15     // 数据输入路径和输出路径
16 
17     String[] args0 = {
18 
19             "hdfs://ljc:9000/buaa/salary/",
20 
21             "hdfs://ljc:9000/buaa/salary/out/"
22 
23     };
24 
25       
26 
27     int ec = ToolRunner.run(new Configuration(), new Salary(), args0);
28 
29       
30 
31     System.exit(ec);
32 
33 }

Configuration 类读取 Hadoop 的配置文件，如 site-core.xml、mapred-site.xml、hdfs-site.xml 等。

Job 对象指定作业执行规范，我们可以用它来控制整个作业的运行。我们在 Hadoop 集群上运行这个作业时，要把代码打包成一个JAR文件(Hadoop在集群上发布这个文件)。不必明确指定 JAR 文件的名称，在 Job 对象的 setJarByClass 方法中传递一个类即可，Hadoop 利用这个类来查找包含它的 JAR 文件，进而找到相关的 JAR 文件。

构造 Job 对象之后，需要指定输入和输出数据的路径。

调用 FileInputFormat 类的静态方法 addInputPath() 来定义输入数据的路径，这个路径可以是单个的文件、一个目录（此时，将目录下所有文件当作输入）或符合特定文件模式的一系列文件。由函数名可知，可以多次调用 addInputPath() 来实现多路径的输入。
调用 FileOutputFormat 类中的静态方法 setOutputPath() 来指定输出路径（只能有一个输出路径）。这个方法指定的是 reduce 函数输出文件的写入目录。在运行作业前该目录是不应该存在的，否则 Hadoop 会报错并拒绝运行作业。这种预防措施的目的是防止数据丢失（长时间运行的作业如果结果被意外覆盖，肯定是件可怕的事情）。
通过 setMapperClass() 和 setReducerClass() 指定 map 类型和reduce 类型。
通过setOutputKeyClass() 和 setOutputValueClass() 控制 map 和 reduce 函数的输出类型，正如本例所示，这两个输出类型一般都是相同的。如果不同，则通过 setMapOutputKeyClass()和setMapOutputValueClass()来设置 map 函数的输出类型。
输入的类型通过 InputFormat 类来控制，我们的例子中没有设置，因为使用的是默认的 TextInputFormat（文本输入格式）。
Job 中的 waitForCompletion() 方法提交作业并等待执行完成。该方法中的布尔参数是个详细标识，所以作业会把进度写到控制台。 waitForCompletion() 方法返回一个布尔值，表示执行的成（true）败（false），这个布尔值被转换成程序的退出代码 0 或者 1。

5.结果

6.其他问题

如果，您认为阅读这篇博客让您有些收获，不妨点击一下右下角的【推荐】。
如果，您希望更容易地发现我的新博客，不妨点击一下左下角的【关注我】。
如果，您对我的博客所讲述的内容有兴趣，请继续关注我的后续博客，我是【刘超★ljc】。

本文版权归作者和博客园共有，欢迎转载，但未经作者同意必须保留此段声明，且在文章页面明显位置给出原文连接，否则保留追究法律责任的权利。