1. MapReduce 的介紹：

   MapReduce 是一個(gè)分布式運(yùn)算程序的編程框架，核心功能是將用戶編寫的業(yè)務(wù)邏輯代碼和自帶默認(rèn)組件整合成一個(gè)完整的分布式運(yùn)算程序，并發(fā)運(yùn)行在一個(gè) Hadoop 集群上。
  MapReduce大體上分三個(gè)部分：
  - MRAppMaster：MapReduce Application Master，分配任務(wù)，協(xié)調(diào)任務(wù)的運(yùn)行
  - MapTask：階段并發(fā)任，負(fù)責(zé) mapper 階段的任務(wù)處理 YARNChild
  - ReduceTask：階段匯總?cè)蝿?wù)，負(fù)責(zé) reducer 階段的任務(wù)處理 YARNChild

讓客戶滿意是我們工作的目標(biāo)，不斷超越客戶的期望值來自于我們對這個(gè)行業(yè)的熱愛。我們立志把好的技術(shù)通過有效、簡單的方式提供給客戶，將通過不懈努力成為客戶在信息化領(lǐng)域值得信任、有價(jià)值的長期合作伙伴，公司提供的服務(wù)項(xiàng)目有：申請域名、雅安服務(wù)器托管、營銷軟件、網(wǎng)站建設(shè)、福海網(wǎng)站維護(hù)、網(wǎng)站推廣。

2.MapReduce編寫代碼的流程：

• 用戶編寫的程序分成三個(gè)部分：Mapper，Reducer，Driver(提交運(yùn)行 MR 程序的客戶端)
• Mapper 的輸入數(shù)據(jù)是 KV 對的形式（KV 的類型可自定義）
• Mapper 的輸出數(shù)據(jù)是 KV 對的形式（KV 的類型可自定義）
• Mapper 中的業(yè)務(wù)邏輯寫在 map()方法中
• map()方法（maptask 進(jìn)程）對每一個(gè)調(diào)用一次
• Reducer 的輸入數(shù)據(jù)類型對應(yīng) Mapper 的輸出數(shù)據(jù)類型，也是 KV 對的形式
• Reducer 的業(yè)務(wù)邏輯寫在 reduce()方法中
• Reducetask 進(jìn)程對每一組相同 k 的組調(diào)用一次 reduce()方法
• 用戶自定義的 Mapper 和 Reducer 都要繼承各自的父類
• 整個(gè)程序需要一個(gè) Drvier 來進(jìn)行提交，提交的是一個(gè)描述了各種必要信息的 job 對象

3.WordCount 案例：

public class MyWordCount {
    public static void main(String[] args) {
        // 指定 hdfs 相關(guān)的參數(shù)
        Configuration conf=new Configuration(true);
        conf.set("fs.defaultFS","hdfs://hadoop01:9000");
        System.setProperty("HADOOP_USER_NAME", "hadoop");
        try {
            // 新建一個(gè) job 任務(wù)
            Job job=Job.getInstance(conf);
            // 設(shè)置 jar 包所在路徑
           job.setJarByClass(MyWordCount.class);
            // 指定 mapper 類和 reducer 類
            job.setMapperClass(Mapper.class);
            job.setReducerClass(MyReduce.class);

            // 指定 maptask 的輸出類型,注意，如果maptask的輸出類型與reducetask輸出類型一樣，mapTask可以不用設(shè)置
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(IntWritable.class);
            // 指定 reducetask 的輸出類型
            job.setOutputKeyClass(Text.class);
            job.setOutputValueClass(Text.class);

            // 指定該 mapreduce 程序數(shù)據(jù)的輸入和輸出路徑
            Path input=new Path("/data/input");
            Path output =new Path("/data/output");
            //一定要保證output不存在
            if(output.getFileSystem(conf).exists(output)){
                output.getFileSystem(conf).delete(output,true);  //遞歸刪除
            }
            FileInputFormat.addInputPath(job,input);
            FileOutputFormat.setOutputPath(job,output);

            // 最后提交任務(wù)
             boolean success = job.waitForCompletion(true);
             System.exit(success?0:-1);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
    private class MyMapper extends Mapper{
        Text mk =new Text();
        IntWritable mv=new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context)
                throws IOException, InterruptedException {
            // 計(jì)算任務(wù)代碼：切割單詞，輸出每個(gè)單詞計(jì) 1 的 key-value 對
             String[] words = value.toString().split("\\s+");
             for(String word:words){
                 mk.set(word);
                 context.write(mk,mv);
             }
        }
    }
    private class MyReduce extends Reducer {
        IntWritable mv=new IntWritable();
        @Override
        protected void reduce(Text key, Iterable values, Context context)
                throws IOException, InterruptedException {
            int sum=0;
            // 匯總計(jì)算代碼：對每個(gè) key 相同的一組 key-value 做匯總統(tǒng)計(jì)
            for(IntWritable value:values){
                sum+=value.get();
            }
            mv.set(sum);
            context.write(key,mv);
        }
    }
}

4. MapReduce 程序的核心運(yùn)行機(jī)制：

1)MapReduce 程序的運(yùn)行流程:

• 一個(gè) mr 程序啟動的時(shí)候，最先啟動的是 MRAppMaster，MRAppMaster 啟動后根據(jù)本次job 的描述信息，計(jì)算出需要的 maptask 實(shí)例數(shù)量，然后向集群申請機(jī)器啟動相應(yīng)數(shù)量的maptask 進(jìn)程
• maptask 進(jìn)程啟動之后，根據(jù)給定的數(shù)據(jù)切片(哪個(gè)文件的哪個(gè)偏移量范圍)范圍進(jìn)行數(shù)據(jù)處理，主體流程為：
  • 利用客戶指定的 InputFormat 來獲取 RecordReader 讀取數(shù)據(jù)，形成輸入 KV 對
  • 將輸入 KV 對傳遞給客戶定義的 map()方法，做邏輯運(yùn)算，并將 map()方法輸出的 KV 對收集到緩存
  • 將緩存中的 KV 對按照 K 分區(qū)排序后不斷溢寫到磁盤文件
• MRAppMaster 監(jiān)控到所有 maptask 進(jìn)程任務(wù)完成之后（真實(shí)情況是，某些 maptask 進(jìn)程處理完成后，就會開始啟動 reducetask 去已完成的 maptask 處 fetch 數(shù)據(jù)），會根據(jù)客戶指定的參數(shù)啟動相應(yīng)數(shù)量的 reducetask 進(jìn)程，并告知 reducetask 進(jìn)程要處理的數(shù)據(jù)范圍（數(shù)據(jù)分區(qū)）
• Reducetask 進(jìn)程啟動之后，根據(jù) MRAppMaster 告知的待處理數(shù)據(jù)所在位置，從若干臺maptask 運(yùn)行所在機(jī)器上獲取到若干個(gè) maptask 輸出結(jié)果文件，并在本地進(jìn)行重新歸并排序，然后按照相同 key 的 KV 為一個(gè)組，調(diào)用客戶定義的 reduce()方法進(jìn)行邏輯運(yùn)算，并收集運(yùn)算輸出的結(jié)果 KV，然后調(diào)用客戶指定的 OutputFormat 將結(jié)果數(shù)據(jù)輸出到外部存儲

  2)MapTask 并行度決定機(jī)制：

     一個(gè) job 的 map 階段并行度由客戶端在提交 job 時(shí)決定，客戶端對 map 階段并行度的規(guī)劃的基本邏輯為：將待處理數(shù)據(jù)執(zhí)行邏輯切片（即按照一個(gè)特定切片大小，將待處理數(shù)據(jù)劃分成邏輯上的多個(gè) split），然后每一個(gè) split 分配一個(gè) mapTask 并行實(shí)例處理。這段邏輯及形成的切片規(guī)劃描述文件，是由FileInputFormat實(shí)現(xiàn)類的getSplits()方法完成的，小編后續(xù)會對MPjob提交過程的源碼進(jìn)行詳細(xì)的分析。
     決定map task的個(gè)數(shù)主要由這幾個(gè)方面：
      -文件的大小
      - 文件的個(gè)數(shù)
      - block的大小
      - 邏輯切片的大小
     總的來說就是，當(dāng)對文件進(jìn)行邏輯劃分的時(shí)候，默認(rèn)的劃分規(guī)則就是一個(gè)split和一個(gè)block的大小一樣，如果文件沒有到一個(gè)block大小，也會被切分出來一個(gè)split，這里有調(diào)優(yōu)點(diǎn)，就是如果處理的文件都是小文件的話，那么機(jī)會并行很多的maptask，導(dǎo)致大量的時(shí)間都浪費(fèi)在了啟動jvm上，此時(shí)可以通過合并小文件或者重用jvm的方式提高效率。
     邏輯切片機(jī)制：
  long splitSize = computeSplitSize(blockSize, minSize, maxSize)
  blocksize：默認(rèn)是 128M，可通過 dfs.blocksize 修改
  minSize：默認(rèn)是 1，可通過 mapreduce.input.fileinputformat.split.minsize 修改
  maxsize：默認(rèn)是 Long.MaxValue，可通過mapreduce.input.fileinputformat.split.maxsize 修改
  因此，如果是想調(diào)小split的大小，那么就將 maxsize調(diào)整到比block小。
  如果是想調(diào)大split的大小，那么就將minSize調(diào)整到比block大。

  3)ReduceTask 并行度決定機(jī)制：

     reducetask 的并行度同樣影響整個(gè) job 的執(zhí)行并發(fā)度和執(zhí)行效率，但與 maptask 的并發(fā)數(shù)由切片數(shù)決定不同，Reducetask 數(shù)量的決定是可以直接手動設(shè)置：job.setNumReduceTasks(4);，默認(rèn)是1個(gè)，如果設(shè)置為0個(gè)表示沒有reduce階段，當(dāng)然也可以設(shè)置多個(gè)，根據(jù)需求，如果有些需要全局計(jì)數(shù)的操作，那么只能設(shè)置1個(gè)reduce，有些可以設(shè)置多個(gè)reducetask的，千萬不要設(shè)置太多，最好設(shè)置的和分區(qū)的個(gè)數(shù)能一一對應(yīng)，不然的會就會有一些reduceTask空跑，導(dǎo)致了不能處理業(yè)務(wù)而且還占用系統(tǒng)資源。

另外有需要云服務(wù)器可以了解下創(chuàng)新互聯(lián)scvps.cn，海內(nèi)外云服務(wù)器15元起步，三天無理由+7*72小時(shí)售后在線，公司持有idc許可證，提供“云服務(wù)器、裸金屬服務(wù)器、高防服務(wù)器、香港服務(wù)器、美國服務(wù)器、虛擬主機(jī)、免備案服務(wù)器”等云主機(jī)租用服務(wù)以及企業(yè)上云的綜合解決方案，具有“安全穩(wěn)定、簡單易用、服務(wù)可用性高、性價(jià)比高”等特點(diǎn)與優(yōu)勢，專為企業(yè)上云打造定制，能夠滿足用戶豐富、多元化的應(yīng)用場景需求。

當(dāng)前名稱：MapReduce的入門-創(chuàng)新互聯(lián)
轉(zhuǎn)載來源：http://weahome.cn/article/cssjoj.html