MapReduce編程模型課件

1、mapreduce編程模型1mapreduce編程模型mapreduce編程模型2概述mapreduce是在總結(jié)大量應(yīng)用的共同特點(diǎn)的基礎(chǔ)上抽象出來的分布式框架，其適用應(yīng)用場景的共同特點(diǎn)是：任務(wù)可被分解為相互獨(dú)立的子問題?；谠撎攸c(diǎn)， mapreduce編程模型給出了其分布式編程方法，共分5個步驟：迭代迭代（iteration）。遍歷輸入數(shù)據(jù)，并將之解析成key/value對。將輸入的key/value對映射映射（map）成另外一些key/value對。依據(jù)key對中間數(shù)據(jù)進(jìn)行分組分組（group）。以組為單位對數(shù)據(jù)進(jìn)行規(guī)約規(guī)約（reduce）。迭代迭代。將最終產(chǎn)生的key/value對保存到輸

2、出文件。好處：組件化和并行化。mapreduce編程模型3mapreduce編程接口體系結(jié)構(gòu)整個編程模型位于應(yīng)用程序?qū)雍蚼apreduce執(zhí)行器之間，可以分為兩層。第一層是最基本的java api，主要有5個可編程組件，分別是inputformat、mapper、partitioner、reducer和outputformat。hadoop自帶了很多直接可用的inputformat、partitioner和outputformat，大部分情況下，用戶只需編寫mapper和reducer即可。第二層是工具層，位于基本java api之上，主要是為了方便用戶編寫復(fù)雜的mapreduce程序和利用其

3、他編程語言增加mapreduce計(jì)算平臺的兼容性而提出來的。mapreduce編程模型4新舊mapreduce api比較從0.20.0版本開始，hadoop同時提供了新舊兩套mapreduce api。新api在舊api基礎(chǔ)上進(jìn)行封裝，使其在擴(kuò)展性和易用性方面更好。二者區(qū)別如下：存放位置-舊api放在org.apache.hadoop.mapred包中；-新api放在org.apache.hadoop.mapreduce包及其子包中。接口變?yōu)槌橄箢?接口，嚴(yán)格的“協(xié)議約束”，只有方法聲明而沒有方法實(shí)，要求所有實(shí)現(xiàn)類（抽象類除外）必須實(shí)現(xiàn)接口中的每個方法。-抽象類，較寬松的“約束協(xié)議”，可為某

4、些方法提供默認(rèn)實(shí) 現(xiàn)，而繼承類則可選擇是否重新實(shí)現(xiàn)這些方法。故而抽象類在類衍化方面更有優(yōu)勢，即具有良好的向后兼容性。上下文封裝-新版api將變量和函數(shù)封裝成各種上下文（context）類，使得api具有更好的易用性和擴(kuò)展性。mapreduce編程模型5mapreduce api基本概念序列化序列化序列化是指將結(jié)構(gòu)化對象轉(zhuǎn)化為字節(jié)流以便于通過網(wǎng)絡(luò)進(jìn)行傳輸或?qū)懭氤志么鎯Φ倪^程。在hadoop mapreduce中，序列化的主要作用是永久存儲和進(jìn)程間通信。為能夠讀取或者存儲java對象，mapreduce編程模型要求用戶輸入和輸出數(shù)據(jù)中的key和value必須是可序列化的。使一個java對象可序列化

5、的方法是讓其對應(yīng)的實(shí)現(xiàn)writable接口。而作為數(shù)據(jù)排序的關(guān)鍵字key所對應(yīng)的類需要實(shí)現(xiàn)writablecomparable接口。mapreduce編程模型6mapreduce api基本概念reporter參數(shù)參數(shù)reporter是mapreduce提供給應(yīng)用程序的工具。如圖3-4所示，應(yīng)用程序可使用reporter中的方法報告完成進(jìn)度（progress）、設(shè)定狀態(tài)消息（setstatus）以及更新計(jì)數(shù)器（incrcounter）。reporter是一個基礎(chǔ)參數(shù)。mapreduce對外提供的大部分組件，包括inputformat、mapper和reducer等，均在其主要方法中添加了該參數(shù)

6、。mapreduce編程模型7mapreduce api基本概念回調(diào)機(jī)制回調(diào)機(jī)制回調(diào)機(jī)制是一種常見的設(shè)計(jì)模式。它將工作流內(nèi)的某個功能按照約定的接口暴露給外部使用者，為外部使用者提供數(shù)據(jù)，或要求外部使用者提供數(shù)據(jù)。hadoop mapreduce對外提供的5個組件（inputformat、mapper、partitioner、reducer和outputformat）實(shí)際上全部屬于回調(diào)接口。當(dāng)用戶按照約定實(shí)現(xiàn)這幾個接口后，mapreduce運(yùn)行時環(huán)境會自動調(diào)用它們。mapreduce編程模型8mapreduce api基本概念回調(diào)機(jī)制回調(diào)機(jī)制如圖3-5所示，mapreduce給用戶暴露了接口ma

7、pper，當(dāng)用戶按照自己的應(yīng)用程序邏輯實(shí)現(xiàn)自己的mymapper后，hadoop mapreduce運(yùn)行時環(huán)境會將輸入數(shù)據(jù)解析成key/value對，并調(diào)用map()函數(shù)迭代處理。mapreduce編程模型9java api解析作業(yè)配置與提交inputformat接口的設(shè)計(jì)與實(shí)現(xiàn)outputformat接口的設(shè)計(jì)與實(shí)現(xiàn)mapper與reducer的解析partitioner接口的設(shè)計(jì)與實(shí)現(xiàn)mapreduce編程模型10作業(yè)配置與提交hadoop配置文件介紹在hadoop中，common、hdfs和mapreduce各有對應(yīng)的配置文件，用于保存對應(yīng)模塊中可配置的參數(shù)。系統(tǒng)默認(rèn)配置文件分別是cor

8、e-default.xml、hdfs-default.xml和mapred-default.xml，它們包含了所有可配置屬性的默認(rèn)值。而管理員自定義配置文件分別是core-site.xml、hdfs-site.xml和mapred-site.xml。它們由管理員設(shè)置，主要用于定義一些新的配置屬性或者覆蓋系統(tǒng)默認(rèn)配置文件中的默認(rèn)值。通常這些配置一旦確定，便不能被修改（如果想修改，需重新啟動hadoop）。需要注意的是，core-default.xml和core-site.xml屬于公共基礎(chǔ)庫的配置文件，默認(rèn)情況下，hadoop總會優(yōu)先加載它們。在hadoop中，每個配置屬性主要包括三個配置參數(shù)：

9、name、value和description，分別表示屬性名、屬性值和屬性描述。hadoop為配置文件添加了兩個新的特性：final參數(shù)和變量擴(kuò)展。mapreduce編程模型11作業(yè)配置與提交mapreduce作業(yè)配置與提交在mapreduce中，每個作業(yè)由兩部分組成：應(yīng)用程序和作業(yè)配置。其中，作業(yè)配置內(nèi)容包括環(huán)境配置和用戶自定義配置兩部分。環(huán)境配置由hadoop自動添加，主要由mapred-default.xml和mapred-site.xml兩個文件中的配置選項(xiàng)組合而成；用戶自定義配置則由用戶自己根據(jù)作業(yè)特點(diǎn)個性化定制而成，比如用戶可設(shè)置作業(yè)名稱，以及mapper/reducer、redu

10、ce task個數(shù)等。mapreduce編程模型12作業(yè)配置與提交mapreduce作業(yè)配置與提交新版api用job類代替了jobconf和jobclient兩個類，這樣，僅使用一個類的同時可完成作業(yè)配置和作業(yè)提交相關(guān)功能，進(jìn)一步簡化了作業(yè)編寫方式。mapreduce編程模型13作業(yè)配置與提交舊api中的作業(yè)配置mapreduce配置模塊代碼結(jié)構(gòu)如圖3-6所示。其中，org.apache.hadoop.conf中的configuration類是配置模塊最底層的類。從圖3-6中可以看出，該類支持以下兩種基本操作。序列化：序列化是將結(jié)構(gòu)化數(shù)據(jù)轉(zhuǎn)換成字節(jié)流，以便于傳輸或存儲。java實(shí)現(xiàn)了自己的一套

11、序列化框架。凡是需要支持序列化的類，均需要實(shí)現(xiàn)writable接口。迭代：為了方便遍歷所有屬性，它實(shí)現(xiàn)了java開發(fā)包中的iterator接口。mapreduce編程模型14作業(yè)配置與提交舊api中的作業(yè)配置configuration類總會依次加載core-default.xml和core-site.xml兩個基礎(chǔ)配置文件。jobconf類描述了一個mapreduce作業(yè)運(yùn)行時需要的所有信息，而mapreduce運(yùn)行時環(huán)境正是根據(jù)jobconf提供的信息運(yùn)行作業(yè)的。jobconf繼承了configuration類，并添加了一些設(shè)置/獲取作業(yè)屬性的setter/getter函數(shù)，以方便用戶編寫m

12、apreduce程序。默認(rèn)情況下，jobconf會自動加載配置文件mapred-default.xml和mapred-site.xmlmapreduce編程模型15作業(yè)配置與提交新api中的作業(yè)配置與新api中的作業(yè)配置相關(guān)的類是job，該類同時具有作業(yè)配置和作業(yè)提交的功能。作業(yè)配置部分的類圖如圖3-7所示。job類繼承了一個新類jobcontext，而context自身則包含一個jobconf類型的成員。注意，jobcontext類僅提供了一些getter方法，而job類中則提供了一些setter方法。mapreduce編程模型16inputformat接口的設(shè)計(jì)與實(shí)現(xiàn)inputformat

13、主要用于描述輸入數(shù)據(jù)的格式，它提供以下兩個功能。 數(shù)據(jù)切分：按照某個策略將輸入數(shù)據(jù)切分成若干個split，以便確定map task個數(shù)以及對應(yīng)的split。 為mapper提供輸入數(shù)據(jù)：給定某個split，能將其解析成一個個key/value對。mapreduce編程模型17inputformat接口的設(shè)計(jì)與實(shí)現(xiàn)舊版api的inputformat解析如圖3-8所示，在舊api中inputformat是一個接口，包含兩種方法。getsplits方法主要完成數(shù)據(jù)切分的功能，它會嘗試著將輸入數(shù)據(jù)切分成numsplits個inputsplit。getrecordreader方法返回一個recordre

14、ader對象，該對象可將輸入的inputsplit解析成若干個key/value對。mapreduce框架在map task執(zhí)行過程中，會不斷調(diào)用recordreader對象中的方法，迭代獲取key/value對并交給map()函數(shù)處理。mapreduce編程模型18inputformat接口的設(shè)計(jì)與實(shí)現(xiàn)inputformat的實(shí)現(xiàn)所有基于文件的inputformat實(shí)現(xiàn)的基類是，它最重要的功能是為各種inputformat提供統(tǒng)一的getsplits函數(shù)。該函數(shù)實(shí)現(xiàn)中最核心的兩個算法是文件切分算法和host選擇算法。（1）文件切分算法）文件切分算法文件切分算法主要用于確定inputsplit

15、的個數(shù)以及每個inputsplit對應(yīng)的數(shù)據(jù)段。以文件為單位切分生成inputsplit。對于每個文件，由以下三個屬性值確定其對應(yīng)的inputsplit的個數(shù)。goalsize：它是根據(jù)用戶期望的inputsplit數(shù)目計(jì)算出來的，即totalsize/numsplits。其中，totalsize為文件總大??；numsplits為用戶設(shè)定的map task個數(shù)，默認(rèn)情況下是1。minsize：inputsplit的最小值，由配置參數(shù)mapred.min.split.size確定，默認(rèn)是1。blocksize：文件在hdfs中存儲的block大小，不同文件可能不同，默認(rèn)是64 mb。三個參數(shù)共同

16、決定inputsplit的最終大小，計(jì)算方法：splitsize = maxminsize, mingoalsize, blocksize mapreduce編程模型19inputformat接口的設(shè)計(jì)與實(shí)現(xiàn)（2）host選擇算法選擇算法待inputsplit切分方案確定后，下一步要確定每個inputsplit的元數(shù)據(jù)信息。這通常由四部分組成：，分別表示inputsplit所在的文件、起始位置、長度以及所在的host（節(jié)點(diǎn)）列表。其中，前三項(xiàng)很容易確定，難點(diǎn)在于host列表的選擇方法。inputsplit的host列表選擇策略直接影響到運(yùn)行過程中的任務(wù)本地性。hdfs上的文件是以block為單

17、位組織的，一個大文件對應(yīng)的block可能遍布整個hadoop集群，而inputsplit的劃分算法可能導(dǎo)致一個inputsplit對應(yīng)多個block ，這些block可能位于不同節(jié)點(diǎn)上，這使得hadoop不可能實(shí)現(xiàn)完全的數(shù)據(jù)本地性。inputsplit對應(yīng)的block可能位于多個節(jié)點(diǎn)上，但考慮到任務(wù)調(diào)度的效率，通常不會把所有節(jié)點(diǎn)加到inputsplit的host列表中，而是選擇包含（該inputsplit）數(shù)據(jù)總量最大的前幾個節(jié)點(diǎn)（hadoop限制最多選擇10個，多余的會過濾掉），以作為任務(wù)調(diào)度時判斷任務(wù)是否具有本地性的主要憑證。為此，設(shè)計(jì)了一個簡單有效的啟發(fā)式算法：首先按照rack包含的數(shù)據(jù)

18、量對rack進(jìn)行排序，然后在rack內(nèi)部按照每個node包含的數(shù)據(jù)量對node排序，最后取前n個node的host作為inputsplit的host列表，這里的n為block副本數(shù)。這樣，當(dāng)任務(wù)調(diào)度器調(diào)度task時，只要將task調(diào)度給位于host列表的節(jié)點(diǎn)，就認(rèn)為該task滿足本地性。mapreduce編程模型20inputformat接口的設(shè)計(jì)與實(shí)現(xiàn)【實(shí)例】某個hadoop集群的網(wǎng)絡(luò)拓?fù)浣Y(jié)構(gòu)如圖3-10所示，hdfs中block副本數(shù)為3，某個inputsplit包含3個block，大小依次是100、150和75，很容易計(jì)算，4個rack包含的（該inputsplit的）數(shù)據(jù)量分別是175

19、、250、150和75。rack2中的node3和node4， rack1中的node1將被添加到該inputsplit的host列表中。從以上host選擇算法可知，當(dāng)inputsplit尺寸大于block尺寸時，map task并不能實(shí)現(xiàn)完全數(shù)據(jù)本地性，也就是說，總有一部分?jǐn)?shù)據(jù)需要從遠(yuǎn)程節(jié)點(diǎn)上讀取，因而當(dāng)使用基于實(shí)現(xiàn)inputformat時，為了提高map task的數(shù)據(jù)本地性，應(yīng)盡量使inputsplit大小與block大小相同。mapreduce編程模型21inputformat接口的設(shè)計(jì)與實(shí)現(xiàn)新版api的inputformat解析新版api的inputformat類圖如圖3-11所示。

20、新api與舊api比較，在形式上發(fā)生了較大變化，但仔細(xì)分析，發(fā)現(xiàn)僅僅是對之前的一些類進(jìn)行了封裝。通過封裝，使接口的易用性和擴(kuò)展性得以增強(qiáng)。mapreduce編程模型22inputformat接口的設(shè)計(jì)與實(shí)現(xiàn)新版api的inputformat解析此外，對于基類，新api中有一個值得注意的改動：inputsplit劃分算法不再考慮用戶設(shè)定的map task個數(shù)，而用mapred.max.split.size（記為maxsize）代替，即inputsplit大小的計(jì)算公式變?yōu)椋簊plitsize = maxminsize, minmaxsize, blocksize mapreduce編程模型23o

21、utputformat接口的設(shè)計(jì)與實(shí)現(xiàn)outputformat主要用于描述輸出數(shù)據(jù)的格式，它能夠?qū)⒂脩籼峁┑膋ey/value對寫入特定格式的文件中。舊版api的outputformat解析如圖3-12所示，在舊版api中，outputformat是一個接口，它包含兩個方法。checkoutputspecs方法一般在用戶作業(yè)被提交到j(luò)obtracker之前，由jobclient自動調(diào)用，以檢查輸出目錄是否合法。getrecordwriter方法返回一個recordwriter類對象。該類中的方法write接收一個key/value對，并將之寫入文件。在task執(zhí)行過程中，mapreduce框架

22、會將map()或者reduce()函數(shù)產(chǎn)生的結(jié)果傳入write方法。mapreduce編程模型24outputformat接口的設(shè)計(jì)與實(shí)現(xiàn)outputformat的實(shí)現(xiàn)基類需要提供所有基于文件的outputformat實(shí)現(xiàn)的公共功能，總結(jié)起來，主要有以下兩個：（1）實(shí)現(xiàn)checkoutputspecs接口該接口在作業(yè)運(yùn)行之前被調(diào)用，默認(rèn)功能是檢查用戶配置的輸出目錄是否存在，如果存在則拋出異常，以防止之前的數(shù)據(jù)被覆蓋。（2）處理side-effect file任務(wù)的side-effect file并不是任務(wù)的最終輸出文件，而是具有特殊用途的任務(wù)專屬文件。它的典型應(yīng)用是執(zhí)行推測式任務(wù)。在hadoo

23、p中，因?yàn)橛布匣?、網(wǎng)絡(luò)故障等原因，同一個作業(yè)的某些任務(wù)執(zhí)行速度可能明顯慢于其他任務(wù)，這種任務(wù)會拖慢整個作業(yè)的執(zhí)行速度。為了對這種“慢任務(wù)”進(jìn)行優(yōu)化，hadoop會為之在另外一個節(jié)點(diǎn)上啟動一個相同的任務(wù)，該任務(wù)便被稱為推測式任務(wù)，最先完成任務(wù)的計(jì)算結(jié)果便是這塊數(shù)據(jù)對應(yīng)的處理結(jié)果。為防止這兩個任務(wù)同時往一個輸出文件中寫入數(shù)據(jù)時發(fā)生寫沖突，會為每個task的數(shù)據(jù)創(chuàng)建一個side-effect file，并將產(chǎn)生的數(shù)據(jù)臨時寫入該文件，待task完成后，再移動到最終輸出目錄中。mapreduce編程模型25outputformat接口的設(shè)計(jì)與實(shí)現(xiàn)side-effect file的相關(guān)操作，比如創(chuàng)建、刪

24、除、移動等，均由outputcommitter完成。它是一個接口， hadoop提供了默認(rèn)實(shí)現(xiàn)，用戶也可以根據(jù)自己的需求編寫outputcommitter實(shí)現(xiàn)，并通過參數(shù)mitter.class指定。outputcommitter接口定義以及對應(yīng)的實(shí)現(xiàn)如表3-2所示。表3-2outputcommitter接口定義以及對應(yīng)的實(shí)現(xiàn)mapreduce編程模型26outputformat接口的設(shè)計(jì)與實(shí)現(xiàn)新版api的outputformat解析如圖3-14所示，除了接口變?yōu)槌橄箢愅?，新api中的outputformat增加了一個新的方法：getoutputcommitter，以允許用戶自己定制合適的ou

25、tputcommitter實(shí)現(xiàn)。mapreduce編程模型27mapper與reducer解析舊版api的mapper/reducer解析mapper/reducer中封裝了應(yīng)用程序的數(shù)據(jù)處理邏輯。為簡化接口，mapreduce要求所有存儲在底層分布式文件系統(tǒng)上的數(shù)據(jù)均要解釋成key/value的形式，并交給mapper/reducer中的map/reduce函數(shù)處理，產(chǎn)生另外一些key/value。mapper類圖如圖3-15所示，包括初始化、map操作和清理三部分。mapreduce編程模型28mapper與reducer解析（1）初始化）初始化mapper繼承了jobconfigurab

26、le接口。該接口中的configure方法允許通過jobconf參數(shù)對mapper進(jìn)行初始化。（2）map操作操作mapreduce框架會通過inputformat中recordreader從inputsplit獲取一個個key/value對，并交給下面的map()函數(shù)處理：void map(k1 key, v1 value, outputcollector output, reporter reporter) throws ioexception;該函數(shù)的參數(shù)除了key和value之外，還包括outputcollector和reporter兩個類型的參數(shù)，分別用于輸出結(jié)果和修改counter

27、值。（3）清理）清理mapper通過繼承closeable接口（它又繼承了java io中的closeable接口）獲得close方法，用戶可通過實(shí)現(xiàn)該方法對mapper進(jìn)行清理。mapreduce編程模型29mapper與reducer解析mapreduce提供了很多mapper/reducer實(shí)現(xiàn)，但大部分功能比較簡單，具體如圖3-16所示。它們對應(yīng)的功能分別是：chainmapper/chainreducer：用于支持鏈?zhǔn)阶鳂I(yè)。identitymapper/identityreducer：對于輸入key/value不進(jìn)行任何處理，直接輸出。invertmapper：交換key/value

28、位置。regexmapper：正則表達(dá)式字符串匹配。tokenmapper：將字符串分割成若干個token（單詞），可用作wordcount的mapper。longsumreducer：以key為組，對long類型的value求累加和。mapreduce編程模型30mapper與reducer解析新版api的mapper/reducer解析新api在舊api基礎(chǔ)上發(fā)生了以下幾個變化：mapper由接口變?yōu)槌橄箢悾也辉倮^承jobconfigurable和closeable兩個接口，而是直接在類中添加了setup和cleanup兩個方法進(jìn)行初始化和清理工作。將參數(shù)封裝到context對象中，這使

29、得接口具有良好的擴(kuò)展性。去掉maprunnable接口，在mapper中添加run方法，以方便用戶定制map()函數(shù)的調(diào)用方法，run默認(rèn)實(shí)現(xiàn)與舊版本中maprunner的run實(shí)現(xiàn)一樣。新api中reducer遍歷value的迭代器類型變?yōu)閖ava.lang.iterable，使得用戶可以采用“foreach”形式遍歷所有value，如下所示：mapreduce編程模型31mapper與reducer解析新版api的mapper/reducer解析mapreduce編程模型32partitioner接口的設(shè)計(jì)與實(shí)現(xiàn)partitioner的作用是對mapper產(chǎn)生的中間結(jié)果進(jìn)行分片，以便將同一

30、分組的數(shù)據(jù)交給同一個reducer處理，它直接影響reduce階段的負(fù)載均衡。舊版api中partitioner的類圖如圖3-20所示。它繼承了jobconfigurable，可通過configure方法初始化。它本身只包含一個待實(shí)現(xiàn)的方法getpartition。該方法包含三個參數(shù)，均由框架自動傳入，前面兩個參數(shù)是key/value，第三個參數(shù)numpartitions表示每個mapper的分片數(shù)，也就是reducer的個數(shù)。mapreduce編程模型33partitioner接口的設(shè)計(jì)與實(shí)現(xiàn)mapreduce提供了兩個partitioner實(shí)現(xiàn)：hashpartitioner和totalo

31、rderpartitioner。其中hashpartitioner是默認(rèn)實(shí)現(xiàn)，它實(shí)現(xiàn)了一種基于哈希值的分片方法，代碼如下：mapreduce編程模型34partitioner接口的設(shè)計(jì)與實(shí)現(xiàn)totalorderpartitioner提供了一種基于區(qū)間的分片方法，通常用在數(shù)據(jù)全排序中。在mapreduce環(huán)境中，容易想到的全排序方案是歸并排序，即在map階段，每個map task進(jìn)行局部排序；在reduce階段，啟動一個reduce task進(jìn)行全局排序。由于作業(yè)只能有一個reduce task，因而reduce階段會成為作業(yè)的瓶頸。為了提高全局排序的性能和擴(kuò)展性， mapreduce 提供了t

32、otalorderpartitioner。它能夠按照大小將數(shù)據(jù)分成若干個區(qū)間（分片），并保證后一個區(qū)間的所有數(shù)據(jù)均大于前一個區(qū)間數(shù)據(jù)，這使得全排序的步驟如下：步驟步驟1數(shù)據(jù)采樣。在client端通過采樣獲取分片的分割點(diǎn)。hadoop自帶了幾個采樣算法，如intercalsampler、randomsampler、splitsampler等（具體見org.apache.hadoop.mapred.lib包中的inputsampler類）。下面舉例說明。采樣數(shù)據(jù)為：b，abc，abd，bcd，abcd，efg，hii，afd，rrr，mnk經(jīng)排序后得到：abc，abcd，abd，afd，b，bcd，efg，hii，mnk，