1 MapReduce概念 和 MapReduce編程模型
什么是MapReduce

• 源于Google的MapReduce論文(2004年12月)
• Hadoop的MapReduce是Google論文的開源實現(xiàn)
• MapReduce優(yōu)點: 海量數(shù)據(jù)離線處理&易開發(fā)
• MapReduce缺點: 實時流式計算

MapReduce分而治之的思想

• 數(shù)錢實例：一堆鈔票，各種面值分別是多少
• 單點策略
• 一個人數(shù)所有的鈔票，數(shù)出各種面值有多少張
• 分治策略
• 每個人分得一堆鈔票，數(shù)出各種面值有多少張
• 匯總，每個人負責(zé)統(tǒng)計一種面值
• 解決數(shù)據(jù)可以切割進行計算的應(yīng)用

MapReduce編程分Map和Reduce階段

• 將作業(yè)拆分成Map階段和Reduce階段
• Map階段 Map Tasks 分：把復(fù)雜的問題分解為若干"簡單的任務(wù)"
• Reduce階段: Reduce Tasks 合：reduce

MapReduce編程執(zhí)行步驟

• 準備MapReduce的輸入數(shù)據(jù)
• 準備Mapper數(shù)據(jù)
• Shuffle
• Reduce處理
• 結(jié)果輸出

編程模型

• 借鑒函數(shù)式編程方式
• 用戶只需要實現(xiàn)兩個函數(shù)接口：

Map(in_keyin_value)

--->(out_keyintermediate_value) list

Reduce(out_keyintermediate_value) list

--->out_value list

• Word Count 詞頻統(tǒng)計案例

2 應(yīng)用MRJob編寫MapReduce代碼
mrjob 簡介

• 使用python開發(fā)在Hadoop上運行的程序 mrjob是最簡單的方式
• mrjob程序可以在本地測試運行也可以部署到Hadoop集群上運行
• 如果不想成為hadoop專家 但是需要利用Hadoop寫MapReduce代碼mrJob是很好的選擇

mrjob 安裝

• 使用pip安裝

pip install mrjob
mrjob實現(xiàn)WordCount

from mrjob.job import MRJob

class MRWordCount(MRJob):

    #每一行從line中輸入
    def mapper(self _ line):
        for word in line.split():
            yield word1

    # word相同的 會走到同一個reduce
    def reducer(self word counts):
        yield word sum(counts)

if __name__ == __main__:
    MRWordCount.run()

運行WordCount代碼

打開命令行 找到一篇文本文檔 敲如下命令:

python mr_word_count.py my_file.txt

運行MRJOB的不同方式

1、內(nèi)嵌(-r inline)方式

特點是調(diào)試方便，啟動單一進程模擬任務(wù)執(zhí)行狀態(tài)和結(jié)果，默認(-r inline)可以省略，輸出文件使用 > output-file 或-o output-file，比如下面兩種運行方式是等價的

python word_count.py -r inline input.txt > output.txt python word_count.py input.txt > output.txt

2、Hadoop(-r hadoop)方式

用于hadoop環(huán)境

python word_count.py -r hadoop hdfs:///test.txt -o  hdfs:///output

支持Hadoop運行調(diào)度控制參數(shù)，如：

1)指定Hadoop任務(wù)調(diào)度優(yōu)先級(VERY_HIGH|HIGH)如：--jobconf mapreduce.job.priority=VERY_HIGH。

2)Map及Reduce任務(wù)個數(shù)限制，如：--jobconf mapreduce.map.tasks=2 --jobconf mapreduce.reduce.tasks=5

3 mrjob 實現(xiàn) topN統(tǒng)計（實驗）
統(tǒng)計數(shù)據(jù)中出現(xiàn)次數(shù)最多的前n個數(shù)據(jù)

import sys
from mrjob.job import MRJobMRStep
import heapq

class TopNWords(MRJob):
    def mapper(self _ line):
        if line.strip() != "":
            for word in line.strip().split():
                yield word1

    #介于mapper和reducer之間，用于臨時的將mapper輸出的數(shù)據(jù)進行統(tǒng)計
    def combiner(self word counts):
        yield wordsum(counts)

    def reducer_sum(self word counts):
        yield None(sum(counts)word)

    #利用heapq將數(shù)據(jù)進行排序，將最大的2個取出
    def top_n_reducer(self_word_cnts):
        for cntword in heapq.nlargest(2word_cnts):
            yield wordcnt

    #實現(xiàn)steps方法用于指定自定義的mapper，comnbiner和reducer方法
    def steps(self):
        #傳入兩個step 定義了執(zhí)行的順序
        return [
            MRStep(mapper=self.mapper
                   combiner=self.combiner
                   reducer=self.reducer_sum)
            MRStep(reducer=self.top_n_reducer)
        ]

def main():
    TopNWords.run()

if __name__==__main__:
    main()

4 MapReduce原理詳解
單機程序計算流程

輸入數(shù)據(jù)--->讀取數(shù)據(jù)--->處理數(shù)據(jù)--->寫入數(shù)據(jù)--->輸出數(shù)據(jù)

Hadoop計算流程

input data：輸入數(shù)據(jù)

InputFormat：對數(shù)據(jù)進行切分，格式化處理

map：將前面切分的數(shù)據(jù)做map處理(將數(shù)據(jù)進行分類，輸出(kv)鍵值對數(shù)據(jù))

shuffle&sort:將相同的數(shù)據(jù)放在一起，并對數(shù)據(jù)進行排序處理

reduce：將map輸出的數(shù)據(jù)進行hash計算，對每個map數(shù)據(jù)進行統(tǒng)計計算

OutputFormat：格式化輸出數(shù)據(jù)

map：將數(shù)據(jù)進行處理

buffer in memory：達到80%數(shù)據(jù)時，將數(shù)據(jù)鎖在內(nèi)存上，將這部分輸出到磁盤上

partitions：在磁盤上有很多"小的數(shù)據(jù)"，將這些數(shù)據(jù)進行歸并排序。

merge on disk：將所有的"小的數(shù)據(jù)"進行合并。

reduce：不同的reduce任務(wù)，會從map中對應(yīng)的任務(wù)中copy數(shù)據(jù)，在reduce中同樣要進行merge操作

5 MapReduce架構(gòu)
MapReduce架構(gòu) 1.X
JobTracker：負責(zé)接收客戶作業(yè)提交，負責(zé)任務(wù)到作業(yè)節(jié)點上運行，檢查作業(yè)的狀態(tài)
TaskTracker：由JobTracker指派任務(wù)，定期向JobTracker匯報狀態(tài)，在每一個工作節(jié)點上永遠只會有一個TaskTracker

MapReduce2.X架構(gòu)

ResourceManager：負責(zé)資源的管理，負責(zé)提交任務(wù)到NodeManager所在的節(jié)點運行，檢查節(jié)點的狀態(tài)
NodeManager：由ResourceManager指派任務(wù)，定期向ResourceManager匯報狀態(tài)
{{image.png(uploading...)}}