Combiner 會繼承Reducer，它是一種mr的優化，用於減少伺服器之間網路頻寬的壓力，它是在map結束後在每臺伺服器中都算出一個值，再傳到shuffle中。適合於求和等每臺伺服器算出的值對最後結果沒有影響的業務中，但是像求平均值等功能會帶來誤差所以不能使用。Combiner會在map結束後，shuffle開始前進行執行，

package com.it18zhang.day05.flow;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import
 
 org.apache.hadoop.mapreduce.Reducer;

public class FlowCountCombiner extends Reducer<Text, LongWritable, Text, LongWritable> {
  @Override
protected void reduce(Text key, Iterable<LongWritable> values,Context context)
        throws IOException, InterruptedException {
      long sum = 0;
      for
 
(LongWritable value: values){
          sum += value.get();
      }

      context.write(key, new LongWritable(sum));

}
}

package com.it18zhang.day05.flow;

import java.io.IOException;
import java.net.URI;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class FlowCountDriver {

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

          String inPath = "";
          String outPath ="";

          if(args.length == 2){
              inPath = args[0];
              outPath = args[1];

          }





         Configuration conf = new Configuration();

         if(!outPath.equals("")){
             FileSystem fs = FileSystem.get(new URI("hdfs://192.168.81.201:8020/") , conf, "centos");
                Path f = new Path(outPath);
                boolean isTrue = fs.delete(f,true);
                if(isTrue){
                    System.out.println("刪除成功");
                } 
         }
         Job job = Job.getInstance(conf);
         //設定主的環境
         job.setJarByClass(FlowCountDriver.class);

         //設定map和reduce
         job.setMapperClass(FlowCountMapper.class);
         job.setReducerClass(FlowCountReducer.class);

         //設定map的輸出型別
         job.setMapOutputKeyClass(Text.class);
         job.setMapOutputValueClass(LongWritable.class);

         //設定reduce的輸出型別
         job.setOutputKeyClass(Text.class);
         job.setOutputValueClass(LongWritable.class);


        //指定job的輸入原始檔案所在目錄
         FileInputFormat.setInputPaths(job, new Path(inPath));
       //指定job的輸出結果所在目錄
         FileOutputFormat.setOutputPath(job, new Path(outPath));

       //將job中配置的相關引數，以及job所用的java類所在的jar包，提交給yarn去執行
        /*job.submit();*/
        boolean res = job.waitForCompletion(true);
        System.exit(res?0:1);
    }
}