Hadoop分散式計算1-MapReduce編程模型

Hadoop解決大規模數據分散式計算的方案是MapReduce。MapReduce既是一個編程模型，又是一個計算框架。也就是說，開發人員必須基於MapReduce編程模型進行編程開發，然後將程序通過MapReduce計算框架分發到Hadoop集群中運行。我們先看一下作為編程模型的MapReduce。

MapReduce編程模型

MapReduce是一種非常簡單又非常強大的編程模型。

簡單在於其編程模型只包含map和reduce兩個過程，map的主要輸入是一對<key , value>值，經過map計算後輸出一對<key , value>值；然後將相同key合併，形成<key , value集合>；再將這個<key , value集合>輸入reduce，經過計算輸出零個或多個<key , value>對。

但是MapReduce同時又是非常強大的，不管是關係代數運算（SQL計算），還是矩陣運算（圖計算），大數據領域幾乎所有的計算需求都可以通過MapReduce編程來實現。

我們以WordCount程序為例。WordCount主要解決文本處理中的詞頻統計問題，就是統計文本中每一個單詞出現的次數。如果只是統計一篇文章的詞頻，幾十K到幾M的數據，那麼寫一個程序，將數據讀入內存，建一個Hash表記錄每個詞出現的次數就可以了，如下圖。

但是如果想統計全世界互聯網所有網頁（數萬億計）的詞頻數（這正是google這樣的搜索引擎典型需求），你不可能寫一個程序把全世界的網頁都讀入內存，這時候就需要用MapReduce編程來解決。

WordCount的MapReduce程序如下。

public class WordCount { public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable>{ private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(Object key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasMoreTokens()) { word.set(itr.nextToken()); context.write(word, one); } } } public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); public void reduce(Text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); } }}

其核心是一個map函數，一個reduce函數。

map函數的輸入主要是一個<key , value>對，在這個例子里，value是要統計的所有文本中的一行數據，key在這裡不重要，我們忽略。

public void map(Object key, Text value, Context context )

map函數的計算過程就是，將這行文本中的單詞提取出來，針對每個單詞輸出一個<word , 1>這樣的<key , value>對。

MapReduce計算框架會將這些<word , 1>收集起來，將相同的word放在一起，形成<word , <1,1,1,1,1,1,1.....>>這樣的<key , value集合>數據，然後將其輸入給reduce函數。

public void reduce(Text key, Iterable<IntWritable> values, Context context )

這裡的reduce的輸入參數values就是由很多個1組成的集合，而key就是具體的單詞word。

reduce函數的計算過程就是，將這個集合里的1求和，再將單詞（word）和這個和（sum）組成一個<key , value>(<word , sum>)輸出。每一個輸出就是一個單詞和它的詞頻統計總和。

假設有兩個block的文本數據需要進行詞頻統計，MapReduce計算過程如下圖。

一個map函數可以針對一部分數據進行運算，這樣就可以將一個大數據切分成很多塊（這也正是HDFS所做的），MapReduce計算框架為每個塊分配一個map函數去計算，從而實現大數據的分散式計算。

推薦閱讀：