使用MapReduce统计最高气温全过程

开发环境：Win10，IEDA，CentOS7，Hadoop2.7.7

我们使用的数据来自美国国家气候数据中心。
数据集下载地址:
https://github.com/tomwhite/hadoop-book/tree/master/input/ncdc/all

数据格式：

气象数据

其中红框内的数据是我们需要数据。第一个是年份，第二个是气温（源文件中的气温被放大了十倍）。

数据有了下一步就是敲代码了，具体思路就是以年份为key，气温为value。
首先新建Maven项目，导入hadoop的依赖：

<properties>
    <hadoopVersion>2.7.7</hadoopVersion>
</properties>
<dependencies>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>${hadoopVersion}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-hdfs</artifactId>
        <version>${hadoopVersion}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-mapreduce-client-core</artifactId>
        <version>${hadoopVersion}</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-client</artifactId>
        <version>${hadoopVersion}</version>
    </dependency>
</dependencies>

然后，新建包com.mapreduce:

新建mapper类MaxTempMapper:

package com.mapreduce;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;

public class MaxTempMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
    @Override
    protected void map(LongWritable key, Text value, Context context) 
        throws IOException, InterruptedException {
        // 将 value 变为 String 格式
        String line = value.toString();
        // 获得年份
        String year = line.substring(15, 19);
        // 获得温度
        int temp = Integer.parseInt(line.substring(87, 92));

        // 存在脏数据 9999，所以要将其过滤
        if (temp != 9999) {
            // 输出年份与温度
            context.write(new Text(year), new IntWritable(temp));
        }

    }
}

reducer类MaxTempReducer:

package com.mapreduce;

import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

public class MaxTempReducer extends Reducer<Text, IntWritable, Text, DoubleWritable> {
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        Integer max = Integer.MIN_VALUE;

        // 得到最大值
        for (IntWritable value : values) {
            max = Math.max(max, value.get());
        }

        // 输出年份与最大温度
        context.write(key, new DoubleWritable(max / 10.0));
    }
}

然后编写Main函数：

package com.mapreduce;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

public class MaxTempApp {
    public static void main(String[] args) throws Exception {
        // 初始化配置文件
        Configuration conf = new Configuration();

        // 仅在本地开发时使用
        //conf.set("fs.defaultFS", "file:///");

        // 初始化文件系统
        FileSystem fs = FileSystem.get(conf);

        // 通过配置文件初始化 job
        Job job = Job.getInstance(conf);

        // 设置 job 名称
        job.setJobName("max Temp");

        // job 入口函数类
        job.setJarByClass(MaxTempApp.class);

        // 设置 mapper 类
        job.setMapperClass(MaxTempMapper.class);

        // 设置 reducer 类
        job.setReducerClass(MaxTempReducer.class);

        // 设置 map 的输出 K-V 类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        // 设置 reduce 的输出 K-V 类型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(DoubleWritable.class);

        // 新建输入输出路径
        Path pin = new Path(args[0]);
        //Path pin2 = new Path("C:/Users/lee/Desktop/hadoop/in/1902");
        Path pout = new Path(args[1]);

        // 打包后自定义输入输出路径
//        Path pin = new Path(args[0]);
//        Path pout = new Path(args[1]);

        // 设置输入路径和输出路径
        FileInputFormat.addInputPath(job, pin);
        //FileInputFormat.addInputPath(job, pin2);
        FileOutputFormat.setOutputPath(job, pout);

        // 判断输出路径是否已经存在，若存在则删除
        if (fs.exists(pout)) {
            fs.delete(pout, true);
        }

        // 执行 job
        job.waitForCompletion(true);
    }
}

执行程序的时候报了一个错误：(null) entry in command string: null chmod 0700
百度了一下原因是windows的system32缺少hadoop.dll文件，去网上下载了一个解决问题。

执行结果：

接着就要打jar包放到集群中运行啦！
打jar包：

把数据和jar包上传到centos上，具体的hadopp集群搭建可以参考：
https://itsmely.github.io/2018/09/22/hadoop-deploy/

接着把数据文件上传到hdfs中：

hadoop fs -mkdir -p /itsme/input/
hadoop fs -mkdir -p /itsme/output/
hadoop fs -put {Linux的文件全路径} /itsme/input/

接着执行jar命令：

hadoop jar mapreduce-1.0-SNAPSHOT.jar com.mapreduce.MaxTempApp /itsme/input/ /itsme/output/

接着查看执行结果：

hadoop fs -ls /itsme/output/
hadoop fs -cat /itsme/output/part-r-00000