編程說明和程序流程

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

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

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

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

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

6、業(yè)配置老接口JobConf代表一個(gè)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è)置實(shí)現(xiàn)的Mapper類；setCo

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

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

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

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

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

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

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

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

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

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

17、個(gè)塊的大小為Split塊大小。RecordReader<K, V>用戶可以定制自己的RecordReader類，也就是實(shí)現(xiàn)一個(gè)RecordReader子類。RecordReader子類必須實(shí)現(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ù)，它對當(dāng)前split中的Key和Value進(jìn)行迭代查找，對于每個(gè)<K, V>執(zhí)行一次Mapper類的Map()函數(shù)。對于文本文件，可以使用LineRecordReader類，該類按行來讀取split塊中的內(nèi)容。Mapper<KIN, VIN, KOUT, VOUT>用戶可以定制自己的Mapper類，也就是實(shí)現(xiàn)一個(gè)Mapper子類。Mapper子類必須實(shí)現(xiàn)void map(K, V, C