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

互聯網用戶數量巨大，日益活躍

?

微博、論壇、電子商務網站等

?

互聯網上的用戶生成數據（User

Generated

Content，

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

隨著信息化的推進，國民經濟、國家安全

等領域的數據不斷增長

物聯網、移動通信電話、手機短信、語音數據

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

?

高分辨率衛星（影像）、城市監控攝像頭（視頻）、…

?

據報道，武漢監控攝像頭已超過25萬個，如采用1080P高清攝

像頭（一天產生數據量40GB以上），整個城市每天新增監控

數據10PB以上應用背景一、大數據概述?隨著信息化的推進，國民經濟、國家安全應用?

科學實驗數據規模巨大，增長迅猛生物工程氣候監測高能物理天文觀測生態環境

….氣候研究華大基因測序目前每天產生數據約15TB，一年超過5PB

一歐洲CERN對撞機每年產生的數據量超過15

PB基因測序應用背景氣候研究華大基因測序目歐洲CERN對撞基因測序應用背景全球數據量?IDC報告預測：未來

十年，全球數據量繼

續迅速增長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全球數據量?IDC報告預測：未來Amounto?

維基（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.

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

Volume：規模大

從PB級到ZB級

1

ZB

~

106*

PB?

Variety：多樣化

結構化、非結構化

文本、圖像、視頻等?

Velocity：變化快

批處理/離線數據、流/實時/在線數據等?

Value/

Veracity：價值稀疏

/數據質量

噪音和無用信息很多一、大數據概述大數據的特點一、大數據概述?Volume：規模大一、大數據概述大數?

大數據技術對經濟社會和科研都在產生重

要影響

–

互聯網產業、電子商務推薦、日常生活

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

EpidemicsusingSearchEngineQueryData?大數據技術對經濟社會和科研都在產生重大數據的影響季節性?

大數據技術對經濟社會和科研都在產生重

要影響

–

科學研究

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

Data”,Science(2011.2):”Dealing

with

data”大數據的影響?大數據技術對經濟社會和科研都在產生重三種科研模式：理論?

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

美元啟動“大數據研發計劃”（

Big

Data

R&D

Initiative

）

美NSF、國防部、能源部、衛生總署等七部委?

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

2013年973新立項項目：2項

“十二五”

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

目征集指南?

國內外學術界的熱點課題

SIGMOD、

VLDB、OSDI、NSDI等著名會議

Nature、Science雜志11大數據成為熱點課題?2012年3月29日，美國政府宣布投資2億11大數據?

傳統技術難以應對大數據的規模

–

數據存儲及訪問的挑戰當前較快硬盤的傳輸速度6Gbps，線性掃描10PB數據，需約19天而百度、Google等互聯網公司每天處理

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

?

可擴展是大規模分布式系統面臨的基礎性問題

–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大數據帶來的挑戰（1）?傳統技術難以應對大數據的規模當前較快硬盤的傳輸速度6?

很多大數據應用對響應時間要求高

–

規模大、響應快：對存儲和處理提出了很大挑戰

–2007年前,Facebook使用數據庫，總數據量15TB

?

目前，Facebook每天新增加的數據約70TB

–

傳統并行數據庫擴展性受限，節點規模很少超過100，

且價格昂貴

?2011年，Facebook系統具有2700多個節點，Google單個數據中心在上

萬個節點集群上存儲了約10PB數據?

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

數據存儲和處理系統？大數據存儲與處理的可擴展難題大數據存儲與處理的可擴展難題數據種類多，需求多樣，關聯復雜

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

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

化的復雜數據，挖掘數據價值？

–

大數據中垃圾和珍寶并存

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

–需要計算機專家和領域專家的配合….大數據面臨的挑戰（2）數據種類多，需求多樣，關聯復雜大數據面臨的挑戰（2）傳統算法在大數據時代可能不再有效

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

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

需要大數據計算思維上的變化

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

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

一點不準確也沒有關系傳統算法結論在大數據時代需要重新評估

–

簡單方法+大數據集可能取得很好的結果大數據面臨的挑戰（3）傳統算法在大數據時代可能不再有效大數據面臨的挑戰（3）?

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

器翻譯領域中基于單詞訓練數據集的語言

模型

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

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

研究表明，簡單算法在小數

據集時效果不佳，但在大數

據集時，簡單算法卻產生了

更好的效果

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

ProceedingsoftheJointConferenceonEmpiricalMethodsinNatural

LanguageProcessingandComputationalNaturalLanguageLearning,2007.16傳統算法結論需要重新評估?2007年,Google公司的Brants等人研究?

大數據時代的算法新理論

–

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

復雜大數據的建模、表示和可視化

–

多源異構大數據：由大到小?

面向大數據的新型存儲和計算系統架構

–大規模并行/分布處理?

大數據（并行）挖掘算法及應用大數據的研究課題?大數據時代的算法新理論大數據的研究課題主要內容一、大數據概述二、應對大數據的系統思維三、MapReduce并行編程詳解2主要內容一、大數據概述2181.

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

可擴展的數據并行處理3.求同存異，聚焦領域放松傳統數據處理技術的約束，如一致性等、行式存儲-列式存儲高可擴展高吞吐率高可靠性……主要內容18二、應對大數據的系統思維181.數據為中心的計算架構高可擴展主要內容18二、應對大1.

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

和

on-chip

L2之間:

帶寬:增長了2250倍

延遲:降低了20倍L3

cache

和DRAM之間：

帶寬:增長了125倍

延遲:降低了4倍DRAM

和disk之間:

帶寬:增長了150倍

延遲:降低了8倍

LAN連接的兩個節點之間

:

帶寬:增長了100倍

延遲:降低了15倍充分利用數據和存儲的局部性（緩存、復制、預取）延遲提升滯后于帶寬Source：CACM（Patterson)1.數據為中心的計算架構過去20年來,計算器件的帶寬充分二、應對大數據的系統思維1.

數據為中心的計算架構（續）20二、應對大數據的計算思維

–

數據分布存儲在計算附近？–

計算盡量利用數據局部性–

存儲架構、互連網絡架構數據密集型計算計算密集型計算

SystemData–

數據存儲與計算相分離–

計算之前加載數據–

規模挑戰：元數據管理+數

據傳輸二、應對大數據的系統思維1.數據為中心的計算架構（續）20221.

數據為中心的計算架構（續）案例：MicrosoftFlatDatacenterStorage(OSDI2012)MinuteSort新架構+高效互連網絡221.數據為中心的計算架構（續）案例：Microsoft?

簡化的可擴展數據并行處理：MapReduce框架Map:

Key1/Value1

（輸入數據）Reduce:

Key2/Value2

（中間數據）Key2/Value2

（中間數據）

Value

（輸出數據）數據按照key進行分區：數據并行Google提出（OSDI’04）

中間數據輸出數據輸入數據222.

化繁為簡，分而治之Map:Key1/Value1（輸入數據）Key2?

特點

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

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

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

開源實現

Hadoop及其變型

成功應用于眾多著名公司

?

Facebook,

Yahoo!

,

AOL,

EBay,

IBM,

….

?

百度，阿里巴巴等MapReduce數據并行框架?特點MapReduce數據并行框架?

MapReduce

革新

MapReduce

Online

(UC

Berkeley)、

HadoopDB

(

(Yale)

)

Hadoop++

(Dittrich

et

al.:

VLDB’2012)

Spark（內存Hadoop，

UCBerkeley）…?

新的數據并行處理框架

Pregel，GraphLab:

圖數據的并行處理框架

Dremel：

快速交互式數據分析系統，PB/s

Storm：流處理數據框架

….學術界和工業界不懈努力?MapReduce革新學術界和工業界不懈努力數據一致性

關系數據庫：強一致性Atomicity

Consistency

Isolation

Durability（

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

分布式環境下，強一致性的代價昂貴，很

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

求同存異，聚焦領域數據一致性3.求同存異，聚焦領域Youcanhaveatmosttwoofthesepropertiesforanyshared-datasystem在分布式系統中，數據一致性、系統可用性、以及對網絡斷分容忍性中，任何時候只能實現其中兩個特性UC

Berkeley的Eric

Brewer提出猜想（2000）MIT的Nancy

Lynch等予以證明（2002）

CAP定理Youcanhaveatmosttwoof在分布

為什么犧牲數據一致性？

犧牲P、A對互聯網上的大數據

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

Available

Soft-stateEventual

Consistency（Base）弱化數據一致性為什么犧牲數據一致性？犧牲C的代價可以接受弱化數據一致性?

聚焦領域應用需求，簡（優）化系統設計

GFS：聚焦于數據“讀多寫少”場景

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

NoSQL

存儲

很多領域應用只需要對數據進行簡單的讀寫

?

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

求同存異，聚焦領域

全能選手

Vs.

特長生?聚焦領域應用需求，簡（優）化系統設計key/value

大數據研究正方興未艾?

Gartner：Hype

Cycle

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

GoogleMapReduce架構設計師JeffreyDeanMapReduce：大規模數據處理處理海量數據（>1TB）"MapReduce特性自動實現分布式并行計算容錯提供狀態監控工具模型抽象簡潔，程序員易用MapReduce特性自動實現分布式并行計算MapReduce特性MapReduce程序是設計用來并行計算大規模海量數據的，這需要把工作流分劃到大量的機器上去，如果組件(component)之間可以任意的共享數據，那這個模型就無法擴展到大規模集群上去（數百或數千個節點），用來保持節點間數據的同步而產生的通信開銷會使得系統在大規模集群上變得不可靠和效率低下所有在MapReduce上的數據元素都是不可變的，這就意味著它們不能夠被更新。如果在一個mapping任務中你改變了一個輸入鍵值對，它并不會反饋到輸入文件；節點間的通信只在產生新的輸出鍵值對((key,value)pairs)時發生，Hadoop系統會把這些輸出傳到下一個執行階段。MapReduce特性MapReduce程序是設計用來并行計MapReducemapping和reducing函數接收數值（鍵，值）對mapper可能把一個輸入map為0個，1個或100個輸出reducer可能計算超過一個的輸入列表并生成一個或多個不同的輸出MapReducemapping和reducing函數接收數MapReduce編程模型用戶只需要實現兩個函數接口：map(in_key,in_value)-> (out_key,intermediate_valuelist)reduce(out_key,intermediate_valuelist)->out_valuelist輸入的key和value的類型和輸出的類型可以是不同的MapReduce編程模型用戶只需要實現兩個函數接口：map將數據源中的記錄（文本中的行、數據庫中條目等）作為map函數中的key*value對例如（filename,line）map()將生成一個或多個中間結果，以及與input相對應的一個outputkeymap將數據源中的記錄（文本中的行、數據庫中條目等）作為mareducemap操作結束后，所有與某指定outkey相對應的中間結果組合為一個列表（list）。reduce()函數將這些中間結果組合為一個或多個對應于同一outputkey的finalvalue每一個outputkey通常只有一個finalvaluereduce()個數可以為0個或多個reducemap操作結束后，所有與某指定outkey相對大數據及MapReduce編程模型課件任務執行過程任務執行過程源文件：GFSMap處理結果：本地存儲Reduce處理結果：GFS日志：GFS文件存儲位置源文件：GFS文件存儲位置Shuffle和Sort當Map開始產生輸出時，并不是簡單的把數據寫到磁盤，因為頻繁的磁盤操作會導致性能嚴重下降。它的處理過程更復雜，數據首先是寫到內存中的一個緩沖區，并進行預排序，以提升效率。Shuffle和Sort當Map開始產生輸出時，并不是CombinerCombinerCombinerCombiner并行化map()函數可以并行執行，為不同的輸入數據集生成不同的中間結果reduce()函數也可以并行執行，分別處理不同的outputkeymap和reduce的處理過程中不發生通信瓶頸：只有當map處理全部結束后，reduce過程才能夠開始并行化map()函數可以并行執行，為不同的輸入數據集生成不同MapReduce的并行執行MapReduce的并行執行Worker故障Master周期性的ping每個worker。如果master在一個確定的時間段內沒有收到worker返回的信息，那么它將把這個worker標記成失效重新執行該節點上已經執行或尚未執行的Map任務重新執行該節點上未完成的Reduce任務，已完成的不再執行Master故障定期寫入檢查點數據從檢查點恢復MapReduce的容錯Worker故障MapReduce的容錯任務備份機制慢的workers會嚴重地拖延整個執行完成的時間由于其他的任務占用了資源磁盤損壞解決方案：推測性的執行（Speculativeexecution）在即將完成時，備份任務多個worker同時進行相同的任務任何一個完成均可可以十分顯著地提高執行效率MapReduce的優化任務備份機制MapReduce的優化本地處理Master調度策略:向GFS詢問獲得輸入文件blocks副本的位置信息Maptasks的輸入數據通常按64MB來劃分(GFSblock大小)按照blocks所在的機器或機器所在機架的范圍進行調度效果絕大部分機器從本地讀取文件作為輸入，節省大量帶寬MapReduce的優化本地處理MapReduce的優化跳過有問題的記錄一些特定的輸入數據常導致Map/Reduce無法運行調試或者修改在每個worker里運行一個信號處理程序，捕獲map或reduce任務崩潰時發出的信號，一旦捕獲，就會向master報告，同時報告輸入記錄的編號信息。如果master看到一條記錄有兩次崩潰信息，那么就會對該記錄進行標記，下次運行的時候，跳過該記錄MapReduce的優化跳過有問題的記錄MapReduce的優化MapReduce示例：單詞計數案例：單詞記數問題(WordCount)給定一個巨大的文本（如1TB），如何計算單詞出現的數目？MapReduce示例：單詞計數案例：單詞記數問題(WordMapReduce示例：單詞計數使用MapReduce求解該問題Step1:自動對文本進行分割MapReduce示例：單詞計數使用MapReduce求解該MapReduce示例：單詞計數使用MapReduce求解該問題Step2:在分割之后的每一對<key,value>進行用戶定義的Map進行處理，再生成新的<key,value>對MapReduce示例：單詞計數使用MapReduce求解該MapReduce示例：單詞計數使用MapReduce求解該問題Step3:對輸出的結果集歸攏(不同mapslot間copy到一起)、排序(sort)(系統自動完成）MapReduce示例：單詞計數使用MapReduce求解該MapReduce示例：單詞計數使用MapReduce求解該問題Step4:通過Reduce操作生成最后結果MapReduce示例：單詞計數使用MapReduce求解該MapReduce示例：單詞計數使用MapReduce求解該問題定義Map和Reduce函數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));MapReduce示例：單詞計數使用MapReduce求解該其他示例分布式檢索map函數挑選出滿足特定模式的行，并將其組裝成元組輸出。reduce函數是一個簡單的確認函數，它完成的工作僅僅是將中間元組拷貝到輸出中。計算URL訪問頻率map函數處理web網頁的訪問日志，并輸出<URL,1>。reduce函數將每個URL的訪問次數加起來，輸出<URL,totalcount>其他示例分布式檢索其他示例翻轉web-link圖在每個作為源的頁面中，檢查其連接URL，并逐個輸出<target,source>元組。reduce函數將連接到每個target的所有source組合起來，形成list列表，輸出<target,list(source)>每個站點的術語向量術語向量表示出在一篇文章中或者一組文章中最重要的單詞，通常以<word,frequency>元組的方式。map函數輸出每個文章的<hostname,termvector>（hostname通過文章的URL分析得到）。reduce函數取出不常用的術語，將其余的相加，得到最終的<hostname,termvector>對其他示例翻轉web-link圖其他示例倒排索引map函數分析每個文檔,然后產生一個(詞,文檔號)對的序列.reduce函數接受一個給定詞的所有對,排序相應的文檔IDs,并且產生一個(詞,文檔ID列表)對.所有的輸出對集形成一個簡單的倒排索引分布式排序map函數從每個記錄提取key,并且產生一個(key,record)對.reduce函數不改變任何的對.其他示例倒排索引“實踐是檢驗真理的唯一標準”實踐證明，MapReduce是出色的分布式計算模型Google宣布，其對分布于1000臺計算機上的1TB數據進行排序僅僅需要68s對4000臺計算機上的1PB數據進行排序處理僅需要6小時2分鐘（每次測試至少會損壞1塊硬盤）在08年1月份，GoogleMapReduce平均每天的數據處理量是20PB，相當于美國國會圖書館當年5月份存檔網絡數據的240倍“實踐是檢驗真理的唯一標準”實踐證明，MapReduce是出Hadoop上的MapReducejob:是客戶端程序想要完成的一系列工作的集合。包括輸入數據，MapReduce程序和配置信息。task:Hadoop將job分解為tasks有兩種類型的task:maptask和reducetaskjobtracker和tasktracker:用來控制job執行的tasktracker運行task，并向jobtracker報告進度信息jobtracker記錄下每一個job的進度信息，如果一個task失敗，jobtracker會將其重新調度到另外的tasktracker上。Hadoop上的MapReducejob:是客戶端程序想要Hadoop-MapReduce工作原理Hadoop-MapReduce工作原理大數據及MapReduce編程模型課件HadoopStreamingandPipesHadoop流允許用Java以外的語言來編寫Map和Reduce函數Hadoop管道C++接口HadoopStreamingandPipesHado流和管道及子進程的關系流和管道及子進程的關系進度和狀態更新進度和狀態更新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;Hadoop-MapReduceMapper接口描述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;}接口描述publicinterfaceMapper<K1MapperpublicstaticclassMapextendsMapper<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);}}}MapperpublicstaticclassMapReducerpublicstaticclassReduceextendsReducer<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));}}ReducerpublicstaticclassRedDriverpublicstaticvoidmain(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);}Driverpublicstaticvoidmain(InputFiles輸入文件一般保存在HDFS中文件的類型不固定，可能是文本的，也有可能是其它形式的文件文件經常很大，甚至有幾十個GBInput會被分成inputsplit，split由record組成。map處理每一個record，并且返回key和value的對MapReduce程序并不需要直接處理InputSplit，由InputFormat創建的InputFiles輸入文件一般保存在HDFS中InputSplitsInputSplit定義了輸入到單個Map任務的輸入數據InputSplit將文件分為64MB的大小hadoop-site.xml中的mapred.min.split.size參數控制這個大小mapred.tasktracker.map.taks.maximum用來控制某一個節點上所有map任務的最大數目InputSplitsInputSplit定義了輸入到單個MRecordReaderInputSplit定義了一項工作的大小，但是沒有定義如何讀取數據RecordReader實際上定義了如何從數據上轉化為一個(key,value)對，從而輸出到Mapper類中TextInputFormat提供了LineRecordReaderRecordReaderInputSplit定義了一項工作的InputFormat定義了這些文件如何分割，讀取InputFile提供了以下一些功能選擇文件或者其它對象，用來作為輸入定義InputSplits，將一個文件分開成為任務為RecordReader提供一個工廠，用來讀取這個文件有一個抽象的類FileInputFormat，所有的輸入格式類都從這個類繼承這個類的功能以及特性。當啟動一個Hadoop任務的時候，一個輸入文件所在的目錄被輸入到FileInputFormat對象中。FileInputFormat從這個目錄中讀取所有文件。然后FileInputFormat將這些文件分割為一個或者多個InputSplits。通過在JobConf對象上設置JobConf.setInputFormat設置文件輸入的格式InputFormat定義了這些文件如何分割，讀取預定義的文件輸入格式InputFormat:Description:Key:Value:TextInputFormatDefaultformat;readslinesoftextfilesThebyteoffsetofthelineThelinecontentsKeyValueInputFormatParseslinesintokey,valpairsEverythinguptothefirsttabcharacterTheremainderofthelineSequenceFileInputFormatAHadoop-specifichigh-performancebinaryformatuser-defineduser-defined預定義的文件輸入格式InputFormat:Descript各種InputFormatTextInputFormat，默認的格式，每一行是一個單獨的記錄，并且作為value，文件的偏移值作為keyKeyValueInputFormat，這個格式每一行也是一個單獨的記錄，但是Key和Value用Tab隔開，是默認的OutputFormat，可以作為中間結果，作為下一步MapReduce的輸入。SequenceFileInputFormat基于塊進行壓縮的格式對于幾種類型數據的序列化和反序列化操作用來將數據快速讀取到Mapper類中各種InputFormatTextInputFormat，默Writable接口Hadoop使用Writable做序列化定義了兩個方法二進制寫入DataOutput流二進制讀取DataInput流Hadoop自帶一系列Writable實現，可以滿足絕大多數需要可以自定義Writable，控制二進制表示和排序Writable接口Hadoop使用Writable做序列化實現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接口的例子public

class

MyWritable的Java基本封裝Writable的Java基本封裝Mapper每一個Mapper類的實例生成了一個Java進程（在某一個InputSplit上執行）有兩個額外的參數OutputCollector以及Reporter，前者用來收集中間結果，后者用來獲得環境參數以及設置當前執行的狀態。現在用Mapper.Context提供給每一個Mapper函數，用來提供上面兩個對象的功能數據壓縮Mapper每一個Mapper類的實例生成了一個Java進程Partition&Shuffle在Map工作完成之后，每一個Map函數會將結果傳到對應的Reducer所在的節點，此時，用戶可以提供一個Partitioner類，用來決定一個給定的(key,value)對傳輸的具體位置Partition&Shuffle在Map工作完成之后，每一Combinerconf.setCombinerClass(Reduce.class);是在本地執行的一個Reducer，滿足一定的條件才能夠執行。Combinerconf.setCombinerClass(Sort傳輸到每一個節點上的所有的Reduce函數接收到得Key,value對會被Hadoop自動排序（即Map生成的結果傳送到某一個節點的時候，會被自動排序）Sort傳輸到每一個節點上的所有的Reduce函數接收到得KReduce做用戶定義的Reduce操作接收到一個OutputCollector的類作為輸出Reduce做用戶定義的Reduce操作OutputFormat寫入到HDFS的所有OutputFormat都繼承自FileOutputFormat每一個Reducer都寫一個文件到一個共同的輸出目錄，文件名是part-nnnnn，其中nnnnn是與每一個reducer相關的一個號（partitionid）JobConf.setOutputFormat()RecordWriter用來指導如何輸出一個記錄到文件中OutputFormat寫入到HDFS的所有OutputFoOutputFormatOutputFormat:DescriptionTextOutputFormatDefault;writeslinesin"key\tvalue"formSequenceFileOutputFormatWritesbinaryfilessuitableforreadingintosubsequentMapReducejobsNullOutputFormatDisregardsitsinputsOutputFormatOutputFormat:Desc容錯由Hadoop系統自己解決主要方法是將失敗的任務進行再次執行TaskTracker會把狀態信息匯報給JobTracker，最終由JobTracker決定重新執行哪一個任務為了加快執行的速度，Hadoop也會自動重復執行同一個任務，以最先執行成功的為準mapred.map.tasks.speculative.executionmapred.reduce.tasks.speculative.execution容錯由Hadoop系統自己解決調優部分屬性除了配置文件之外還可以在MapReduce作業中動態修改在MapReduce執行過程中，特別是Shuffle階段，盡量使用內存緩沖區存儲數據，減少磁盤溢寫次數；同時在作業執行過程中增加并行度，都能夠顯著提高系統性能，這也是配置優化的一個重要依據。由于每個Hadoop集群的機器和硬件之間都存在一定差別，所以Hadoop框架應根據其集群特性做配置優化調優部分屬性除了配置文件之外還可以在MapReduce作業中IO屬性優化主要包括在Shuffle階段中相關的I/O過程的屬性io.sort.factor屬性int類型，Map端和Reduce端使用

該屬性設置在Map端和Reduce端都使用到的對文件Sort時一次合并的最大流，其默認值是10，即一次合并10個流。在集群中，將其適當增大能夠提高并行度以縮短合并所需時間。將此默認值增加到100是比較常見的。io.sort.mb屬性int類型，Map端使用，Map輸出進行排序時使用的環形內存緩沖區的大小，以M字節為單位，默認是100M。如果允許，應該增加它的值來減少磁盤溢寫的次數以提高性能。io.sort.record.percent屬性float類型，Map端使用，設置保留的io.sort.mb的比例用來存儲Map輸出的記錄邊界，剩余的空間用來存儲Map輸出記錄本身，默認是0.05IO屬性優化主要包括在Shuffle階段中相關的I/O過程的IO屬性優化io.sort.spill.percent屬性float類型，Map端使用，設置Map輸出內存緩沖和邊界記錄索引兩者使用比例的閾值，達到此值后開始溢寫磁盤的過程，默認是0.80io.file.buffer.size屬性int類型，MapReduce作業使用，設置MapReduce作業的I/O操作中所提供的緩沖區的大小，以字節為單位，默認是4096字節。這是一個比較保守的設置，通過增大它的大小能夠減少I/O次數以提高性能。如果系統允許，64KB（65536字節）至128KB（131072字節）是較普遍的選擇。mapred.job.shuffle.input.buffer.percent屬性float類型，Reduce端使用，該屬性設置整個堆空間的百分比，用于Shuffle的復制階段分配給Map輸出緩存，默認是0.70，適當增大比例可以使Map輸出不被溢寫到磁盤，能夠提高系統性能。mapred.job.shuffle.merge.percent屬性float類型，Reduce端使用，該屬性設置Map輸出緩存中使用比例的閾值，用于啟動合并輸出和磁盤溢寫的過程，默認是0.66。如果允許，適當增大其比例能夠減少磁盤溢寫次數，提高系統性能IO屬性優化io.sort.spill.percent屬性Job提交方法submit()submit函數會把Job提交給對應的Cluster，不等待Job執行結束立刻返回。把Job實例的狀態設置為JobState.RUNNING，從而來表示Job正在進行中。在Job運行過程中，可以調用getJobState()來獲取Job的運行狀態waitForCompletion(boolean)waitForCompletion函數會提交Job到對應的Cluster，并等待Job執行結束。函數的boolean參數表示是否打印Job執行的相關信息。返回的結果是一個boolean變量，用來標識Job的執行結果Job提交方法submit()唐卓博ust_tz@126.com湖南大學信息科學與工程學院2014年8月大數據及其并行編程模型概述唐卓博數據及其并行主要內容一、大數據概述二、應對大數據的系統思維三、MapReduce并行編程詳解2注：本課件前30頁PPT來源于國防科大李東升教授：“大數據時代的挑戰和探索”主要內容一、大數據概述2注：本課件前30頁PPT來源于互聯網應用數據急劇增長

互聯網用戶數量巨大，日益活躍

?

微博、論壇、電子商務網站等

?

互聯網上的用戶生成數據（User

Generated

Content，

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

隨著信息化的推進，國民經濟、國家安全

等領域的數據不斷增長

物聯網、移動通信電話、手機短信、語音數據

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

?

高分辨率衛星（影像）、城市監控攝像頭（視頻）、…

?

據報道，武漢監控攝像頭已超過25萬個，如采用1080P高清攝

像頭（一天產生數據量40GB以上），整個城市每天新增監控

數據10PB以上應用背景一、大數據概述?隨著信息化的推進，國民經濟、國家安全應用?

科學實驗數據規模巨大，增長迅猛生物工程氣候監測高能物理天文觀測生態環境

….氣候研究華大基因測序目前每天產生數據約15TB，一年超過5PB

一歐洲CERN對撞機每年產生的數據量超過15

PB基因測序應用背景氣候研究華大基因測序目歐洲CERN對撞基因測序應用背景全球數據量?IDC報告預測：未來

十年，全球數據量繼

續迅速增長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全球數據量?IDC報告預測：未來Amounto?

維基（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.

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

Volume：規模大

從PB級到ZB級

1

ZB

~

106*

PB?

Variety：多樣化

結構化、非結構化

文本、圖像、視頻等?

Velocity：變化快

批處理/離線數據、流/實時/在線數據等?

Value/

Veracity：價值稀疏

/數據質量

噪音和無用信息很多一、大數據概述大數據的特點一、大數據概述?Volume：規模大一、大數據概述大數?

大數據技術對經濟社會和科研都在產生重

要影響

–

互聯網產業、電子商務推薦、日常生活

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

EpidemicsusingSearchEngineQueryData?大數據技術對經濟社會和科研都在產生重大數據的影響季節性?

大數據技術對經濟社會和科研都在產生重

要影響

–

科學研究

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

Data”,Science(2011.2):”Dealing

with

data”大數據的影響?大數據技術對經濟社會和科研都在產生重三種科研模式：理論?

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

美元啟動“大數據研發計劃”（

Big

Data

R&D

Initiative

）

美NSF、國防部、能源部、衛生總署等七部委?

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

2013年973新立項項目：2項

“十二五”

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

目征集指南?

國內外學術界的熱點課題

SIGMOD、

VLDB、OSDI、NSDI等著名會議

Nature、Science雜志11大數據成為熱點課題?2012年3月29日，美國政府宣布投資2億11大數據?

傳統技術難以應對大數據的規模

–

數據存儲及訪問的挑戰當前較快硬盤的傳輸速度6Gbps，線性掃描10PB數據，需約19天而百度、Google等互聯網公司每天處理

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

?

可擴展是大規模分布式系統面臨的基礎性問題

–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大數據帶來的挑戰（1）?傳統技術難以應對大數據的規模當前較快硬盤的傳輸速度6?

很多大數據應用對響應時間要求高

–

規模大、響應快：對存儲和處理提出了很大挑戰

–2007年前,Facebook使用數據庫，總數據量15TB

?

目前，Facebook每天新增加的數據約70TB

–

傳統并行數據庫擴展性受限，節點規模很少超過100，

且價格昂貴

?2011年，Facebook系統具有2700多個節點，Google單個數據中心在上

萬個節點集群上存儲了約10PB數據?

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

數據存儲和處理系統？大數據存儲與處理的可擴展難題大數據存儲與處理的可擴展難題數據種類多，需求多樣，關聯復雜

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

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

化的復雜數據，挖掘數據價值？

–

大數據中垃圾和珍寶并存

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

–需要計算機專家和領域專家的配合….大數據面臨的挑戰（2）數據種類多，需求多樣，關聯復雜大數據面臨的挑戰（2）傳統算法在大數據時代可能不再有效

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

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

需要大數據計算思維上的變化

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

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

一點不準確也沒有關系傳統算法結論在大數據時代需要重新評估

–

簡單方法+大數據集可能取得很好的結果大數據面臨的挑戰（3）傳統算法在大數據時代可能不再有效大數據面臨的挑戰（3）?

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

器翻譯領域中基于單詞訓練數據集的語言

模型

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

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

研究表明，簡單算法在小數

據集時效果不佳，但在大數

據集時，簡單算法卻產生了

更好的效果

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

ProceedingsoftheJointConferenceonEmpiricalMethodsinNatural

LanguageProcessingandComputationalNaturalLanguageLearning,2007.16傳統算法結論需要重新評估?2007年,Google公司的Brants等人研究?

大數據時代的算法新理論

–

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

復雜大數據的建模、表示和可視化

–

多源異構大數據：由大到小?

面向大數據的新型存儲和計算系統架構

–大規模并行/分布處理?

大數據（并行）挖掘算法及應用大數據的研究課題?大數據時代的算法新理論大數據的研究課題主要內容一、大數據概述二、應對大數據的系統思維三、MapReduce并行編程詳解2主要內容一、大數據概述2181.

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

可擴展的數據并行處理3.求同存異，聚焦領域放松傳統數據處理技術的約束，如一致性等、行式存儲-列式存儲高可擴展高吞吐率高可靠性……主要內容18二、應對大數據的系統思維181.數據為中心的計算架構高可擴展主要內容18二、應對大1.

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

和

on-chip

L2之間:

帶寬:增長了2250倍

延遲:降低了20倍L3

cache

和DRAM之間：

帶寬:增長了125倍

延遲:降低了4倍DRAM

和disk之間:

帶寬:增長了150倍

延遲:降低了8倍

LAN連接的兩個節點之間

:

帶寬:增長了100倍

延遲:降低了15倍充分利用數據和存儲的局部性（緩存、復制、預取）延遲提升滯后于帶寬Source：CACM（Patterson)1.數據為中心的計算架構過去20年來,計算器件的帶寬充分二、應對大數據的系統思維1.

數據為中心的計算架構（續）20二、應對大數據的計算思維

–

數據分布存儲在計算附近？–

計算盡量利用數據局部性–

存儲架構、互連網絡架構數據密集型計算計算密集型計算

SystemData–

數據存儲與計算相分離–

計算之前加載數據–

規模挑戰：元