MapReduce極簡教程

一個有趣的例子

你想數出一摞牌中有多少張黑桃。直觀方式是一張一張檢查而且數出有多少張是黑桃？

 

MapReduce方法則是：

給在座的全部玩家中分配這摞牌

讓每一個玩家數本身手中的牌有幾張是黑桃，而後把這個數目彙報給你

你把全部玩家告訴你的數字加起來，獲得最後的結論

拆分

MapReduce合併了兩種經典函數：

映射（Mapping）對集合裏的每一個目標應用同一個操做。即，若是你想把表單裏每一個單元格乘以二，那麼把這個函數單獨地應用在每一個單元格上的操做就屬於mapping。

化簡（Reducing ）遍歷集合中的元素來返回一個綜合的結果。即，輸出表單裏一列數字的和這個任務屬於reducing。

從新審視上面的例子

從新審視咱們原來那個分散紙牌的例子，咱們有MapReduce數據分析的基本方法。友情提示：這不是個嚴謹的例子。在這個例子裏，人表明計算機，由於他們同時工做，因此他們是個集羣。在大多數實際應用中，咱們假設數據已經在每臺計算機上了 – 也就是說把牌分發出去並非MapReduce的一步。（事實上，在計算機集羣中如何存儲文件是Hadoop的真正核心。）

經過把牌分給多個玩家而且讓他們各自數數，你就在並行執行運算，由於每一個玩家都在同時計數。這同時把這項工做變成了分佈式的，由於多個不一樣的人在解決同一個問題的過程當中並不須要知道他們的鄰居在幹什麼。

經過告訴每一個人去數數，你對一項檢查每張牌的任務進行了映射。 你不會讓他們把黑桃牌遞給你，而是讓他們把你想要的東西化簡爲一個數字。

另一個有意思的狀況是牌分配得有多均勻。MapReduce假設數據是洗過的（shuffled）- 若是全部黑桃都分到了一我的手上，那他數牌的過程可能比其餘人要慢不少。

若是有足夠的人的話，問一些更有趣的問題就至關簡單了 - 好比「一摞牌的平均值（二十一點算法）是什麼」。你能夠經過合併「全部牌的值的和是什麼」及「咱們有多少張牌」這兩個問題來獲得答案。用這個和除以牌的張數就獲得了平均值。

MapReduce算法的機制要遠比這複雜得多，可是主體思想是一致的 – 經過分散計算來分析大量數據。不管是Facebook、NASA，仍是小創業公司，MapReduce都是目前分析互聯網級別數據的主流方法。

什麼是大數據？

大數據是大量數據的集合，數據量之大以致於用傳統的計算方法沒法處理如此龐大的數據。好比，Facebook和Youtube在平常中搜集和管理的大量數據就屬於大數據的範疇。大數據不只僅是指數據的規模和數量龐大，它一般還包括如下一個或多個方面：處理數據的速度、數據的種類、體積以及複雜度。

爲何是MapReduce?

傳統的企業系統有一箇中央服務器來保存和處理數據。下圖爲傳統的企業系統的原理圖。傳統的模型不適合處理海量的數據，也不適用於標準的數據庫。並且，中央處理系統在同時處理多個文件的時候遇到了瓶頸。

 

Google使用了一個叫MapReduce的算法解決了這個瓶頸。MapReduce把一個任務拆分紅了多個小任務，並把子任務分配到多臺計算機上進行工做。最終，每臺計算機上的計算結果會被蒐集起來併合併成最終的結果。

 

MapReduce是如何工做的？

MapReduce算法包含兩部分重要的任務：Map和Reduce.

Map任務把一個數據集轉化成另外一個數據集，單獨的元素會被拆分紅鍵值對(key-value pairs).

Reduce任務把Map的輸出做爲輸入，把這些鍵值對的數據合併成一個更小的鍵值對數據集.

讓咱們經過下圖瞭解一下MapReduce每一個階段的工做，並理解他們的重要性。

 

Input Phase- 在本階段咱們使用一個Record Reader對輸入文件中的每一條數據轉換爲鍵值對的形式，並把這些處理好的數據發送給Mapper。

Map- Map是是用戶自定義的一個函數，此函數接收一系列的鍵值對數據並對它們進行處理，最後生成0個或多個鍵值對數據。

Intermediate Keys- 由mapper生成的鍵值對數據被稱爲中間狀態的鍵值對。

Shuffle and Sort- Reducer任務一般以Shuffle（攪動）和Sort（排序）開始。程序把分好組的鍵值對數據下載到本機，Reducer會在本機進行運行。這些獨立的鍵值對數據會按照鍵值進行排序並造成一個較大的數據序列，數據序列中鍵值相等的鍵值對數據會被分在相同的一組，這樣易於在Reducer任務中進行迭代操做。

Reducer- Reducer任務把分好組的鍵值對數據做爲輸入，而且對每個鍵值對都執行Reducer函數。在這個階段，程序會以不一樣的方式對數據進行合併、篩選。一旦執行完畢，Reducer會生成0個或多個鍵值對數據，並提供給最後一個處理步驟。

Output Phase- 在輸出階段，經過record writer把從Reducer函數輸出的鍵值對數據按照必定的格式寫入到文件中。

讓咱們經過下圖來進一步瞭解Map和Reduce這兩個任務是如何工做的。

 

MapReduce例子

讓咱們以一個真實的例子來理解MapReduce的威力。Twitter天天都會收到50億條（有那麼多？）推特，約每秒3000條。下圖展現了Twitter是如何利用MapReduce來管理這些數據的。

 

從上述插圖中咱們能夠看到MapReduce執行了如下這些行爲 -

Tokenize- 處理器把推文以鍵值對的形式存放在maps中。

Filter- 把不想要的數據從maps中剔除，把篩選好的數據以鍵值對的形式保存。

Count- 對每一個單詞生成一個計數器。

Aggregate Counter- Prepares an aggregate of similar counter values into small manageable units.

基於MapReduce的處理過程示例--文檔詞頻統計：WordCount

設有4組原始文本數據：

Text 1:the weather is goodText 2:today is good

Text 3:good weather is goodText 4:today has good weather

傳統的串行處理方式(Java)：

String[] text =newString[] { 「hello world」, 「hello every one」, 「say hello to everyone in the world」 ｝;

HashTable ht=newHashTable();for(i = 0; i < 3; ++i) {

StringTokenizer st=newStringTokenizer(text[i]);while(st.hasMoreTokens()) {

String word=st.nextToken();if(!ht.containsKey(word)) {

ht.put(word,newInteger(1));

}else{intwc = ((Integer)ht.get(word)).intValue() +1;//計數加1ht.put(word,newInteger(wc));

}

}

}for(Iterator itr=ht.KeySet().iterator();  itr.hasNext(); ) {

String word=(String)itr.next();

System.out.print(word+ 「: 」+ (Integer)ht.get(word)+「;   」);

}

輸出：good:  5;   has: 1;  is: 3;   the: 1;   today: 2;    weather: 3

基於MapReduce的處理過程示例--文檔詞頻統計：WordCount

MapReduce處理方式

使用4個map節點：

map節點1:

輸入：(text1, 「the weather is good」)

輸出：(the, 1), (weather, 1), (is, 1), (good, 1)

map節點2:

輸入：(text2, 「today is good」)

輸出：(today, 1), (is, 1), (good, 1)

map節點3:

輸入：(text3, 「good weather is good」)

輸出：(good, 1), (weather, 1), (is, 1), (good, 1)

map節點4:

輸入：(text3, 「today has good weather」)

輸出：(today, 1), (has, 1), (good, 1), (weather, 1)

使用3個reduce節點：

 

MapReduce處理方式

MapReduce僞代碼(實現Map和Reduce兩個函數)：

Class Mapper method map(String input_key, String input_value)://input_key: text document name//input_value: document contentsforeach word w in input_value:

EmitIntermediate(w,"1");

Class Reducer method reduce(String output_key, Iterator intermediate_values)://output_key: a word//output_values: a list of countsintresult = 0;foreach v in intermediate_values:

result+=ParseInt(v);

Emit(output_key， result);

3.上升到構架-自動並行化並隱藏低層細節

如何提供統一的計算框架

MapReduce提供一個統一的計算框架，可完成：

—計算任務的劃分和調度

—數據的分佈存儲和劃分

—處理數據與計算任務的同步

—結果數據的收集整理(sorting, combining, partitioning,…)

—系統通訊、負載平衡、計算性能優化處理

—處理系統節點出錯檢測和失效恢復

MapReduce最大的亮點

—經過抽象模型和計算框架把須要作什麼(what need to do)與具體怎麼作(how to do)分開了，爲程序員提供一個抽象和高層的編程接口和框架

—程序員僅須要關心其應用層的具體計算問題，僅需編寫少許的處理應用自己計算問題的程序代碼

—如何具體完成這個並行計算任務所相關的諸多系統層細節被隱藏起來,交給計算框架去處理：從分佈代碼的執行，到大到數千小到單個節點集羣的自動調度使用

MapReduce提供的主要功能

—任務調度：提交的一個計算做業(job)將被劃分爲不少個計算任務(tasks), 任務調度功能主要負責爲這些劃分後的計算任務分配和調度計算節點(map節點或reducer節點); 同時負責監控這些節點的執行狀態, 並負責map節點執行的同步控制(barrier); 也負責進行一些計算性能優化處理, 如對最慢的計算任務採用多備份執行、選最快完成者做爲結果

—數據/代碼互定位：爲了減小數據通訊，一個基本原則是本地化數據處理(locality)，即一個計算節點儘量處理其本地磁盤上所分佈存儲的數據，這實現了代碼向數據的遷移；當沒法進行這種本地化數據處理時，再尋找其它可用節點並將數據從網絡上傳送給該節點(數據向代碼遷移)，但將盡量從數據所在的本地機架上尋找可用節點以減小通訊延遲

—出錯處理：以低端商用服務器構成的大規模MapReduce計算集羣中,節點硬件(主機、磁盤、內存等)出錯和軟件有bug是常態，所以,MapReducer須要能檢測並隔離出錯節點，並調度分配新的節點接管出錯節點的計算任務

—分佈式數據存儲與文件管理：海量數據處理須要一個良好的分佈數據存儲和文件管理系統支撐,該文件系統可以把海量數據分佈存儲在各個節點的本地磁盤上,但保持整個數據在邏輯上成爲一個完整的數據文件；爲了提供數據存儲容錯機制,該文件系統還要提供數據塊的多備份存儲管理能力

—Combiner和Partitioner:爲了減小數據通訊開銷,中間結果數據進入reduce節點前須要進行合併(combine)處理,把具備一樣主鍵的數據合併到一塊兒避免重複傳送; 一個reducer節點所處理的數據可能會來自多個map節點, 所以, map節點輸出的中間結果需使用必定的策略進行適當的劃分(partitioner)處理，保證相關數據發送到同一個reducer節點

參考：http://batxue.com/html/jsgh/20170921/1642.html。

歡迎你們加入架構師微信羣進行交流，可加羣主微信：