1.1MapReduce的作用

假設有一個計算檔案中單詞個數的需求，檔案比較多也比較大，在單擊執行的時候機器的記憶體受限，磁碟受限，運算能力受限，而一旦將單機版程式擴充套件到叢集來分散式執行，將極大增加程式的複雜度和開發難度，因此這個工作可能完成不了。針對以上這個案例，MapReduce在這裡能起到什麼作用呢，引入MapReduce框架後，開發人員可以將絕大部分工作集中在業務邏輯的開發上，而將分散式計算中的複雜性交由框架來處理。

可見在程式由單機版擴成分散式時，會引入大量的複雜工作。為了提高開發效率，可以將分散式程式中的公共功能封裝成框架，讓開發人員可以將精力集中於業務邏輯。而

1.2MapReduce架構圖

MapReduce 也採用了 Master/Slave（M/S）架構。它主要由以下幾個元件組成 ：Client、JobTracker、 TaskTracker 和 Task。下面分別對這幾個元件進行介紹。

（1）Client

 使用者編寫的MapReduce程式通過Client提交到JobTracker端；同時使用者可通過Client提供的一些介面檢視作業執行狀態。在Hadoop內部用“作業” （Job）表示MapReduce程式。一個 MapReduce程式可對應若干個作業，而每個作業會被分解成若干個Map/Reduce

（2）JobTracker

JobTracker 主要負責資源監控和作業排程。JobTracker 監控所有 TaskTracker 與作業Job的健康狀況，一旦發現失敗情況後，其會將相應的任務轉移到其他節點；同時，JobTracker 會跟蹤任務的執行進度、資源使用量等資訊，並將這些資訊告訴任務排程器，而排程器會在資源出現空閒時，選擇合適的任務使用這些資源。在Hadoop 中，任務排程器是一個可插拔的模組，使用者可以根據自己的需要設計相應的排程器。

（3）TaskTracker

TaskTracker會週期性地通過Heartbeat將本節點上資源的使用情況和任務的執行進度彙報給

（4）Task

Task 分為 Map Task 和 Reduce Task 兩種，均由TaskTracker啟動。從上一小節中我們知道，HDFS以固定大小的block 為基本單位儲存資料，而對於MapReduce 而言，其處理單位是split。 split 與 block 的對應關係如下圖所示。split 是一個邏輯概念，它只包含一些元資料資訊，比如 資料起始位置、資料長度、資料所在節點等。它的劃分方法完全由使用者自己決定。但需要注意的是，split的多少決定了Map Task的數目，因為每個split會交由一個Map Task處理。

Map Task 執行過程如下圖所示。由該圖可知，Map Task 先將對應的split 迭代解析成一 個個 key/value 對，依次呼叫使用者自定義的map() 函式進行處理，最終將臨時結果存放到本地磁碟上，其中臨時資料被分成若干個partition(分片)，每個partition 將被一個Reduce Task處理。

Reduce Task 執行過程如下圖所示。該過程分為三個階段：

①從遠端節點上讀取Map Task 中間結果（稱為“Shuffle階段”）；

②按照key對key/value 對進行排序（稱為“Sort階段”）；

③依次讀取 <key, value list>，呼叫使用者自定義的 reduce() 函式處理，並將最終結果存到HDFS上（稱為“Reduce 階段”）。

MapReduce是一種並行程式設計模式，利用這種模式軟體開發者可以輕鬆地編寫出分散式並行程式。在Hadoop的體系結構中，MapReduce是一個簡單易用的軟體框架，基於它可以將任務分發到由上千臺商用機器組成的叢集上，並以一種可靠容錯的方式並行處理大量的資料集，實現Hadoop的並行任務處理功能。MapReduce框架是由一個單獨執行在主節點的JobTrack和執行在每個叢集從節點的TaskTrack共同組成的。

主節點負責排程構成一個作業的所有任務，這些任務分佈在不同的節點上。主節點監控它們的執行情況，並且重新執行之前失敗的任務；

從節點僅負責由主節點指派的任務。

當一個Job任務被提交時，JobTrack接收到提交作業和其配置資訊之後，就會配置資訊等發給從節點，同時排程任務並監控TaskTrack的執行。

1.3MapReduce程式執行演示

Hadoop的釋出包中內建了一個hadoop-mapreduce-example-2.6.5.jar，這個jar包中有各種MR示例程式，可以通過以下步驟執行：

啟動hdfs，yarn，然後在叢集中的任意一臺伺服器上啟動執行程式（比如執行wordcount）：

hadoop jar hadoop-mapreduce-example-2.6.5.jar wordcount  /wordcount/data /wordcount/out

2.1程式設計規範

1) 使用者編寫的程式分成三個部分：Mapper，Reducer，Driver(提交執行mr程式的客戶端)

2) Mapper的輸入資料是KV對的形式（KV的型別可自定義）

3) Mapper的輸出資料是KV對的形式（KV的型別可自定義）

4) Mapper中的業務邏輯寫在map()方法中

5) map()方法（maptask程序）對每一個<K,V>呼叫一次

6) Reducer的輸入資料型別對應Mapper的輸出資料型別，也是KV

7) Reducer的業務邏輯寫在reduce()方法中

8) Reducetask程序對每一組相同k的<k,v>組呼叫一次reduce()方法

9) 使用者自定義的Mapper和Reducer都要繼承各自的父類

10) 整個程式需要一個Drvier來進行提交，提交的是一個描述了各種必要資訊的job物件

2.2wordcount 示例編寫

需求：在一堆給定的文字檔案中統計輸出每一個單詞出現的總次數

(1)定義一個mapper類

//首先要定義四個泛型的型別
//keyin:  LongWritable    valuein: Text
//keyout: Text            valueout:IntWritable

public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
    //map方法的生命週期：  框架每傳一行資料就被呼叫一次
    //key :  這一行的起始點在檔案中的偏移量
    //value: 這一行的內容
    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
        //拿到一行資料轉換為string
        String line = value.toString();
        //將這一行切分出各個單詞
        String[] words = line.split(" ");
        //遍歷陣列，輸出<單詞，1>
        for(String word:words){
            context.write(new Text(word), new IntWritable(1));
        }
    }
}

(2)定義一個reducer類

//生命週期：框架每傳遞進來一個kv 組，reduce方法被呼叫一次
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        //定義一個計數器
        int count = 0;
        //遍歷這一組kv的所有v，累加到count中
        for(IntWritable value:values){
            count += value.get();
        }
        context.write(key, new IntWritable(count));
    }
}

(3)定義一個主類，用來描述job並提交job

public class WordCountRunner {
    //把業務邏輯相關的資訊（哪個是mapper，哪個是reducer，要處理的資料在哪裡，輸出的結果放哪裡……）描述成一個job物件
    //把這個描述好的job提交給叢集去執行
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        Job wcjob = Job.getInstance(conf);
        //指定我這個job所在的jar包
//        wcjob.setJar("/home/hadoop/wordcount.jar");
        wcjob.setJarByClass(WordCountRunner.class);
        
        wcjob.setMapperClass(WordCountMapper.class);
        wcjob.setReducerClass(WordCountReducer.class);
        //設定我們的業務邏輯Mapper類的輸出key和value的資料型別
        wcjob.setMapOutputKeyClass(Text.class);
        wcjob.setMapOutputValueClass(IntWritable.class);
        //設定我們的業務邏輯Reducer類的輸出key和value的資料型別
        wcjob.setOutputKeyClass(Text.class);
        wcjob.setOutputValueClass(IntWritable.class);
        
        //指定要處理的資料所在的位置
        FileInputFormat.setInputPaths(wcjob, "hdfs://hdp-server01:9000/wordcount/data/big.txt");
        //指定處理完成之後的結果所儲存的位置
        FileOutputFormat.setOutputPath(wcjob, new Path("hdfs://hdp-server01:9000/wordcount/output/"));
        
        //向yarn叢集提交這個job
        boolean res = wcjob.waitForCompletion(true);
        System.exit(res?0:1);
    }

2.3叢集執行模式

1) 將mapreduce程式提交給yarn叢集resourcemanager，分發到很多的節點上併發執行

2) 處理的資料和輸出結果應該位於hdfs檔案系統

3) 提交叢集的實現步驟：

將程式打成JAR包，然後在叢集的任意一個節點上用hadoop命令啟動hadoop jar wordcount.jar cn.bigdata.mrsimple.WordCountDriver inputpath outputpath

技有限公司和部落格園共有，歡迎轉載，但未經作者同意必須保留此段宣告，且在文章頁面明顯位置給出原文連線，否則保留追究法律責任的權利。
技術諮詢：