package com.zhangdan.count;import java.io.IOException;
import java.util.StringTokenizer;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import com.zhangdan.count.TopTemperature.TopMap.TopReduce;public class TopTemperature {public static class TopMap extends Mapper<LongWritable,Text,Text,TemperBeansWritable>  {public void map(LongWritable key,Text value,Context context) throws IOException,InterruptedException{String line=value.toString();StringTokenizer str=new StringTokenizer(line);while(str.hasMoreElements()){String thekey=str.nextToken();int value1=Integer.parseInt(str.nextToken());int value2=Integer.parseInt(str.nextToken());//System.out.println(thekey+":"+value1+"  "+value2);context.write(new Text(thekey),new TemperBeansWritable(value1,value2));}}public static class TopReduce extends Reducer<Text,TemperBeansWritable,Text,TemperBeansWritable>{public void reduce(Text key,Iterable<TemperBeansWritable> value,Context context) throws IOException,InterruptedException{int min=100;int max=-100;for(TemperBeansWritable temper:value){if(min>temper.getLowvalue()) min=temper.getLowvalue();if(max<temper.getHighvalue()) max=temper.getHighvalue();System.out.println(key+":"+temper.getLowvalue()+"  "+temper.getHighvalue());}//System.out.println(key+":"+min+"  "+max);context.write(key, new TemperBeansWritable(min,max));}}}public static void main(String []args) throws Exception{Configuration conf =new Configuration();Job job=new Job(conf);job.setJarByClass(TemperBeansWritable.class);job.setJobName("temperture");job.setOutputKeyClass(Text.class);job.setOutputValueClass(TemperBeansWritable.class);job.setMapperClass(TopMap.class);job.setReducerClass(TopReduce.class);job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);FileInputFormat.addInputPath(job, new Path(args[0]));FileOutputFormat.setOutputPath(job, new Path(args[1]));job.waitForCompletion(true);}}

package com.zhangdan.count;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;import org.apache.hadoop.io.Writable;/*** * @author vlab 此自定义数据类型只适合value型，如果是key类，必须实现writableComparable接口*/
public class TemperBeansWritable implements Writable {private int lowvalue;private int highvalue;public TemperBeansWritable() {// TODO Auto-generated constructor stub
    }public TemperBeansWritable(int lowvalue, int highvalue) {this.lowvalue = lowvalue;this.highvalue = highvalue;}public static TemperBeansWritable read(DataInput in) throws IOException {TemperBeansWritable w = new TemperBeansWritable();w.readFields(in);return w;}public int getLowvalue() {return lowvalue;}public int getHighvalue() {return highvalue;}public void setLowvalue(int lowvalue) {this.lowvalue = lowvalue;}public void setHighvalue(int highvalue) {this.highvalue = highvalue;}@Overridepublic void readFields(DataInput in) throws IOException {// TODO Auto-generated method stubthis.lowvalue = in.readInt();this.highvalue = in.readInt();}@Overridepublic void write(DataOutput out) throws IOException {// TODO Auto-generated method stub
        out.writeInt(lowvalue);out.writeInt(highvalue);}@Overridepublic String toString() {return this.lowvalue + "\t" + this.highvalue;}@Overridepublic int hashCode() {final int prime = 31;int result = 1;result = prime * result + highvalue;result = prime * result + lowvalue;return result;}@Overridepublic boolean equals(Object obj) {if (this == obj)return true;if (obj == null)return false;if (getClass() != obj.getClass())return false;TemperBeansWritable other = (TemperBeansWritable) obj;if (highvalue != other.highvalue)return false;if (lowvalue != other.lowvalue)return false;return true;}}