hadoop-temperture(自定义value数据类型)

本实例是：用hadoop的mapreduce思想来求解每年中的最高温度和最低温度（假设都是整型的温度数据）

1.mapreduce程序

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

    }

}

2.自定义value的数据类型

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

    @Override
    public void readFields(DataInput in) throws IOException {
        // TODO Auto-generated method stub
        this.lowvalue = in.readInt();
        this.highvalue = in.readInt();

    }

    @Override
    public void write(DataOutput out) throws IOException {
        // TODO Auto-generated method stub
        out.writeInt(lowvalue);
        out.writeInt(highvalue);
    }

    @Override
    public String toString() {
        return this.lowvalue + "\t" + this.highvalue;
    }

    @Override
    public int hashCode() {
        final int prime = 31;
        int result = 1;
        result = prime * result + highvalue;
        result = prime * result + lowvalue;
        return result;
    }

    @Override
    public 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;
    }

}