編程說明和程序流程

1、How To Write Map-Reduce On Hadoop概述Hadoop Map-Reduce是一個使用簡易的軟件框架，基于它寫出來的應(yīng)用程序能夠運行在 由上千個商用機器組成的大型集群上，并以一種可靠容錯的方式并行處理上T級別的數(shù)據(jù)集。一個Map-Reduce 作業(yè)（job） 通常會把輸入的數(shù)據(jù)集切分為若干獨立的數(shù)據(jù)塊，由 map任務(wù)（task）以完全并行的方式處理它們?？蚣軙ap的輸出先進行排序， 然后把結(jié)果輸入給reduce任務(wù)。通常作業(yè)的輸入和輸出都會被存儲在文件系統(tǒng)中。 整個框架負責任務(wù)的調(diào)度和監(jiān)控，以及重新執(zhí)行已經(jīng)失敗的任務(wù)。通常，Map-

2、Reduce框架和分布式文件系統(tǒng)是運行在 一組相同的節(jié)點上的，也就是說，計算節(jié)點和存儲節(jié)點通常在一起。這種配置允許框架在那些 已經(jīng)存好數(shù)據(jù)的節(jié)點上高效地調(diào)度任務(wù)，這可以使整個集群的網(wǎng)絡(luò)帶寬被非常高效地利用。Map-Reduce框架由單獨一個master JobTracker 和每個集群節(jié)點一個 slave TaskTracker共同組成。這個master負責調(diào)度構(gòu)成一個作業(yè)的所有任務(wù)， 這些任務(wù)分布在不同的slave上，master監(jiān)控它們的執(zhí)行，重新執(zhí)行已經(jīng)失敗的任務(wù)。而slave 僅負責執(zhí)行由master指派的任務(wù)。應(yīng)用程序至少應(yīng)該指明輸入/輸出的位置（路徑）

3、，并通過實現(xiàn)合適的接口或抽象類提供map和reduce函數(shù)。 再加上其他作業(yè)的參數(shù)，就構(gòu)成了作業(yè)配置（job configuration）。然后，Hadoop的job client 提交作業(yè)（jar包/可執(zhí)行程序 等）和配置信息給JobTracker，后者負責分發(fā)這些軟件和配置信息 給slave、調(diào)度任務(wù)且監(jiān)控它們的執(zhí)行，同時提供狀態(tài)和診斷信息給job-client。雖然Hadoop框架是用JavaTM實現(xiàn)的，但Map-Reduce應(yīng)用程序則不一定要用 Java來寫 。· Hadoop Streaming是一種運行作業(yè)的實用工具，它允許用戶創(chuàng)建和運行任何可執(zhí)行程序 （例如：

4、Shell工具）來做為mapper和reducer。· Hadoop Pipes是一個與SWIG兼容的C+ API （沒有基于JNITM技術(shù)），它也可用于實現(xiàn)Map-Reduce應(yīng)用程序。輸入與輸出Map-Reduce框架運轉(zhuǎn)在<key, value> 鍵值對上，也就是說， 框架把作業(yè)的輸入看為是一組<key, value> 鍵值對，同樣也產(chǎn)出一組 <key, value> 鍵值對做為作業(yè)的輸出，這兩組鍵值對的類型可能不同。框架需要對key和value的類(classes)進行序列化操作， 因此，這些類需要實

5、現(xiàn) Writable接口。 另外，為了方便框架執(zhí)行排序操作，key類必須實現(xiàn) WritableComparable接口。一個Map-Reduce 作業(yè)的輸入和輸出類型如下所示：(input) <k1, v1> -> map -> <k2, v2> -> combine -> <k2, v2> -> reduce -> <k3, v3> (output)作

6、業(yè)配置老接口JobConf代表一個Map-Reduce作業(yè)的配置。JobConf的功能已被新的類Configuration和Job替換。Configuration類描述了資源，這些資源大多都是從XML配置文件中讀取的屬性和值組成。比如來自core-default.xml和core-site.xml。Job描述了用戶角度的視圖，它允許用戶配置、提交、控制它的執(zhí)行和查詢狀態(tài)。只有在作業(yè)提交之后才能使用set方法。setNumReduceTasks() 設(shè)置Reduce的任務(wù)數(shù)；setJarByClass() 設(shè)置Jar包的類來源；setMapperClass()設(shè)置實現(xiàn)的Mapper類；setCo

7、mbinerClass()設(shè)置實現(xiàn)的Combiner類；setReducerClass()設(shè)置實現(xiàn)的Reducer類；setInputFormatClass()設(shè)置實現(xiàn)的InputFormat類；setOutputKeyClass()設(shè)置最終的輸出key類；setOutputValueClass()設(shè)置最終的輸出Value類。默認情況下使用的是FileInputFormat類作為InputFormat類。MapReduce編程接口參數(shù)作用缺省值其它實現(xiàn)InputFormat 將輸入的數(shù)據(jù)集切割成小數(shù)據(jù)集 InputSplits, 每一個 InputSplit 將由一個 Mapper 負責處理。

8、此外 InputFormat 中還提供一個 RecordReader 的實現(xiàn), 將一個 InputSplit 解析成 <key,value> 對提供給 map 函數(shù)。TextInputFormat(針對文本文件，按行將文本文件切割成 InputSplits, 并用 LineRecordReader 將 InputSplit 解析成 <key,value> 對，key 是行在文件中的位置，value 是文件中的一行) SequenceFileInputFormat OutputFormat 提供一個 RecordWriter 的實現(xiàn)，負責輸出最終結(jié)果TextOutputF

9、ormat(用 LineRecordWriter 將最終結(jié)果寫成純文件文件,每個 <key,value> 對一行，key 和 value 之間用 tab 分隔) SequenceFileOutputFormatOutputKeyClass 輸出的最終結(jié)果中 key 的類型LongWritable  OutputValueClass 輸出的最終結(jié)果中 value 的類型Text  MapperClass Mapper 類，實現(xiàn) map 函數(shù)，完成輸入的 <key,value> 到中間結(jié)果的映射IdentityMapper(將輸入的 <key,va

10、lue> 原封不動的輸出為中間結(jié)果) LongSumReducer,LogRegexMapper,InverseMapper CombinerClass 實現(xiàn) combine 函數(shù)，將中間結(jié)果中的重復(fù) key 做合并null(不對中間結(jié)果中的重復(fù) key 做合并)  ReducerClass Reducer 類，實現(xiàn) reduce 函數(shù)，對中間結(jié)果做合并，形成最終結(jié)果IdentityReducer(將中間結(jié)果直接輸出為最終結(jié)果) AccumulatingReducer, LongSumReducer InputPath 設(shè)定 job 的輸入目錄, job 運行時會處理輸入目錄下

11、的所有文件null  OutputPath 設(shè)定 job 的輸出目錄，job 的最終結(jié)果會寫入輸出目錄下null  MapOutputKeyClass 設(shè)定 map 函數(shù)輸出的中間結(jié)果中 key 的類型如果用戶沒有設(shè)定的話，使用 OutputKeyClass  MapOutputValueClass 設(shè)定 map 函數(shù)輸出的中間結(jié)果中 value 的類型如果用戶沒有設(shè)定的話，使用 OutputValuesClass  OutputKeyComparator 對結(jié)果中的 key 進行排序時的使用的比較器WritableComparable  Pa

12、rtitionerClass 對中間結(jié)果的 key 排序后，用此 Partition 函數(shù)將其劃分為R份,每份由一個 Reducer 負責處理。HashPartitioner(使用 Hash 函數(shù)做 partition) KeyFieldBasedPartitioner PipesPartitioner 在這里，我們重點討論如何去實現(xiàn)InputFormat、InputSplit、RecordReader、Mapper和Reducer類。新接口可在org.apache.hadoop.mapreduce.*名字空間下找到。相關(guān)類的工作流程：（1）首先是InputFormat類，該類的getSpli

13、ts()方法生成元素為InputSplit的List，這個類是在Job提交的時候被執(zhí)行的，并且createRecordReader()方法創(chuàng)建相應(yīng)的讀記錄類RecordReader；（2）對于每個InputSplit都由一個RecordReader類來處理，產(chǎn)生<K, V>對，然后將結(jié)果交由Mapper類來處理，一個InputSplit對應(yīng)于一個RecordReader和一個Mapper；（3）通過RecordReader類中的nextKeyValue()的迭代，每個RecordReader產(chǎn)生的<K, V>對，執(zhí)行一次Mapper類中的map()；（4）Mapper產(chǎn)

14、生的<K, V>對經(jīng)過組合形成<K, List of V>, 然后Reducer的reduce()對這樣的組合進行操作，并產(chǎn)生新的<K, V>對，有R個Reducer就對所有的<K, List of V>分成R份，每份將產(chǎn)生一個輸出，由OutputFormat類來處理。InputFormat<K, V>用戶可以定制自己的InputFormat類，也就是實現(xiàn)一個InputFormat子類，需要注意的是K類型必須實現(xiàn)WritableComparable接口，V類型必須實現(xiàn)Writable接口。對于文件處理可以繼承FileInputForm

15、at類，再細一點，如果文本文件，可以直接使用TextInputFormat類，如果二進制文件，可以直接使用SequenceFileInputFormat類。對于InputFormat的子類，必須實現(xiàn)public List<InputSplit> getSplits(JobContext context)方法和public RecordReader<K, V> createRecordReader(InputSplit split, TaskAttemptContext context)方法。這里牽涉到另外四個類InputSplit、RecordReader、JobCon

16、text和TaskAttemptContext。JobContext是運行期的作業(yè)上下文，TaskAttemptContext是運行期的嘗試任務(wù)上下文。InputSplit用戶可以定制自己的InputSplit類，也就是實現(xiàn)一個InputSplit子類，并且實現(xiàn)Writable接口。InputSplit子類必須實現(xiàn)getLength()和String getLocation()方法，前者獲取當前Split的長度，后者獲取Split分布的主機。當然還必須實現(xiàn)Writable接口的readFields()和write()。對于文件類型的數(shù)據(jù)，可以使用FileSplit，該類是以不超過分布在主機的單

17、個塊的大小為Split塊大小。RecordReader<K, V>用戶可以定制自己的RecordReader類，也就是實現(xiàn)一個RecordReader子類。RecordReader子類必須實現(xiàn)void initialize(InputSplit split, TaskAttemptContext context)，boolean nextKeyValue(), K getCurrentKey(), V getCurrentValue(), float getProgress()和void close()。在initialize()函數(shù)的參數(shù)split就是上面的InputSplit子類，這類面最關(guān)鍵的就是nextKeyValue()函數(shù)，它對當前split中的Key和Value進行迭代查找，對于每個<K, V>執(zhí)行一次Mapper類的Map()函數(shù)。對于文本文件，可以使用LineRecordReader類，該類按行來讀取split塊中的內(nèi)容。Mapper<KIN, VIN, KOUT, VOUT>用戶可以定制自己的Mapper類，也就是實現(xiàn)一個Mapper子類。Mapper子類必須實現(xiàn)void map(K, V, C