• 我们使用之前搭建好的Hadoop环境，可参见：

《【Hadoop环境搭建】Centos6.8搭建hadoop伪分布模式》    

• 示例程序为《Hadoop权威指南3》中的获取最高温度的示例程序；

---

数据准备

输入数据为：sample.txt

1. 0067011990999991950051507004+68750+023550FM-12+038299999V0203301N00671220001CN9999999N9+00001+99999999999
2. 0043011990999991950051512004+68750+023550FM-12+038299999V0203201N00671220001CN9999999N9+00221+99999999999
3. 0043011990999991950051518004+68750+023550FM-12+038299999V0203201N00261220001CN9999999N9-00111+99999999999
4. 0043012650999991949032412004+62300+010750FM-12+048599999V0202701N00461220001CN0500001N9+01111+99999999999
5. 0043012650999991949032418004+62300+010750FM-12+048599999V0202701N00461220001CN0500001N9+00781+99999999999

---

将samle.txt上传至HDFS

1. hadoop fs -put /home/hadoop/ncdcData/sample.txt input

---

项目结构

---

MaxTemperatureMapper类

1. package com.ll.maxTemperature;
3. import java.io.IOException;
5. import org.apache.hadoop.io.IntWritable;
6. import org.apache.hadoop.io.LongWritable;
7. import org.apache.hadoop.io.Text;
8. import org.apache.hadoop.mapreduce.Mapper;
10. public class MaxTemperatureMapper extends
11. Mapper<LongWritable, Text, Text, IntWritable> {
13. private static final int MISSING = 9999;
15. @Override
16. public void map(LongWritable key, Text value, Context context)
17. throws IOException, InterruptedException {
18. String line = value.toString();
19. String year = line.substring(15, 19);
20. int airTemperature;
21. if (line.charAt(87) == '+') {
    // parseInt doesn't like leading plus
22. // signs
23. airTemperature = Integer.parseInt(line.substring(88, 92));
24. } else {
25. airTemperature = Integer.parseInt(line.substring(87, 92));
26. }
27. String quality = line.substring(92, 93);
28. if (airTemperature != MISSING && quality.matches("[01459]")) {
29. context.write(new Text(year), new IntWritable(airTemperature));
30. }
31. }
32. }
33. // ^^ MaxTemperatureMapper

---

MaxTemperatureReducer类

1. package com.ll.maxTemperature;
3. import java.io.IOException;
5. import org.apache.hadoop.io.IntWritable;
6. import org.apache.hadoop.io.Text;
7. import org.apache.hadoop.mapreduce.Reducer;
9. public class MaxTemperatureReducer extends
10. Reducer<Text, IntWritable, Text, IntWritable> {
12. @Override
13. public void reduce(Text key, Iterable<IntWritable> values, Context context)
14. throws IOException, InterruptedException {
15. int maxValue = Integer.MIN_VALUE;
16. for (IntWritable value : values) {
17. maxValue = Math.max(maxValue, value.get());
18. }
19. context.write(key, new IntWritable(maxValue));
20. }
21. }
22. // ^^ MaxTemperatureReducer

---

MaxTemperature类（主函数）

1. package com.ll.maxTemperature;
3. import org.apache.hadoop.fs.Path;
4. import org.apache.hadoop.io.IntWritable;
5. import org.apache.hadoop.io.Text;
6. import org.apache.hadoop.mapreduce.Job;
7. import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
8. import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
10. public class MaxTemperature {
12. public static void main(String[] args) throws Exception {
13. if (args.length != 2) {
14. args = new String[] {
15. "hdfs://localhost:9000/user/hadoop/input/sample.txt",
16. "hdfs://localhost:9000/user/hadoop/out2" };
17. }
19. Job job = new Job(); // 指定作业执行规范
20. job.setJarByClass(MaxTemperature.class);
21. job.setJobName("Max temperature");
23. FileInputFormat.addInputPath(job, new Path(args[0]));
24. FileOutputFormat.setOutputPath(job, new Path(args[1])); // Reduce函数输出文件的写入路径
26. job.setMapperClass(MaxTemperatureMapper.class);
27. job.setCombinerClass(MaxTemperatureReducer.class);
28. job.setReducerClass(MaxTemperatureReducer.class);
30. job.setOutputKeyClass(Text.class);
31. job.setOutputValueClass(IntWritable.class);
33. System.exit(job.waitForCompletion(true) ? 0 : 1);
34. }
35. }
36. // ^^ MaxTemperature

解释说明：

输入路径为：hdfs://localhost:9000/user/hadoop/input/sample.txt

这部分由两部分组成：

1. hdfs://localhost:9000/；
2. /user/hadoop/input/sample.txt

其中

hdfs://localhost:9000/

由文件core-size.xml进行设置：

其中

/user/hadoop/input/sample.txt

就是上面准备数据时sample.txt存放的路径：

输出路径为：

hdfs://localhost:9000/user/hadoop/out2

需要注意的是，在执行MapReduce时，这个输出路径一定不要存在，否则会出错。

---

pom.xml

1. <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
2. xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
3. <modelVersion>4.0.0</modelVersion>
5. <groupId>com.ll</groupId>
6. <artifactId>MapReduceTest</artifactId>
7. <version>0.0.1-SNAPSHOT</version>
8. <packaging>jar</packaging>
10. <name>MapReduceTest</name>
11. <url>http://maven.apache.org</url>
13. <properties>
14. <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
15. <hadoopVersion>1.2.1</hadoopVersion>
16. <junit.version>3.8.1</junit.version>
17. </properties>
19. <dependencies>
20. <dependency>
21. <groupId>junit</groupId>
22. <artifactId>junit</artifactId>
23. <version>${
    junit.version}</version>
24. <scope>test</scope>
25. </dependency>
26. <!-- Hadoop -->
27. <dependency>
28. <groupId>org.apache.hadoop</groupId>
29. <artifactId>hadoop-core</artifactId>
30. <version>${
    hadoopVersion}</version>
31. <!-- Hadoop -->
32. </dependency>
33. </dependencies>
34. </project>

---

程序测试

Hadoop环境准备

我们使用之前搭建好的Hadoop环境，可参见：

《【Hadoop环境搭建】Centos6.8搭建hadoop伪分布模式》

 

---

生成jar包

下面是生成jar包过程

---

上传服务器并运行测试

使用默认的输入输出路径：

1. hadoop jar mc.jar

指定输入输出路径：

1. hadoop jar /home/hadoop/jars/mc.jar hdfs://localhost:9000/user/hadoop/input/sample.txt hdfs://localhost:9000/user/hadoop/out5