大數(shù)據(jù)及MapReduce編程模型94

唐卓博ust_tz@126.com湖南大學信息科學與工程學院2014年8月大數(shù)據(jù)及其并行編程模型概述主要內容一、大數(shù)據(jù)概述二、應對大數(shù)據(jù)的系統(tǒng)思維三、MapReduce并行編程詳解2注：本課件前30頁PPT來源于國防科大李東升教授：“大數(shù)據(jù)時代的挑戰(zhàn)和探索”互聯(lián)網(wǎng)應用數(shù)據(jù)急劇增長

互聯(lián)網(wǎng)用戶數(shù)量巨大，日益活躍

?

微博、論壇、電子商務網(wǎng)站等

?

互聯(lián)網(wǎng)上的用戶生成數(shù)據(jù)（User

Generated

Content，

UGC）淘寶網(wǎng)每天新增數(shù)據(jù)40TB以上百度每天處理10PB量級的數(shù)據(jù)，總數(shù)據(jù)量達1000PB應用背景注：本課件前30頁PPT來源于國防科大李東升教授：“大數(shù)據(jù)時代的挑戰(zhàn)和探索”一、大數(shù)據(jù)概述?

隨著信息化的推進，國民經(jīng)濟、國家安全

等領域的數(shù)據(jù)不斷增長

物聯(lián)網(wǎng)、移動通信電話、手機短信、語音數(shù)據(jù)

遙感、公共安全、醫(yī)療、交通、情報等很多領域

?

高分辨率衛(wèi)星（影像）、城市監(jiān)控攝像頭（視頻）、…

?

據(jù)報道，武漢監(jiān)控攝像頭已超過25萬個，如采用1080P高清攝

像頭（一天產(chǎn)生數(shù)據(jù)量40GB以上），整個城市每天新增監(jiān)控

數(shù)據(jù)10PB以上應用背景?

科學實驗數(shù)據(jù)規(guī)模巨大，增長迅猛生物工程氣候監(jiān)測高能物理天文觀測生態(tài)環(huán)境

….氣候研究華大基因測序目前每天產(chǎn)生數(shù)據(jù)約15TB，一年超過5PB

一歐洲CERN對撞機每年產(chǎn)生的數(shù)據(jù)量超過15

PB基因測序應用背景全球數(shù)據(jù)量?IDC報告預測：未來

十年，全球數(shù)據(jù)量繼

續(xù)迅速增長Amount

of

digital

informationcreated

and

replicated

in

a

year––––年均增長率超過40%2009年0.8ZB2020年35ZB1ZB~106PB月球容量4GB的DVD光用容量4GB的DVD光盤存儲，DVD可從地球排至月球G-T-P-E-Z-Y?

維基（Wiki）百科的定義

Big

data

is

a

collection

of

data

sets

so

large

and

complex

that

it

becomes

difficult

to

process

using

on-hand

database

management

tools

?

IDC的定義

Big

data

technologies

describe

a

new

generation

of

technologies

and

architectures,

designed

to

economically

extract

value

from

very

large

volumes

of

a

wide

variety

of

data,

by

enabling

high-velocity

capture,

discovery,

and/or

analysis.

什么是大數(shù)據(jù)大數(shù)據(jù)是超大、復雜的數(shù)據(jù)集，現(xiàn)有的數(shù)據(jù)庫管理技術難以應對大數(shù)據(jù)技術描述了新一代的技術和架構，通過高速的數(shù)據(jù)獲取、發(fā)現(xiàn)和分析技術，以經(jīng)濟的方式從各種超大規(guī)模的數(shù)據(jù)中提取價值一、大數(shù)據(jù)概述?

Volume：規(guī)模大

從PB級到ZB級

1

ZB

~

106*

PB?

Variety：多樣化

結構化、非結構化

文本、圖像、視頻等?

Velocity：變化快

批處理/離線數(shù)據(jù)、流/實時/在線數(shù)據(jù)等?

Value/

Veracity：價值稀疏

/數(shù)據(jù)質量

噪音和無用信息很多一、大數(shù)據(jù)概述大數(shù)據(jù)的特點?

大數(shù)據(jù)技術對經(jīng)濟社會和科研都在產(chǎn)生重

要影響

–

互聯(lián)網(wǎng)產(chǎn)業(yè)、電子商務推薦、日常生活

大數(shù)據(jù)的影響季節(jié)性流感是一個重要的公共衛(wèi)生問題：WHO估計，全球每年25萬至50萬人因此死亡及時監(jiān)測到疾病的傳播情況，盡快采取應對措施2008年，Google通過處理網(wǎng)絡搜索日志中的幾千億查詢數(shù)據(jù)，訓練建立流感疾病監(jiān)測的數(shù)學模型，比美國病控制和預防中心提前1-2周給出流感的傳播情況論文發(fā)表在Nature(2009.2):DetectingInfluenza

EpidemicsusingSearchEngineQueryData?

大數(shù)據(jù)技術對經(jīng)濟社會和科研都在產(chǎn)生重

要影響

–

科學研究

三種科研模式：理論、實驗、計算第四模式：數(shù)據(jù)密集型的科學發(fā)現(xiàn)圖靈獎獲得者JimGray2007年提出專輯：Nature(2008.9):”Big

Data”,Science(2011.2):”Dealing

with

data”大數(shù)據(jù)的影響?

2012年3月29日，美國政府宣布投資2億

美元啟動“大數(shù)據(jù)研發(fā)計劃”（

Big

Data

R&D

Initiative

）

美NSF、國防部、能源部、衛(wèi)生總署等七部委?

我國科技部和基金委等部門高度重視

2013年973新立項項目：2項

“十二五”

國家科技計劃信息技術領域2013年度備選項

目征集指南?

國內外學術界的熱點課題

SIGMOD、

VLDB、OSDI、NSDI等著名會議

Nature、Science雜志11大數(shù)據(jù)成為熱點課題?

傳統(tǒng)技術難以應對大數(shù)據(jù)的規(guī)模

–

數(shù)據(jù)存儲及訪問的挑戰(zhàn)當前較快硬盤的傳輸速度6Gbps，線性掃描10PB數(shù)據(jù)，需約19天而百度、Google等互聯(lián)網(wǎng)公司每天處理

的數(shù)據(jù)量超過10PB案例源于：北航/愛丁堡樊文飛教授

?

可擴展是大規(guī)模分布式系統(tǒng)面臨的基礎性問題

–Jim

Gray（圖靈獎獲得者）將可擴展問題列為信

息技術領域需解決的16個長遠問題之首Jim

Gray.

What

Next?

A

Few

Remaining

Problems

in

Information

Technology.

ACM

Turing

Award

Lecture

(1999).

Available

at

http:///enus/um/people/gray/talks/Gray_Turing_FCRC.ppt大數(shù)據(jù)帶來的挑戰(zhàn)（1）?

很多大數(shù)據(jù)應用對響應時間要求高

–

規(guī)模大、響應快：對存儲和處理提出了很大挑戰(zhàn)

–2007年前,Facebook使用數(shù)據(jù)庫，總數(shù)據(jù)量15TB

?

目前，F(xiàn)acebook每天新增加的數(shù)據(jù)約70TB

–

傳統(tǒng)并行數(shù)據(jù)庫擴展性受限，節(jié)點規(guī)模很少超過100，

且價格昂貴

?2011年，F(xiàn)acebook系統(tǒng)具有2700多個節(jié)點，Google單個數(shù)據(jù)中心在上

萬個節(jié)點集群上存儲了約10PB數(shù)據(jù)?

如何設計可擴展、低成本、快速響應的大

數(shù)據(jù)存儲和處理系統(tǒng)？大數(shù)據(jù)存儲與處理的可擴展難題數(shù)據(jù)種類多，需求多樣，關聯(lián)復雜

–文本、圖像、圖形、視頻等

–在線/流數(shù)據(jù)、離線/批處理等如何建模、存儲、查詢、分析和理解多樣

化的復雜數(shù)據(jù)，挖掘數(shù)據(jù)價值？

–

大數(shù)據(jù)中垃圾和珍寶并存

–大海撈針、去粗取精、去偽存真

–需要計算機專家和領域專家的配合….大數(shù)據(jù)面臨的挑戰(zhàn)（2）傳統(tǒng)算法在大數(shù)據(jù)時代可能不再有效

多項式時間算法O(Nk)，N太大

需要計算復雜性和算法設計理論上的變革

需要大數(shù)據(jù)計算思維上的變化

例如，從確定性計算到非精確性計算

商品在線推薦：只需要計算出前10名相關的結果，有

一點不準確也沒有關系傳統(tǒng)算法結論在大數(shù)據(jù)時代需要重新評估

–

簡單方法+大數(shù)據(jù)集可能取得很好的結果大數(shù)據(jù)面臨的挑戰(zhàn)（3）?

2007年,Google公司的Brants等人研究了機

器翻譯領域中基于單詞訓練數(shù)據(jù)集的語言

模型

比較了當時最先進的KN算法

與其提出的一個簡單算法SB

研究表明，簡單算法在小數(shù)

據(jù)集時效果不佳，但在大數(shù)

據(jù)集時，簡單算法卻產(chǎn)生了

更好的效果

T.Brants,A.C.Popat,etal.LargeLanguageModelsinMachineTranslation.

ProceedingsoftheJointConferenceonEmpiricalMethodsinNatural

LanguageProcessingandComputationalNaturalLanguageLearning,2007.16傳統(tǒng)算法結論需要重新評估?

大數(shù)據(jù)時代的算法新理論

–

新的計算復雜性和算法設計理論?

復雜大數(shù)據(jù)的建模、表示和可視化

–

多源異構大數(shù)據(jù)：由大到小?

面向大數(shù)據(jù)的新型存儲和計算系統(tǒng)架構

–大規(guī)模并行/分布處理?

大數(shù)據(jù)（并行）挖掘算法及應用大數(shù)據(jù)的研究課題主要內容一、大數(shù)據(jù)概述二、應對大數(shù)據(jù)的系統(tǒng)思維三、MapReduce并行編程詳解2181.

數(shù)據(jù)為中心的計算架構計算和存儲唇齒相依2.化繁為簡，分而治之

可擴展的數(shù)據(jù)并行處理3.求同存異，聚焦領域放松傳統(tǒng)數(shù)據(jù)處理技術的約束，如一致性等、行式存儲-列式存儲高可擴展高吞吐率高可靠性……主要內容18二、應對大數(shù)據(jù)的系統(tǒng)思維1.

數(shù)據(jù)為中心的計算架構過去20年來,計算器件的帶寬提升了100–2000倍，而延遲改善只有5-20倍CPU

和

on-chip

L2之間:

帶寬:增長了2250倍

延遲:降低了20倍L3

cache

和DRAM之間：

帶寬:增長了125倍

延遲:降低了4倍DRAM

和disk之間:

帶寬:增長了150倍

延遲:降低了8倍

LAN連接的兩個節(jié)點之間

:

帶寬:增長了100倍

延遲:降低了15倍充分利用數(shù)據(jù)和存儲的局部性（緩存、復制、預?。┭舆t提升滯后于帶寬Source：CACM（Patterson)二、應對大數(shù)據(jù)的系統(tǒng)思維1.

數(shù)據(jù)為中心的計算架構（續(xù)）20二、應對大數(shù)據(jù)的計算思維

–

數(shù)據(jù)分布存儲在計算附近？–

計算盡量利用數(shù)據(jù)局部性–

存儲架構、互連網(wǎng)絡架構數(shù)據(jù)密集型計算計算密集型計算

SystemData–

數(shù)據(jù)存儲與計算相分離–

計算之前加載數(shù)據(jù)–

規(guī)模挑戰(zhàn)：元數(shù)據(jù)管理+數(shù)

據(jù)傳輸221.

數(shù)據(jù)為中心的計算架構（續(xù)）案例：MicrosoftFlatDatacenterStorage(OSDI2012)MinuteSort新架構+高效互連網(wǎng)絡?

簡化的可擴展數(shù)據(jù)并行處理：MapReduce框架Map:

Key1/Value1

（輸入數(shù)據(jù)）Reduce:

Key2/Value2

（中間數(shù)據(jù)）Key2/Value2

（中間數(shù)據(jù)）

Value

（輸出數(shù)據(jù)）數(shù)據(jù)按照key進行分區(qū)：數(shù)據(jù)并行Google提出（OSDI’04）

中間數(shù)據(jù)輸出數(shù)據(jù)輸入數(shù)據(jù)222.

化繁為簡，分而治之?

特點

每個Map/Reduce任務相對獨立，執(zhí)行的任務簡單

簡單，易于擴展（應用無需修改）、容錯性好（復算）

缺點：Map和Reduce階段之間需要大量的數(shù)據(jù)交換?

開源實現(xiàn)

Hadoop及其變型

成功應用于眾多著名公司

?

Facebook,

Yahoo!

,

AOL,

EBay,

IBM,

….

?

百度，阿里巴巴等MapReduce數(shù)據(jù)并行框架?

MapReduce

革新

MapReduce

Online

(UC

Berkeley)、

HadoopDB

(

(Yale)

)

Hadoop++

(Dittrich

et

al.:

VLDB’2012)

Spark（內存Hadoop，

UCBerkeley）…?

新的數(shù)據(jù)并行處理框架

Pregel，GraphLab:

圖數(shù)據(jù)的并行處理框架

Dremel：

快速交互式數(shù)據(jù)分析系統(tǒng)，PB/s

Storm：流處理數(shù)據(jù)框架

….學術界和工業(yè)界不懈努力數(shù)據(jù)一致性

關系數(shù)據(jù)庫：強一致性Atomicity

Consistency

Isolation

Durability（

ACID）寫操作完成后，任何后續(xù)讀操作將得到最新值?

分布式環(huán)境下，強一致性的代價昂貴，很

多應用也無需強一致性弱化數(shù)據(jù)一致性，提升可擴展性和可靠性3.

求同存異，聚焦領域Youcanhaveatmosttwoofthesepropertiesforanyshared-datasystem在分布式系統(tǒng)中，數(shù)據(jù)一致性、系統(tǒng)可用性、以及對網(wǎng)絡斷分容忍性中，任何時候只能實現(xiàn)其中兩個特性UC

Berkeley的Eric

Brewer提出猜想（2000）MIT的Nancy

Lynch等予以證明（2002）

CAP定理

為什么犧牲數(shù)據(jù)一致性？

犧牲P、A對互聯(lián)網(wǎng)上的大數(shù)據(jù)

應用來說難以容忍犧牲C的代價可以接受應用開發(fā)稍顯復雜很多應用并不關心C弱（最終）一致性Basically

Available

Soft-stateEventual

Consistency（Base）弱化數(shù)據(jù)一致性?

聚焦領域應用需求，簡（優(yōu)）化系統(tǒng)設計

GFS：聚焦于數(shù)據(jù)“讀多寫少”場景

滿足可擴展性、可用性等多種需求的平衡?

NoSQL

存儲

很多領域應用只需要對數(shù)據(jù)進行簡單的讀寫

?

不需要復雜的SQL操作，如skyline查詢、多表join等key/value存儲放棄SQL的某些要求列式存儲283.

求同存異，聚焦領域

全能選手

Vs.

特長生

大數(shù)據(jù)研究正方興未艾?

Gartner：Hype

Cycle

2012主要內容一、大數(shù)據(jù)概述二、應對大數(shù)據(jù)的系統(tǒng)思維三、MapReduce并行編程詳解2MapReduce起源：Google搜索每一次搜索200+CPU200TB以上數(shù)據(jù)1010CPU周期0.1秒內響應5￠廣告收入計算問題簡單，但求解困難待處理數(shù)據(jù)量巨大（PB級），只有分布在成百上千個節(jié)點上并行計算才能在可接受的時間內完成如何進行并行分布式計算？如何分發(fā)待處理數(shù)據(jù)？如何處理分布式計算中的錯誤？簡單的問題，計算并不簡單！MapReduce：大規(guī)模數(shù)據(jù)處理處理海量數(shù)據(jù)（>1TB）上百/上千CPU實現(xiàn)并行處理簡單地實現(xiàn)以上目的"GoogleEarthuses70.5TB:70TBfortherawimageryand500GBfortheindexdata."From:/2006/09/how-much-data-does-google-store.html分而治之DivideandConquer

GoogleMapReduce架構設計師JeffreyDeanMapReduce特性自動實現(xiàn)分布式并行計算容錯提供狀態(tài)監(jiān)控工具模型抽象簡潔，程序員易用MapReduce特性MapReduce程序是設計用來并行計算大規(guī)模海量數(shù)據(jù)的，這需要把工作流分劃到大量的機器上去，如果組件(component)之間可以任意的共享數(shù)據(jù)，那這個模型就無法擴展到大規(guī)模集群上去（數(shù)百或數(shù)千個節(jié)點），用來保持節(jié)點間數(shù)據(jù)的同步而產(chǎn)生的通信開銷會使得系統(tǒng)在大規(guī)模集群上變得不可靠和效率低下所有在MapReduce上的數(shù)據(jù)元素都是不可變的，這就意味著它們不能夠被更新。如果在一個mapping任務中你改變了一個輸入鍵值對，它并不會反饋到輸入文件；節(jié)點間的通信只在產(chǎn)生新的輸出鍵值對((key,value)pairs)時發(fā)生，Hadoop系統(tǒng)會把這些輸出傳到下一個執(zhí)行階段。MapReducemapping和reducing函數(shù)接收數(shù)值（鍵，值）對mapper可能把一個輸入map為0個，1個或100個輸出reducer可能計算超過一個的輸入列表并生成一個或多個不同的輸出MapReduce編程模型用戶只需要實現(xiàn)兩個函數(shù)接口：map(in_key,in_value)-> (out_key,intermediate_valuelist)reduce(out_key,intermediate_valuelist)->out_valuelist輸入的key和value的類型和輸出的類型可以是不同的map將數(shù)據(jù)源中的記錄（文本中的行、數(shù)據(jù)庫中條目等）作為map函數(shù)中的key*value對例如（filename,line）map()將生成一個或多個中間結果，以及與input相對應的一個outputkeyreducemap操作結束后，所有與某指定outkey相對應的中間結果組合為一個列表（list）。reduce()函數(shù)將這些中間結果組合為一個或多個對應于同一outputkey的finalvalue每一個outputkey通常只有一個finalvaluereduce()個數(shù)可以為0個或多個任務執(zhí)行過程源文件：GFSMap處理結果：本地存儲Reduce處理結果：GFS日志：GFS文件存儲位置Shuffle和Sort當Map開始產(chǎn)生輸出時，并不是簡單的把數(shù)據(jù)寫到磁盤，因為頻繁的磁盤操作會導致性能嚴重下降。它的處理過程更復雜，數(shù)據(jù)首先是寫到內存中的一個緩沖區(qū)，并進行預排序，以提升效率。CombinerCombiner并行化map()函數(shù)可以并行執(zhí)行，為不同的輸入數(shù)據(jù)集生成不同的中間結果reduce()函數(shù)也可以并行執(zhí)行，分別處理不同的outputkeymap和reduce的處理過程中不發(fā)生通信瓶頸：只有當map處理全部結束后，reduce過程才能夠開始MapReduce的并行執(zhí)行Worker故障Master周期性的ping每個worker。如果master在一個確定的時間段內沒有收到worker返回的信息，那么它將把這個worker標記成失效重新執(zhí)行該節(jié)點上已經(jīng)執(zhí)行或尚未執(zhí)行的Map任務重新執(zhí)行該節(jié)點上未完成的Reduce任務，已完成的不再執(zhí)行Master故障定期寫入檢查點數(shù)據(jù)從檢查點恢復MapReduce的容錯任務備份機制慢的workers會嚴重地拖延整個執(zhí)行完成的時間由于其他的任務占用了資源磁盤損壞解決方案：推測性的執(zhí)行（Speculativeexecution）在即將完成時，備份任務多個worker同時進行相同的任務任何一個完成均可可以十分顯著地提高執(zhí)行效率MapReduce的優(yōu)化本地處理Master調度策略:向GFS詢問獲得輸入文件blocks副本的位置信息Maptasks的輸入數(shù)據(jù)通常按64MB來劃分(GFSblock大小)按照blocks所在的機器或機器所在機架的范圍進行調度效果絕大部分機器從本地讀取文件作為輸入，節(jié)省大量帶寬MapReduce的優(yōu)化跳過有問題的記錄一些特定的輸入數(shù)據(jù)常導致Map/Reduce無法運行調試或者修改在每個worker里運行一個信號處理程序，捕獲map或reduce任務崩潰時發(fā)出的信號，一旦捕獲，就會向master報告，同時報告輸入記錄的編號信息。如果master看到一條記錄有兩次崩潰信息，那么就會對該記錄進行標記，下次運行的時候，跳過該記錄MapReduce的優(yōu)化MapReduce示例：單詞計數(shù)案例：單詞記數(shù)問題(WordCount)給定一個巨大的文本（如1TB），如何計算單詞出現(xiàn)的數(shù)目？MapReduce示例：單詞計數(shù)使用MapReduce求解該問題Step1:自動對文本進行分割MapReduce示例：單詞計數(shù)使用MapReduce求解該問題Step2:在分割之后的每一對<key,value>進行用戶定義的Map進行處理，再生成新的<key,value>對MapReduce示例：單詞計數(shù)使用MapReduce求解該問題Step3:對輸出的結果集歸攏(不同mapslot間copy到一起)、排序(sort)(系統(tǒng)自動完成）MapReduce示例：單詞計數(shù)使用MapReduce求解該問題Step4:通過Reduce操作生成最后結果MapReduce示例：單詞計數(shù)使用MapReduce求解該問題定義Map和Reduce函數(shù)map(Stringinput_key,Stringinput_value)://input_key:documentname//input_value:documentcontents

foreachwordwininput_value:

EmitIntermediate(w,"1");reduce(Stringoutput_key,Iteratorintermediate_values)://output_key:aword//output_values:alistofcounts

intresult=0;

foreachvinintermediate_values:result+=ParseInt(v);

Emit(AsString(result));其他示例分布式檢索map函數(shù)挑選出滿足特定模式的行，并將其組裝成元組輸出。reduce函數(shù)是一個簡單的確認函數(shù)，它完成的工作僅僅是將中間元組拷貝到輸出中。計算URL訪問頻率map函數(shù)處理web網(wǎng)頁的訪問日志，并輸出<URL,1>。reduce函數(shù)將每個URL的訪問次數(shù)加起來，輸出<URL,totalcount>其他示例翻轉web-link圖在每個作為源的頁面中，檢查其連接URL，并逐個輸出<target,source>元組。reduce函數(shù)將連接到每個target的所有source組合起來，形成list列表，輸出<target,list(source)>每個站點的術語向量術語向量表示出在一篇文章中或者一組文章中最重要的單詞，通常以<word,frequency>元組的方式。map函數(shù)輸出每個文章的<hostname,termvector>（hostname通過文章的URL分析得到）。reduce函數(shù)取出不常用的術語，將其余的相加，得到最終的<hostname,termvector>對其他示例倒排索引map函數(shù)分析每個文檔,然后產(chǎn)生一個(詞,文檔號)對的序列.reduce函數(shù)接受一個給定詞的所有對,排序相應的文檔IDs,并且產(chǎn)生一個(詞,文檔ID列表)對.所有的輸出對集形成一個簡單的倒排索引分布式排序map函數(shù)從每個記錄提取key,并且產(chǎn)生一個(key,record)對.reduce函數(shù)不改變任何的對.“實踐是檢驗真理的唯一標準”實踐證明，MapReduce是出色的分布式計算模型Google宣布，其對分布于1000臺計算機上的1TB數(shù)據(jù)進行排序僅僅需要68s對4000臺計算機上的1PB數(shù)據(jù)進行排序處理僅需要6小時2分鐘（每次測試至少會損壞1塊硬盤）在08年1月份，GoogleMapReduce平均每天的數(shù)據(jù)處理量是20PB，相當于美國國會圖書館當年5月份存檔網(wǎng)絡數(shù)據(jù)的240倍Hadoop上的MapReducejob:是客戶端程序想要完成的一系列工作的集合。包括輸入數(shù)據(jù)，MapReduce程序和配置信息。task:Hadoop將job分解為tasks有兩種類型的task:maptask和reducetaskjobtracker和tasktracker:用來控制job執(zhí)行的tasktracker運行task，并向jobtracker報告進度信息jobtracker記錄下每一個job的進度信息，如果一個task失敗，jobtracker會將其重新調度到另外的tasktracker上。Hadoop-MapReduce工作原理HadoopStreamingandPipesHadoop流允許用Java以外的語言來編寫Map和Reduce函數(shù)Hadoop管道C++接口流和管道及子進程的關系進度和狀態(tài)更新Hadoop-MapReduceMapperpublicstaticclass**MapperextendsMapper<Object,Text,Text,IntWritable>Reducer

publicstaticclass**Reducerextendseducer<Text,IntWritable,Text,IntWritable>DriverPackage

org.apache.hadoop.mapreduce.Job; org.apache.hadoop.mapreduce.Mapper; org.apache.hadoop.mapreduce.Reducer;接口描述publicinterfaceMapper<K1,V1,K2,V2>extendsJobConfigurable,Closeable{voidmap(K1key,V1value,OutputCollector<K2,V2>output,Reporterreporter)throwsIOException;}publicinterfaceReducer<K2,V2,K3,V3>extendsJobConfigurable,Closeable{voidreduce(K2key,Iterator<V2>values,OutputCollector<K3,V3>output,Reporterreporter)throwsIOException;}MapperpublicstaticclassMapextendsMapper<LongWritable,Text,Text,IntWritable>{privatefinalstaticIntWritableone=newIntWritable(1);privateTextword=newText();publicvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{Stringline=value.toString();StringTokenizertokenizer=newStringTokenizer(line);while(tokenizer.hasMoreTokens()){word.set(tokenizer.nextToken());context.write(word,one);}}}ReducerpublicstaticclassReduceextendsReducer<Text,IntWritable,Text,IntWritable>{publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intsum=0;for(IntWritableval:values){ sum+=val.get();}context.write(key,newIntWritable(sum));}}Driverpublicstaticvoidmain(String[]args)throwsException{Configurationconf=newConfiguration();Jobjob=newJob(conf,"wordcount");job.setJarByClass(WordCount.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);job.setMapperClass(Map.class);job.setReducerClass(Reduce.class);job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(TextOutputFormat.class);FileInputFormat.addInputPath(job,newPath(args[0]));FileOutputFormat.setOutputPath(job,newPath(args[1]));job.waitForCompletion(true);}InputFiles輸入文件一般保存在HDFS中文件的類型不固定，可能是文本的，也有可能是其它形式的文件文件經(jīng)常很大，甚至有幾十個GBInput會被分成inputsplit，split由record組成。map處理每一個record，并且返回key和value的對MapReduce程序并不需要直接處理InputSplit，由InputFormat創(chuàng)建的InputSplitsInputSplit定義了輸入到單個Map任務的輸入數(shù)據(jù)InputSplit將文件分為64MB的大小hadoop-site.xml中的mapred.min.split.size參數(shù)控制這個大小mapred.tasktracker.map.taks.maximum用來控制某一個節(jié)點上所有map任務的最大數(shù)目RecordReaderInputSplit定義了一項工作的大小，但是沒有定義如何讀取數(shù)據(jù)RecordReader實際上定義了如何從數(shù)據(jù)上轉化為一個(key,value)對，從而輸出到Mapper類中TextInputFormat提供了LineRecordReaderInputFormat定義了這些文件如何分割，讀取InputFile提供了以下一些功能選擇文件或者其它對象，用來作為輸入定義InputSplits，將一個文件分開成為任務為RecordReader提供一個工廠，用來讀取這個文件有一個抽象的類FileInputFormat，所有的輸入格式類都從這個類繼承這個類的功能以及特性。當啟動一個Hadoop任務的時候，一個輸入文件所在的目錄被輸入到FileInputFormat對象中。FileInputFormat從這個目錄中讀取所有文件。然后FileInputFormat將這些文件分割為一個或者多個InputSplits。通過在JobConf對象上設置JobConf.setInputFormat設置文件輸入的格式預定義的文件輸入格式InputFormat:Description:Key:Value:TextInputFormatDefaultformat;readslinesoftextfilesThebyteoffsetofthelineThelinecontentsKeyValueInputFormatParseslinesintokey,valpairsEverythinguptothefirsttabcharacterTheremainderofthelineSequenceFileInputFormatAHadoop-specifichigh-performancebinaryformatuser-defineduser-defined各種InputFormatTextInputFormat，默認的格式，每一行是一個單獨的記錄，并且作為value，文件的偏移值作為keyKeyValueInputFormat，這個格式每一行也是一個單獨的記錄，但是Key和Value用Tab隔開，是默認的OutputFormat，可以作為中間結果，作為下一步MapReduce的輸入。SequenceFileInputFormat基于塊進行壓縮的格式對于幾種類型數(shù)據(jù)的序列化和反序列化操作用來將數(shù)據(jù)快速讀取到Mapper類中Writable接口Hadoop使用Writable做序列化定義了兩個方法二進制寫入DataOutput流二進制讀取DataInput流Hadoop自帶一系列Writable實現(xiàn)，可以滿足絕大多數(shù)需要可以自定義Writable，控制二進制表示和排序實現(xiàn)Writable接口的例子public

class

MyWritable

implements

Writable

{

//

Some

data

private

int

counter;

private

long

timestamp;

public

void

write(DataOutput

out)

throws

IOException

{

out.writeInt(counter);

out.writeLong(timestamp);

}

public

void

readFields(DataInput

in)

throws

IOException

{

counter

=

in.readInt();

timestamp

=

in.readLong();

}

public

static

MyWritable

read(DataInput

in)

throws

IOException

{

MyWritable

w

=

new

MyWritable();

w.readFields(in);

return

w;

}

}

Writable的Java基本封裝Mapper每一個Mapper類的實例生成了一個Java進程（在某一個InputSplit上執(zhí)行）有兩個額外的參數(shù)OutputCollector以及Reporter，前者用來收集中間結果，后者用來獲得環(huán)境參數(shù)以及設置當前執(zhí)行的狀態(tài)。現(xiàn)在用Mapper.Context提供給每一個Mapper函數(shù)，用來提供上面兩個對象的功能數(shù)據(jù)壓縮Partition&Shuffle在Map工作完成之后，每一個Map函數(shù)會將結果傳到對應的Reducer所在的節(jié)點，此時，用戶可以提供一個Partitioner類，用來決定一個給定的(key,value)對傳輸?shù)木唧w位置Combinerconf.setCombinerClass(Reduce.class);是在本地執(zhí)行的一個Reducer，滿足一定的條件才能夠執(zhí)行。Sort傳輸?shù)矫恳粋€節(jié)點上的所有的Reduce函數(shù)接收到得Key,value對會被Hadoop自動排序（即Map生成的結果傳送到某一個節(jié)點的時候，會被自動排序）Reduce做用戶定義的Reduce操作接收到一個OutputCollector的類作為輸出OutputFormat寫入到HDFS的所有OutputFormat都繼承自FileOutputFormat每一個Reducer都寫一個文件到一個共同的輸出目錄，文件名是part-nnnnn，其中nnnnn是與每一個reducer相關的一個號（partitionid）JobConf.setOutputFormat()RecordWriter用來指導如何輸出一個記錄到文件中OutputFormatOutputFormat:DescriptionTextOutputFormatDefault;writeslinesin"key\tvalue"formSequenceFileOutputFo