Haddop 计算一年最低气温

从http://www1.ncdc.noaa.gov/pub/data/noaa/2001/这里下载几个数据文件，拷贝到hadoop服务器目录cd /u01/hadoop/hadoop-2.7.3/test里面，并拷贝到HDFS上。

cd /u01/hadoop/hadoop-2.7.3/test

zcat *.gz > 2001.txt

cd ..

bin/hdfs dfs -copyFromLocal test/2001.txt /tmp/2001.txt

编写代码

1、编写Mapper，从文本中读取年份和温度数据，代码如下

package com.jack.hadoop.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 MinTemperatureMapper 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) == '+') {

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

        }

	}

}

Reducer

package com.jack.hadoop.temperature;



import java.io.IOException;



import org.apache.hadoop.io.IntWritable;

import org.apache.hadoop.io.Text;

import org.apache.hadoop.mapreduce.Reducer;



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



	@Override

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

    		throws IOException, InterruptedException {

        int minValue = Integer.MAX_VALUE;

        for(IntWritable value : values) {

            minValue = Math.min(minValue, value.get());

        }



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

    }

}

Job

package com.jack.hadoop.temperature;



import java.io.IOException;



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 MinTemperature {

	

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

		Job job = new Job();



        job.setJarByClass(MinTemperature.class);



        job.setJobName("Min temperature");



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



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



        job.setMapperClass(MinTemperatureMapper.class);



        job.setReducerClass(MinTemperatureReducer.class);



        job.setOutputKeyClass(Text.class);



        job.setOutputValueClass(IntWritable.class);



        try {

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

		} catch (ClassNotFoundException e) {

			// TODO Auto-generated catch block

			e.printStackTrace();

		} catch (InterruptedException e) {

			// TODO Auto-generated catch block

			e.printStackTrace();

		}

	}



}

把所有的class文件打jar包，

在hadoop上运行程序

bin/hadoop jar test/temperature.jar com.jack.hadoop.temperature.MinTemperature /tmp/2001.txt /tmp/out

运行完成，查看运行结果

bin/hdfs dfs -cat /tmp/out/part-r-00000

2001	-115