1970年，IBM的研究員E.F.Codd博士在刊物《Communication of the ACM》上發表了壹篇名為“A Relational Model of Data for Large Shared Data Banks”的論文，提出了關系模型的概念，標誌著關系數據庫的誕生，隨後幾十年，關系數據庫及其結構化查詢語言SQL成為程序員必須掌握的基本技能之壹。
2005年4月，Jeffrey Dean和Sanjay Ghemawat在國際會議OSDI上發表“MapReduce: Simplified Data Processing on Large Cluster”，標誌著google的大規模數據處理系統MapReduce公開。受這篇論文的啟發，當年秋天，Hadoop 由 Apache Software Foundation 公司作為 Lucene 的子項目 Nutch 的壹部分正式被引入，2006 年 3 月份，MapReduce 和 Nutch Distributed File System (NDFS) 分別被納入稱為 Hadoop 的項目中。如今，Hadoop已經被超過50%的互聯網公司使用，其他很多公司正準備使用Hadoop來處理海量數據，隨著Hadoop越來越受歡迎，也許在將來的某段時間，Hadoop會成為程序員必須掌握的技能之壹，如果真是這樣的話，學會如何在Hadoop上編寫MapReduce程序便是學習Hadoop的開始。
本文介紹了在Hadoop上編寫MapReduce程序的基本方法，包括MapReduce程序的構成，不同語言開發MapReduce的方法等。
2. Hadoop 作業構成
2.1 Hadoop作業執行流程
用戶配置並將壹個Hadoop作業提到Hadoop框架中，Hadoop框架會把這個作業分解成壹系列map tasks 和reduce tasks。Hadoop框架負責task分發和執行，結果收集和作業進度監控。
下圖給出了壹個作業從開始執行到結束所經歷的階段和每個階段被誰控制（用戶 or Hadoop框架）。
下圖詳細給出了用戶編寫MapRedue作業時需要進行那些工作以及Hadoop框架自動完成的工作：
在編寫MapReduce程序時，用戶分別通過InputFormat和OutputFormat指定輸入和輸出格式，並定義Mapper和Reducer指定map階段和reduce階段的要做的工作。在Mapper或者Reducer中，用戶只需指定壹對key/value的處理邏輯，Hadoop框架會自動順序叠代解析所有key/value，並將每對key/value交給Mapper或者Reducer處理。表面上看來，Hadoop限定數據格式必須為key/value形式，過於簡單，很難解決復雜問題，實際上，可以通過組合的方法使key或者value（比如在key或者value中保存多個字段，每個字段用分隔符分開，或者value是個序列化後的對象，在Mapper中使用時，將其反序列化等）保存多重信息，以解決輸入格式較復雜的應用。
2.2 用戶的工作
用戶編寫MapReduce需要實現的類或者方法有：
（1） InputFormat接口
用戶需要實現該接口以指定輸入文件的內容格式。該接口有兩個方法
1
2
3
4
5
6
7
8
9
10
11
public interface InputFormat<K, V> {
InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;
RecordReader<K, V> getRecordReader(InputSplit split,
JobConf job,
Reporter reporter) throws IOException;
}
其中getSplits函數將所有輸入數據分成numSplits個split，每個split交給壹個map task處理。getRecordReader函數提供壹個用戶解析split的叠代器對象，它將split中的每個record解析成key/value對。
Hadoop本身提供了壹些InputFormat：
（2）Mapper接口
用戶需繼承Mapper接口實現自己的Mapper，Mapper中必須實現的函數是
1
2
3
4
5
6
7
8
9
void map(K1 key,
V1 value,
OutputCollector<K2,V2> output,
Reporter reporter
) throws IOException
其中，<K1 V1>是通過Inputformat中的RecordReader對象解析處理 的，OutputCollector獲取map()的輸出結果，Reporter保存了當前task處理進度。
Hadoop本身提供了壹些Mapper供用戶使用：
（3）Partitioner接口
用戶需繼承該接口實現自己的Partitioner以指定map task產生的key/value對交給哪個reduce task處理，好的Partitioner能讓每個reduce task處理的數據相近，從而達到負載均衡。Partitioner中需實現的函數是
getPartition( K2 key, V2 value, int numPartitions)
該函數返回<K2 V2>對應的reduce task ID。
用戶如果不提供Partitioner，Hadoop會使用默認的（實際上是個hash函數）。
（4）Combiner
Combiner使得map task與reduce task之間的數據傳輸量大大減小，可明顯提高性能。大多數情況下，Combiner與Reducer相同。
（5）Reducer接口
用戶需繼承Reducer接口實現自己的Reducer，Reducer中必須實現的函數是
1
2
3
4
5
6
7
8
9
void reduce(K2 key,
Iterator<V2> values,
OutputCollector<K3,V3> output,
Reporter reporter
) throws IOException
Hadoop本身提供了壹些Reducer供用戶使用：
（6）OutputFormat
用戶通過OutputFormat指定輸出文件的內容格式，不過它沒有split。每個reduce task將其數據寫入自己的文件，文件名為part-nnnnn，其中nnnnn為reduce task的ID。
Hadoop本身提供了幾個OutputFormat:
3. 分布式緩存
Haoop中自帶了壹個分布式緩存，即DistributedCache對象，方便map task之間或者reduce task之間***享壹些信息，比如某些實際應用中，所有map task要讀取同壹個配置文件或者字典，則可將該配置文件或者字典放到分布式緩存中。
4. 多語言編寫MapReduce作業
Hadoop采用java編寫，因而Hadoop天生支持java語言編寫作業，但在實際應用中，有時候，因要用到非java的第三方庫或者其他原因，要采用C/C++或者其他語言編寫MapReduce作業，這時候可能要用到Hadoop提供的壹些工具。
如果妳要用C/C++編寫MpaReduce作業，可使用的工具有Hadoop Streaming或者Hadoop Pipes。
如果妳要用Python編寫MapReduce作業，可以使用Hadoop Streaming或者Pydoop。
如果妳要使用其他語言，如shell，php，ruby等，可使用Hadoop Streaming。
關於Hadoop Streaming編程，可參見我的這篇博文：《Hadoop Streaming編程》（/projects/pydoop/
關於Hadoop pipes編程，可參見《Hadoop Tutorial 2.2 — Running C++ Programs on Hadoop》。
5. 編程方式比較
（1）java。 Hadoop支持的最好最全面的語言，而且提供了很多工具方便程序員開發。
（2）Hadoop Streaming。 它最大的優點是支持多種語言，但效率較低，reduce task需等到map 階段完成後才能啟動；它不支持用戶自定義InputFormat，如果用戶想指定輸入文件格式，可使用java語言編寫或者在命令行中指定分隔符；它采用標準輸入輸出讓C/C++與java通信，因而只支持text數據格式。
（3）Hadoop Pipes。 專門為C/C++語言設計，由於其采用了socket方式讓C/C++與java通信，因而其效率較低（其優勢在於，但作業需要大量，速度很快）。它支持用戶（用C/C++）編寫RecordReader。
（4）Pydoop。它是專門方便python程序員編寫MapReduce作業設計的，其底層使用了Hadoop Streaming接口和libhdfs庫。
6. 總結
Hadoop使得分布式程序的編寫變得異常簡單，很多情況下，用戶只需寫map()和reduce()兩個函數即可（InputFormat，Outputformat可用系統缺省的）。正是由於Hadoop編程的簡單性，越來越多的公司或者研究單位開始使用Hadoop。