mapreduce編程說明和程序流程

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

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

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

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

5、架執(zhí)行排序操作，key 類必須實現(xiàn) WritableComparable接口。一個 Map-Reduce作業(yè)的輸入和輸出類型如下所示：(input) <k1, v1> -> map -> <k2, v2> -> combine -> <k2,v2> -> reduce -> <k3, v3> (output)可編輯修改精品資料作業(yè)配置老接口 JobConf代表一個 Map-Reduce作業(yè)的配置。 JobConf的功能已被新的類 Configuration和 Job 替換。 Configuration類描述了資

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

7、ass()設置實現(xiàn)的 InputFormat類；setOutputKeyClass()設置最終的輸出 key 類；setOutputValueClass()設置最終的輸出 Value 類。默認情況下使用的是FileInputFormat類作為 InputFormat類。MapReduce編程接口參數(shù)作用缺省值其它實可編輯修改精品資料現(xiàn)InputForm將輸入的數(shù)據(jù)集切割TextInputFormatSequenat成小數(shù)據(jù)集(針對文本文件，按行將文本文ceFileInInputSplits,每一個件切割成 InputSplits,并用putForInputSplit將由一個LineRecord

8、Reader將matMapper 負責處理。此 InputSplit解析成外 InputFormat中<key,value>對，key 是行在還提供一個文件中的位置， value是文件RecordReader的實中的一行 )現(xiàn), 將一個InputSplit解析成<key,value>對提供給 map 函數(shù)。OutputFor提供一個TextOutputFormatSequenmatRecordWriter的實(用 LineRecordWriter將最 ceFileO現(xiàn)，負責輸出最終結果終結果寫成純文件文件 ,每個utputFo<key,value> 對一行，

9、key和 rmatvalue 之間用 tab 分隔 )OutputKey輸出的最終結果中LongWritableClasskey 的類型OutputVal輸出的最終結果中Text可編輯修改精品資料ueClassvalue 的類型MapperClaMapper類，實現(xiàn)IdentityMapperLongSussmap 函數(shù)，完成輸入(將輸入的 <key,value>原mRedu的 <key,value>到中封不動的輸出為中間結果 )cer,間結果的映射LogRegexMapper,InverseMapperCombinerC實現(xiàn) combine函數(shù)， nulllass將中

10、間結果中的重復(不對中間結果中的重復keykey 做合并做合并 )ReducerClaReducer類，實現(xiàn)IdentityReducerAccumssreduce函數(shù)，對中間(將中間結果直接輸出為最終ulating結果做合并，形成最終結果 )Reduce結果r,LongSumReducerInputPath設定 job的輸入目錄 ,nulljob 運行時會處理輸入可編輯修改精品資料目錄下的所有文件OutputPat設定 job 的輸出目錄， nullhjob 的最終結果會寫入輸出目錄下MapOutput設定 map 函數(shù)輸出如果用戶沒有設定的話，使用KeyClass的中間結果中 key 的O

11、utputKeyClass類型MapOutput設定 map 函數(shù)輸出如果用戶沒有設定的話，使用ValueClass的中間結果中 valueOutputValuesClass的類型OutputKey對結果中的 key 進行WritableComparableComparato排序時的使用的比較r器Partitioner對中間結果的 key 排HashPartitionerKeyFielClass序后，用此 Partition(使用 Hash 函數(shù)做dBased函數(shù)將其劃分為 R 份 ,partition)Partitio每份由一個 Reducerner負責處理。PipesPartitioner

12、可編輯修改精品資料在這里，我們重點討論如何去實現(xiàn)InputFormat、InputSplit、RecordReader、 Mapper和 Reducer類。新接口可在org.apache.hadoop.mapreduce.*名字空間下找到。相關類的工作流程：（ 1 ）首先是 InputFormat類，該類的 getSplits() 方法生成元素為 InputSplit的 List ，這個類是在 Job 提交的時候被執(zhí)行的，并且createRecordReader()方法創(chuàng)建相應的讀記錄類RecordReader ；（ 2 ）對于每個 InputSplit都由一個 RecordReader類來處

13、理，產(chǎn)生 <K, V> 對，然后將結果交由 Mapper類來處理，一個 InputSplit對應于一個 RecordReader 和一個Mapper ；（ 3 ）通過 RecordReader 類中的 nextKeyValue() 的迭代，每個RecordReader產(chǎn)生的 <K, V> 對，執(zhí)行一次 Mapper類中的 map() ；（4 ）Mapper產(chǎn)生的 <K, V>對經(jīng)過組合形成 <K, List of V>,然后 Reducer 的 reduce() 對這樣的組合進行操作，并產(chǎn)生新的 <K , V >對，有 R 個 Red

14、ucer就對所有的 <K,List of V> 分成 R 份，每份將產(chǎn)生一個輸出，由OutputFormat類來處理。InputFormat<K, V>用戶可以定制自己的InputFormat類，也就是實現(xiàn)一個InputFormat子類，需要注意的是K 類型必須實現(xiàn) WritableComparable接口， V 類型必須實現(xiàn) Writable 接口。對于文件處理可以繼承 FileInputFormat 類，再細一點，如果文本文件，可以直接使用 TextInputFormat 類，如果二進制文件，可以直接使用 SequenceFileInputFormat類。對于 In

15、putFormat的子類，必須實現(xiàn)public List<InputSplit> getSplits(JobContext context)方法和 publicRecordReader<K, V> createRecordReader(InputSplit split,可編輯修改精品資料TaskAttemptContext context)方法。這里牽涉到另外四個類InputSplit、RecordReader、JobContext和 TaskAttemptContext。JobContext是運行期的作業(yè)上下文， TaskAttemptContext是運行期的嘗試任務

16、上下文。InputSplit用戶可以定制自己的InputSplit類，也就是實現(xiàn)一個InputSplit子類，并且實現(xiàn) Writable接口。 InputSplit子類必須實現(xiàn) getLength() 和 StringgetLocation()方法，前者獲取當前Split 的長度，后者獲取Split分布的主機。當然還必須實現(xiàn) Writable接口的 readFields() 和 write() 。對于文件類型的數(shù)據(jù)，可以使用 FileSplit ，該類是以不超過分布在主機的單個塊的大小為Split 塊大小。RecordReader<K, V>用戶可以定制自己的RecordReade

17、r類，也就是實現(xiàn)一個RecordReader子類。 RecordReader子類必須實現(xiàn)void initialize(InputSplit split,TaskAttemptContext context)，boolean nextKeyValue(), KgetCurrentKey(), V getCurrentValue(), float getProgress()和 void close() 。在 initialize() 函數(shù)的參數(shù) split 就是上面的 InputSplit 子類，這類面最關鍵的就是 nextKeyValue() 函數(shù)，它對當前 split 中的 Key 和 Value 進行迭代查找，對于每個 <K, V> 執(zhí)行一次 Mapper 類的 Map() 函數(shù)。對于文本文件，可以使用LineRecordReader類，該類按行來讀取split塊中的內容。Mapper<KIN, VIN, KOUT, VOUT>可編輯修改精品資料用戶可以定制自己的Mapper類，也就是實現(xiàn)一個 Mapper子類。Mapper子類必須實現(xiàn) void map(K, V, Context)方法，每處理一