問題描述

現在有個sequenceFile檔案裡面記錄著年份和溫度，key是年份value是溫度，找出每年的最高氣溫然後按照年份遞增排序。因為reducer預設會對key進行排序，解決辦法有兩種：第一種使用一個reducer，這樣做會造成效能問題，因為所有的key都發往了一臺機器。第二種是使用分割槽函式對年份進行分段，在每一個段是遞增排序，幾個reducer處理後的檔案拼接後在整體上也是有序的。

自定義的Mapper只需要把key-value發往Reducer即可：

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
/**
 * MaxTempMapper
 */
public class MaxTempMapper extends Mapper<IntWritable, IntWritable, IntWritable, IntWritable>{
    protected void map(IntWritable key, IntWritable value, Context context) throws IOException, InterruptedException {
        context.write(key,value);
    }
}
 

自定義的Reducer遍歷values，找出最大氣溫即可：

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;

/**
 * Reducer
 */
public class MaxTempReducer extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{
    /**
     * reduce
     */
    protected void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int max = Integer.MIN_VALUE ;
        for(IntWritable iw : values){
            max = max > iw.get() ? max : iw.get() ;
        }
        context.write(key,new IntWritable(max));
    }

    public static void main(String[] args) {
        System.out.println(Integer.MIN_VALUE);
    }
}
 

啟動執行類"MaxTempApp"除了設定job執行的基本引數之外，還要設定分割槽類函式，和取樣器：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.partition.InputSampler;
import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;

/**
 * 求每個年份最高氣溫，然後按照年份遞增全排序
 * reducer數量為3
 */
public class MaxTempApp {

    public static void main(String[] args) throws Exception{
        Configuration conf = new Configuration();//配置物件
        conf.set("fs.defaultFS","file:///");//本地檔案協議
        Job job = Job.getInstance(conf);//job物件
        job.setJobName("maxTemperatureByTotalPartition");
        job.setInputFormatClass(SequenceFileInputFormat.class);
        job.setJarByClass(MaxTempApp.class);

        //job新增輸入路徑
        FileInputFormat.setInputPaths(job,new Path(args[0]));
        //job新增輸出路徑
        FileOutputFormat.setOutputPath(job,new Path(args[1]));
        job.setMapperClass(MaxTempMapper.class);
        job.setReducerClass(MaxTempReducer.class);
        job.setMapOutputKeyClass(IntWritable.class);
        job.setMapOutputValueClass(IntWritable.class);
        job.setOutputKeyClass(IntWritable.class);
        job.setOutputValueClass(IntWritable.class);
        // job設定reducer個數
        job.setNumReduceTasks(3);

        //job設定全排序分割槽類,設定分割槽結果儲存路徑
        job.setPartitionerClass(TotalOrderPartitioner.class);
        TotalOrderPartitioner.setPartitionFile(job.getConfiguration(),new Path("/home/hadoop/seq/par.lst"));

        //job建立隨機取樣器物件，0.2樣本被採納的概率，4000是樣本數量
        InputSampler.Sampler<IntWritable,IntWritable> sampler = new InputSampler.RandomSampler<IntWritable, IntWritable>(0.2,4000);
        InputSampler.writePartitionFile(job,sampler);

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

檢視取樣器的分割槽結果，我們的key值從1970-2069 我們設定的reducer個數為3，取樣器會生成兩個值把key分成三段：
key值在2004之前,2004-2035之間,(>=2035)之後

檢視輸出結果，生成了三個檔案每個檔案在相應的key區間裡有序，由於key區間也是有序的，全排序完成。