最近在研究Hadoop,發現網上的一些關於Hadoop的資料都是以前的1.X版本的,包括MapReduce的工作原理,都是以前的一些過時了的東西,所以自己重新整理了一些新2.X版本的MapReduce的工作原理

下面我畫了一張圖,便於理解MapReduce得整個工作原理



下面對上面出現的一些名詞進行介紹


ResourceManager：是YARN資源控制框架的中心模組，負責叢集中所有的資源的統一管理和分配。它接收來自NM(NodeManager)的彙報，建立AM，並將資源派送給AM(ApplicationMaster)。

NodeManager:簡稱NM，NodeManager是ResourceManager在每臺機器的上代理，負責容器的管理，並監控他們的資源使用情況（cpu，記憶體，磁碟及網路等），以及向 ResourceManager提供這些資源使用報告。

ApplicationMaster:以下簡稱AM。YARN中每個應用都會啟動一個AM，負責向RM申請資源，請求NM啟動container，並告訴container做什麼事情。

Container：資源容器。YARN中所有的應用都是在container之上執行的。AM也是在container上執行的，不過AM的container是RM申請的。

1.  Container是YARN中資源的抽象，它封裝了某個節點上一定量的資源（CPU和記憶體兩類資源）。

2.  Container由ApplicationMaster向ResourceManager申請的，由ResouceManager中的資源排程器非同步分配給ApplicationMaster；
3. Container的執行是由ApplicationMaster向資源所在的NodeManager發起的，Container執行時需提供內部執行的任務命令（可以是任何命令，比如java、Python、C++程序啟動命令均可）以及該命令執行所需的環境變數和外部資源（比如詞典檔案、可執行檔案、jar包等）。
另外，一個應用程式所需的Container分為兩大類，如下：
       （1） 執行ApplicationMaster的Container：這是由ResourceManager（向內部的資源排程器）申請和啟動的，使用者提交應用程式時，可指定唯一的ApplicationMaster所需的資源；
       （2） 執行各類任務的Container：這是由ApplicationMaster向ResourceManager申請的，並由ApplicationMaster與NodeManager通訊以啟動之。
以上兩類Container可能在任意節點上，它們的位置通常而言是隨機的，即ApplicationMaster可能與它管理的任務執行在一個節點上。


整個MapReduce的過程大致分為 Map-->Shuffle（排序）-->Combine（組合）-->Reduce

下面通過一個單詞計數案例來理解各個過程
1）將檔案拆分成splits(片)，並將每個split按行分割形成<key,value>對，如圖所示。這一步由MapReduce框架自動完成，其中偏移量即key值



                    
                    分割過程



將分割好的<key,value>對交給使用者定義的map方法進行處理，生成新的<key,value>對，如下圖所示。


                    執行map方法
                    



得到map方法輸出的<key,value>對後，Mapper會將它們按照key值進行Shuffle（排序），並執行Combine過程，將key至相同value值累加，得到Mapper的最終輸出結果。如下圖所示。

                     Map端排序及Combine過程



Reducer先對從Mapper接收的資料進行排序，再交由使用者自定義的reduce方法進行處理，得到新的<key,value>對，並作為WordCount的輸出結果，如下圖所示。


                    Reduce端排序及輸出結果





下面看怎麼用Java來實現WordCount單詞計數的功能

首先看Map過程
Map過程需要繼承org.apache.hadoop.mapreduce.Mapper包中 Mapper 類，並重寫其map方法。

/**
     *
     * @author 湯高
     *    Mapper<LongWritable, Text, Text, IntWritable>中  LongWritable,IntWritable是Hadoop資料型別表示長整型和整形
     *
     *    LongWritable, Text表示輸入型別 (比如本應用單詞計數輸入是 偏移量(字串中的第一個單詞的其實位置),對應的單詞(值))
     *    Text, IntWritable表示輸出型別  輸出是單詞  和他的個數
     *  注意：map函式中前兩個引數LongWritable key, Text value和輸出型別不一致
     *      所以後面要設定輸出型別 要使他們一致
     */
    //Map過程
    public static class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        /***
         *
         */
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
                throws IOException, InterruptedException {
            //預設的map的value是每一行,我這裡自定義的是以空格分割
            String[] vs = value.toString().split("\\s");
            for (String v : vs) {
                //寫出去
                context.write(new Text(v), ONE);
            }

        }
    }

Reduce過程
Reduce過程需要繼承org.apache.hadoop.mapreduce包中 Reducer 類，並 重寫 其reduce方法。Map過程輸出<key,values>中key為單個單詞，而values是對應單詞的計數值所組成的列表，Map的輸出就是Reduce的輸入，所以reduce方法只要遍歷values並求和，即可得到某個單詞的總次數。

//Reduce過程
    /***
     * @author 湯高
     * Text, IntWritable輸入型別,從map過程獲得 既map的輸出作為Reduce的輸入
     * Text, IntWritable輸出型別
     */
    public static class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values,
                Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            int count=0;
            for(IntWritable v:values){
                count+=v.get();//單詞個數加一
            }
            
            context.write(key, new IntWritable(count));
        }
        
    }

最後執行MapReduce任務

public static void main(String[] args) {
        
        Configuration conf=new Configuration();
        try {
            //args從控制檯獲取路徑 解析得到域名
            String[] paths=new GenericOptionsParser(conf,args).getRemainingArgs();
            if(paths.length<2){
                throw new RuntimeException("必須輸出 輸入 和輸出路徑");
            }
            //得到一個Job 並設定名字
            Job job=Job.getInstance(conf,"wordcount");
            //設定Jar 使本程式在Hadoop中執行
            job.setJarByClass(WordCount.class);
            //設定Map處理類
            job.setMapperClass(WordCountMapper.class);
            //設定map的輸出型別,因為不一致,所以要設定
            job.setMapOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            //設定Reduce處理類
            job.setReducerClass(WordCountReducer.class);
            //設定輸入和輸出目錄
            FileInputFormat.addInputPath(job, new Path(paths[0]));
            FileOutputFormat.setOutputPath(job, new Path(paths[1]));
            //啟動執行
            System.exit(job.waitForCompletion(true) ? 0:1);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

即可求得每個單詞的個數

下面把整個過程的原始碼附上,有需要的朋友可以拿去測試

package hadoopday02;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
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;
import org.apache.hadoop.util.GenericOptionsParser;

public class WordCount {
    //計數變數
    private static final IntWritable ONE = new IntWritable(1);
    /**
     *
     * @author 湯高
     *    Mapper<LongWritable, Text, Text, IntWritable>中  LongWritable,IntWritable是Hadoop資料型別表示長整型和整形
     *
     *    LongWritable, Text表示輸入型別 (比如本應用單詞計數輸入是 偏移量(字串中的第一個單詞的其實位置),對應的單詞(值))
     *    Text, IntWritable表示輸出型別  輸出是單詞  和他的個數
     *  注意：map函式中前兩個引數LongWritable key, Text value和輸出型別不一致
     *      所以後面要設定輸出型別 要使他們一致
     */
    //Map過程
    public static class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        /***
         *
         */
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, IntWritable>.Context context)
                throws IOException, InterruptedException {
            //預設的map的value是每一行,我這裡自定義的是以空格分割
            String[] vs = value.toString().split("\\s");
            for (String v : vs) {
                //寫出去
                context.write(new Text(v), ONE);
            }

        }
    }
    //Reduce過程
    /***
     * @author 湯高
     * Text, IntWritable輸入型別,從map過程獲得 既map的輸出作為Reduce的輸入
     * Text, IntWritable輸出型別
     */
    public static class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable>{
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values,
                Reducer<Text, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
            int count=0;
            for(IntWritable v:values){
                count+=v.get();//單詞個數加一
            }
            
            context.write(key, new IntWritable(count));
        }
        
    }
    
    public static void main(String[] args) {
        
        Configuration conf=new Configuration();
        try {
            //args從控制檯獲取路徑 解析得到域名
            String[] paths=new GenericOptionsParser(conf,args).getRemainingArgs();
            if(paths.length<2){
                throw new RuntimeException("必須輸出 輸入 和輸出路徑");
            }
            //得到一個Job 並設定名字
            Job job=Job.getInstance(conf,"wordcount");
            //設定Jar 使本程式在Hadoop中執行
            job.setJarByClass(WordCount.class);
            //設定Map處理類
            job.setMapperClass(WordCountMapper.class);
            //設定map的輸出型別,因為不一致,所以要設定
            job.setMapOutputKeyClass(Text.class);
            job.setOutputValueClass(IntWritable.class);
            //設定Reduce處理類
            job.setReducerClass(WordCountReducer.class);
            //設定輸入和輸出目錄
            FileInputFormat.addInputPath(job, new Path(paths[0]));
            FileOutputFormat.setOutputPath(job, new Path(paths[1]));
            //啟動執行
            System.exit(job.waitForCompletion(true) ? 0:1);
        } catch (IOException e) {
            e.printStackTrace();
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    

}

好了，整個MapReduce的工作流程就分析到這裡了,上面全是個人學習歸納的,如果有什麼需要改善的地方,歡迎大家指正,一起進步
碼字不易,轉載請指明出處http://blog.csdn.net/tanggao1314/article/details/51275812