hadoop之 mapreduce Combiner

許多mapreduce做業會受限與集羣的帶寬，所以儘可能下降map和reduce任務之間的數據傳輸是有必要的。Hadoop容許用戶針對map任務的輸出指定一個combiner函數處理map任務的輸出，並做爲reduce函數的輸入。由於combine是優化方案，因此Hadoop沒法肯定針對map輸出記錄須要調用多少次combine函數。in the other word，無論調用多少次combine函數，reducer的輸出結果都是同樣的。
The contract for the combiner function constrains the type of function that may be used。
combiner函數協議會制約可用的函數類型。舉個例子：

假設第一個map輸出以下：

(1950, 0)
(1950, 20)
(1950, 10)

第二個map輸出以下：

(1950, 25)
(1950, 15)

reduce函數被調用時，其輸入是

(1950, [0, 20, 10, 25, 15])

結果：

(1950, 25)

若是調用combine函數，像reduce函數同樣去尋找 每一個map的輸出的最大溫度。那麼輸出結果應該是：

(1950, [20, 25])

reduce 輸出結果和之前同樣。可用經過下面的表達式來講明氣溫數值的函數調用：

max(0, 20, 10, 25, 15) = max(max(0, 20, 10), max(25, 15)) = max(20, 25) = 25

並非全部函數都有這個屬性。例如，咱們計算平均氣溫，就不能使用平均函數做爲combiner。

mean(0, 20, 10, 25, 15) = 14

可是：

mean(mean(0, 20, 10), mean(25, 15)) = mean(10, 20) = 15

combiner函數不能取代reducer。但它能有效減小mapper和reducer之間的數據傳輸量。

指定一個 combiner

Job job = Job.getInstance();
            job.setJarByClass(MaxTemperatureJob.class);
            job.setJobName("max temperature");
            //方法爲何不保持一致，不是一我的寫的？
            FileInputFormat.addInputPath(job, new Path(INPUT_PATH));
            FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));

            job.setMapperClass(MaxTemperatureMapper.class);
            job.setReducerClass(MaxTemperatureReducer.class);
            //設置combiner
            job.setCombinerClass(MaxTemperatureReducer.class);
            
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            
           // job.setInputFormatClass();

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