MapReduce執行過程

1、設定input，告訴程式輸入的資料在那兒。

通過InputFormat介面子類（FileInputFormat, TextInputFormat），
（1）讀取資料
（2）將資料轉換成key-value形式交給Mapper的map()方法進行處理
預設key=行偏移量(LongWritable)，value=行資料(Text)

//設定input
Path inpath = new Path(arg0[0]);
FileInputFormat.addInputPath(job, inpath);

2、執行Mapper的map()方法

多個Mapper並行執行Mapper的map(KEYIN,VALUEIN,KEYOUT,VALUEOUT)方法。

3、shuffle

1、資料溢寫，分割槽，排序
多個map並行工作相互不影響，每個map工作結束後會將輸出的key-value對先儲存在記憶體(memory)快取區中，當記憶體緩衝區中資料的數量達到一定量後會將記憶體緩衝區中的資料溢寫(spill)到本地磁碟中,此時可能會將記憶體緩衝中的資料寫到多個磁碟檔案中，再將資料寫到磁碟檔案的過程中，會對資料進行以下操作：
（1）分割槽：因為shuffle過程結束後，在reduce過程中，多個reduce會拉取map輸出的資料並執行reduce操作，此時一個reduce會對應一個分割槽，即，不同的reduce處理不同分割槽的資料。所以此處的分割槽可以理解為：通過分割槽後，輸出有多個檔案，但是不同檔案中儲存的資料都是夠被同一個reduce處理的資料，或者同一檔案中的資料按照不同分割槽進行分開存放。
（2）將各個分割槽（檔案）中的資料按照key進行排序。

2、檔案合併
當map輸出的資料全部寫到磁碟後，會合並小檔案（merge），將上一過程中輸出的多個小檔案進行合併，在合併的過程中也會進行分割槽和排序，同之前一樣，分割槽即是把能被同一個reduce操作的資料放在一起，排序即是在各個分割槽中將資料按照key進行排序。此時通過合併，分割槽，排序後生成的大檔案存放在map程式本機的本地磁碟，並等待reduce來拷貝資料。

3、Combiner和壓縮
當執行到此處，可以進行以下兩個可設定的操作：
（1）Combiner（map端的reduce），檔案合併的時候，可以執行reduce端操作（將相同key的value合併在一起），這種操作不是每個reduce程式都可以進行，只有不影響最終結果的情況下，才能夠進行Combiner操作。
（2）壓縮：在此處可以將合併後的大檔案進行壓縮，壓縮後可以減少磁碟的佔用量，當reduce copy時也可以減少網路IO。

4、reduce資料copy
接下來reduce task會到map task執行的主機上拷貝自己要處理的資料。具體的過程如下
（1）先將資料放記憶體緩衝區，緩衝區寫滿後溢寫到磁碟。
（2）溢寫磁碟的過程中，會將拷貝的資料進行合併，排序處理。
（3）group：在資料copy結束併合並結束後，將拷貝的資料進行按照key進行分組，即，將相同key的value放在一起，這裡通過比較器進行比較key是否相同。
（4）將分組後的資料傳遞給reduce()方法進行處理

4、執行Reducer的reduce()方法

執行reduce(KEYIN, VALUEIN, KEYOUT,VALUEOUT)方法

5、設定output，執行後的結果存放在那兒

通過介面OutputFormat的子類（FileOutputFormat，TextOutputFormat等）
預設每個key-value輸出一行，key和value中間的分隔符為製表符，key，value輸出是呼叫key和value的toString()方法。

//設定output
Path outpath = new Path(arg0[1]);
FileOutputFormat.setOutputPath(job, outpath);

Shuffle過程中的設定

    public  int run(String[] arg0) throws Exception{
//        Configuration configuration = new Configuration();
        //讀取配置檔案資訊
        Configuration conf = getConf();
//        conf.set("mapreduce.map.output.compress", "true");//設定開啟壓縮，預設為false
//        conf.set("mapreduce.map.output.compress.codec", "org.apache.hadoop.io.compress.Lz4Codec");//設定壓縮演算法,CompressionCodec為壓縮演算法的父類
//        CompressionCodec

        //建立Job
        Job job = Job.getInstance(conf, this.getClass().getSimpleName());
        //設定執行的jar
        job.setJarByClass(this.getClass());

        //設定input
        Path inpath = new Path(arg0[0]);
        FileInputFormat.addInputPath(job, inpath);

        //設定map
        //TODO    需要修改ModelMapper
        job.setMapperClass(ModelMapper.class);
        //TODO    需要修改map的輸出型別
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
//*******************shuffle************************

//        job.setPartitionerClass(cls);     //設定分割槽
//        job.setSortComparatorClass(cls);    //設定排序
//        job.setCombinerClass(cls);          //設定combiner
//        job.setGroupingComparatorClass(cls);    //設定group

//*******************shuffle************************
        //設定reduce
        //TODO    需要修改ModelReduce
        job.setReducerClass(ModelReduce.class);
        //TODO    需要修改reduce的輸出型別
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        //設定output
        Path outpath = new Path(arg0[1]);
        FileOutputFormat.setOutputPath(job, outpath);

        //提交job,設定為true會在執行的時候列印日誌資訊
        boolean isSuccess = job.waitForCompletion(true);
        return isSuccess ? 0 : 1;
    }