简单的MapReduce例子，从一批数据中获取每年的最高温度

样例数据只有年份和温度，用MapReduce来取每年的最高温度。

/* MaxTemperature.java */

import java.io.IOException;

import org.apache.hadoop.fs.Path;

import org.apache.hadoop.io.*;

import org.apache.hadoop.mapreduce.*;

import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;

import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MaxTemperature {

static class MaxTemperatureMapper extends Mapper<LongWritable, Text, Text, IntWritable> {

public void map(LongWritable key, Text value, Context context)

throws IOException, InterruptedException {

String line = value.toString();

// 从每行的数据中分解出年和温度

String year = line.substring(0, 4);

int airTemperature = Integer.parseInt(line.substring(5, 8));

context.write(new Text(year), new IntWritable(airTemperature));

}

}

static class MaxTemperatureReducer extends Reducer<Text, IntWritable, Text, IntWritable> {

public void reduce(Text key, Iterable<IntWritable> values, Context context)

throws IOException, InterruptedException {

int maxValue = Integer.MIN_VALUE;

// key为年份, values为该年每天的温度，求出最大的温度值

for (IntWritable value : values) {

maxValue = Math.max(maxValue, value.get());

}

context.write(key, new IntWritable(maxValue));

}

}

public static void main(String[] args) throws Exception {

if(args.length !=2) {

System.err.println("Usage: MaxTemperature <input path> <output path>");

System.exit(-1);

}

Job job = new Job();

job.setJarByClass(MaxTemperature.class);

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

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

job.setMapperClass(MaxTemperatureMapper.class);

job.setReducerClass(MaxTemperatureReducer.class);

job.setOutputKeyClass(Text.class);

job.setOutputValueClass(IntWritable.class);

System.exit(job.waitForCompletion(true)?0:1);

}

}

打包在hadoop环境下运行，结果的输出会在output的目录下。

hadoop jar MaxTemperature.jar MaxTemperature sample.txt output