Combiner

Combiner編程(1.5可選步驟，視情況而定！)

• 每一個map可能會產生大量的輸出，combiner的作用就是在map端對輸出先做一次合並，以減少傳輸到reducer的數據量。
• combiner最基本是實現本地key的歸並，combiner具有類似本地的reduce功能。 如果不用combiner，那么，所有的結果

　　都是reduce完成，效率會相對低下。使用combiner，先完成的map會在本地聚合，提升速度。

• 注意：Combiner的輸出是Reducer的輸入，Combiner絕不能改變最終的計算結果。所以從我的想法來看，Combiner只應該

　　用於那種Reduce的輸入key/value與輸出key/value類型完全一致，且不影響最終結果的場景。比如累加，最大值等。

package combine;

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.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 問：為什么使用Combiner？
 * 答：Combiner發生在Map端，對數據進行規約處理，數據量變小了，傳送到reduce端的數據量變小了，傳輸時間變短，作業的整體時間變短。
 * 
 * 問：為什么Combiner不作為MR運行的標配，而是可選步驟呢？
 * 答：因為不是所有的算法都適合使用Combiner處理，例如求平均數。
 *
 * 問：Combiner本身已經執行了reduce操作，為什么在Reducer階段還要執行reduce操作呢？
 * 答：combiner操作發生在map端的，處理一個任務所接收的文件中的數據，不能跨map任務執行；只有reduce可以接收多個map任務處理的數據。  25  *
 */
public class WordCountApp {
    static final String INPUT_PATH = "hdfs://chaoren:9000/hello";
    static final String OUT_PATH = "hdfs://chaoren:9000/out";
    
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        final FileSystem fileSystem = FileSystem.get(new URI(INPUT_PATH), conf);
        final Path outPath = new Path(OUT_PATH);
        if(fileSystem.exists(outPath)){
            fileSystem.delete(outPath, true);
        }
        
        final Job job = new Job(conf , WordCountApp.class.getSimpleName());
        //1.1指定讀取的文件位於哪里
        FileInputFormat.setInputPaths(job, INPUT_PATH);
        //指定如何對輸入文件進行格式化，把輸入文件每一行解析成鍵值對
        //job.setInputFormatClass(TextInputFormat.class);
        
        //1.2 指定自定義的map類
        job.setMapperClass(MyMapper.class);
        //map輸出的<k,v>類型。如果<k3,v3>的類型與<k2,v2>類型一致，則可以省略
        //job.setMapOutputKeyClass(Text.class);
        //job.setMapOutputValueClass(LongWritable.class);
        
        //1.3 分區
        //job.setPartitionerClass(HashPartitioner.class);
        //有一個reduce任務運行
        //job.setNumReduceTasks(1);
        
        //1.4 TODO 排序、分組
        
        //1.5 規約
        job.setCombinerClass(MyCombiner.class);  60         
        //2.2 指定自定義reduce類
        job.setReducerClass(MyReducer.class);
        //指定reduce的輸出類型
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);
        
        //2.3 指定寫出到哪里
        FileOutputFormat.setOutputPath(job, outPath);
        //指定輸出文件的格式化類
        //job.setOutputFormatClass(TextOutputFormat.class);
        
        //把job提交給JobTracker運行
        job.waitForCompletion(true);
    }
    
    /**
     * KEYIN    即k1        表示行的偏移量
     * VALUEIN    即v1        表示行文本內容
     * KEYOUT    即k2        表示行中出現的單詞
     * VALUEOUT    即v2        表示行中出現的單詞的次數，固定值1
     */
    static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable>{
        protected void map(LongWritable k1, Text v1, Context context) throws java.io.IOException ,InterruptedException {
            final String[] splited = v1.toString().split("\t");
            for (String word : splited) {
                context.write(new Text(word), new LongWritable(1));
                System.out.println("Mapper輸出<"+word+","+1+">");
            }
        };
    }
    
    /**
     * KEYIN    即k2        表示行中出現的單詞
     * VALUEIN    即v2        表示行中出現的單詞的次數
     * KEYOUT    即k3        表示文本中出現的不同單詞
     * VALUEOUT    即v3        表示文本中出現的不同單詞的總次數
     *
     */
    static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable>{
        protected void reduce(Text k2, java.lang.Iterable<LongWritable> v2s, Context ctx) throws java.io.IOException ,InterruptedException {
            //顯示次數表示redcue函數被調用了多少次，表示k2有多少個分組
            System.out.println("MyReducer輸入分組<"+k2.toString()+",...>");
            long times = 0L;
            for (LongWritable count : v2s) {
                times += count.get();
                //顯示次數表示輸入的k2,v2的鍵值對數量
                System.out.println("MyReducer輸入鍵值對<"+k2.toString()+","+count.get()+">");
            }
            ctx.write(k2, new LongWritable(times));
        };
    }
    
    
    static class MyCombiner extends Reducer<Text, LongWritable, Text, LongWritable>{
        protected void reduce(Text k2, java.lang.Iterable<LongWritable> v2s, Context ctx) throws java.io.IOException ,InterruptedException {
            //顯示次數表示redcue函數被調用了多少次，表示k2有多少個分組
            System.out.println("Combiner輸入分組<"+k2.toString()+",...>");
            long times = 0L;
            for (LongWritable count : v2s) {
                times += count.get();
                //顯示次數表示輸入的k2,v2的鍵值對數量
                System.out.println("Combiner輸入鍵值對<"+k2.toString()+","+count.get()+">");
            }
            
            ctx.write(k2, new LongWritable(times));
            //顯示次數表示輸出的k2,v2的鍵值對數量
            System.out.println("Combiner輸出鍵值對<"+k2.toString()+","+times+">");
        };
    }
}