zcat *.gz>1993.txtbin/hadoop fs -put 1993.txt /feixu/input

package com.oracle.temperature;import java.io.IOException;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Mapper;

public class AvgTemperatureMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

    private static final int MISSING = 9999;    @Override    public void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {          String line = value.toString();        String year = line.substring(15, 19);        int airTemperature;

        if (line.charAt(87) == '+') { // parseInt doesn't like leading plus signs            airTemperature = Integer.parseInt(line.substring(88, 92));        } else {            airTemperature = Integer.parseInt(line.substring(87, 92));        }

        String quality = line.substring(92, 93);        if (airTemperature != MISSING && quality.matches("[01459]")) {          context.write(new Text(year), new IntWritable(airTemperature));        }    }}

import java.io.IOException;import org.apache.hadoop.io.IntWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;

public class AvgTemperatureReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

    @Override    public void reduce(Text key, Iterable<IntWritable> values, Context context)  throws IOException, InterruptedException {        long totalNum = 0;      long totalValue = 0;     int average = 0;        for (IntWritable value : values) {         totalNum ++;            totalValue +=value.get();        }

        average = (int)(totalValue/totalNum);        context.write(key, new IntWritable(average));    }}

import org.apache.hadoop.fs.Path;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;

public class AvgTemperature {

    public static void main(String[] args) throws Exception {        if (args.length != 2) {             System.err.println("Usage: AverageTemperature <input path> <output path>");              System.exit(-1);        }

        Job job = new Job();        job.setJarByClass(AvgTemperature.class);        job.setJobName("Average temperature");

        FileInputFormat.addInputPath(job, new Path(args[0]));        FileOutputFormat.setOutputPath(job, new Path(args[1]));

        job.setMapperClass(AvgTemperatureMapper.class);        job.setReducerClass(AvgTemperatureReducer.class);        job.setOutputKeyClass(Text.class);        job.setOutputValueClass(IntWritable.class);        System.exit(job.waitForCompletion(true) ? 0 : 1);    }}

运行JOB