[MapReduce_add_2] MapReduce 实现年度最高气温统计

++023450FM-+000599999V0202701N015919999999N0000001N9-+99999102001ADDGF108991999999999999999999

++023450FM-+000599999V0202901N008219999999N0000001N9-+99999102001ADDGF104991999999999999999999

++023450FM-+000599999V0209991C000019999999N0000001N9-+99999102001ADDGF108991999999999999999999

++023450FM-+000599999V0201801N008219999999N0000001N9-+99999101831ADDGF108991999999999999999999

++023450FM-+000599999V0201801N009819999999N0000001N9-+99999101761ADDGF108991999999999999999999

package hadoop.mr.maxtemp;

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;

/**

 * Mapper 类

 */

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));

        }

    }

}

package hadoop.mr.maxtemp;

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;

/**

 * Reducer 类

 */

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));

    }

}

package hadoop.mr.maxtemp;

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;

/**

 * max Temp APP

 */

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("E:/file/temp");

        Path pout = new Path("E:/test/wc/out");

        // 打包后自定义输入输出路径

//        Path pin = new Path(args[0]);

//        Path pout = new Path(args[1]);

        // 设置输入路径和输出路径

        FileInputFormat.addInputPath(job, pin);

        FileOutputFormat.setOutputPath(job, pout);

        // 判断输出路径是否已经存在，若存在则删除

        if (fs.exists(pout)) {

            fs.delete(pout, true);

        }

        // 执行 job

        job.waitForCompletion(true);

    }

}

hadoop jar myhadoop-1.0-SNAPSHOT.jar hadoop.mr.maxtemp.MaxTempApp /testdata/temp /testdata/out