Hadoop權(quán)威指南-中文版

上傳人：1*** IP屬地：廣西上傳時(shí)間：2023-11-03 格式：DOC 頁(yè)數(shù)：71 大?。?50KB 積分：15 舉報(bào) 版權(quán)申訴

已閱讀5頁(yè)，還剩66頁(yè)未讀，繼續(xù)免費(fèi)閱讀

版權(quán)說(shuō)明：本文檔由用戶提供并上傳，收益歸屬內(nèi)容提供方，若內(nèi)容存在侵權(quán)，請(qǐng)進(jìn)行舉報(bào)或認(rèn)領(lǐng)

文檔簡(jiǎn)介

初識(shí)Hadoop

1.1

數(shù)據(jù)！數(shù)據(jù)

1.2

數(shù)據(jù)的存儲(chǔ)和分析

1.3

相較于其他系統(tǒng)

1.4

Hadoop發(fā)展簡(jiǎn)史

1.5

ApacheHadoop項(xiàng)目

MapReduce簡(jiǎn)介

2.1

一個(gè)氣象數(shù)據(jù)集

2.2

使用UnixTools來(lái)分析數(shù)據(jù)

2.3

使用Hadoop進(jìn)行數(shù)據(jù)分析

2.4

分布化

2.5

Hadoop流

2.6

Hadoop管道

Hadoop分布式文件系統(tǒng)

3.1

HDFS的設(shè)計(jì)

3.2

HDFS的概念

3.3

命令行接口

3.4

Hadoop文件系統(tǒng)

3.5

Java接口

3.6

數(shù)據(jù)流

3.7

通過(guò)distcp進(jìn)行并行復(fù)制

3.8

Hadoop歸檔文件

Hadoop的I/O

4.1

數(shù)據(jù)完整性

4.2

壓縮

4.3

序列化

4.4

基于文件的數(shù)據(jù)結(jié)構(gòu)

111

MapReduce應(yīng)用開(kāi)發(fā)

125

5.1

API的配置

126

5.2

配置開(kāi)發(fā)環(huán)境

128

5.3

編寫單元測(cè)試

134

5.4

本地運(yùn)行測(cè)試數(shù)據(jù)

138

5.5

在集群上運(yùn)行

144

5.6

作業(yè)調(diào)優(yōu)

159

5.7

MapReduce的工作流

162

MapReduce的工作原理

166

6.1

運(yùn)行MapReduce作業(yè)

166

6.2

失敗

172

6.3

作業(yè)的調(diào)度

174

6.4

shuffle和排序

175

6.6

任務(wù)的執(zhí)行

181

MapReduce的類型與格式

188

7.1

MapReduce類型

188

7.3

輸出格式

217

MapReduce特性

227

8.1

計(jì)數(shù)器

227

8.2

排序

235

8.3

聯(lián)接

252

8.4

次要數(shù)據(jù)的分布

258

8.5

MapReduce的類庫(kù)

263

Hadoop集群的安裝

264

9.1

集群說(shuō)明

264

9.2

集群的建立和安裝

268

9.3

SSH配置

270

9.4

Hadoop配置

271

9.5

安裝之后

286

9.6

Hadoop集群基準(zhǔn)測(cè)試

286

9.7

云計(jì)算中的Hadoop

290

Hadoop的管理

293

10.1

HDFS

293

10.2

監(jiān)控

306

10.3

維護(hù)

313

Pig簡(jiǎn)介

321

11.1

安裝和運(yùn)行Pig

322

11.2

實(shí)例

325

11.3

與數(shù)據(jù)庫(kù)比較

329

11.4

PigLatin

330

11.5

用戶定義函數(shù)

343

11.6

數(shù)據(jù)處理操作符

353

11.7

Pig實(shí)踐提示與技巧

363

Hbase簡(jiǎn)介

366

12.1

HBase基礎(chǔ)

366

12.2

概念

367

12.3

安裝

371

12.4

客戶端

374

12.5

示例

377

12.6

HBase與RDBMS的比較

385

12.7

實(shí)踐

390

ZooKeeper簡(jiǎn)介

394

13.1

ZooKeeper的安裝和運(yùn)行

395

13.2

范例

396

13.3

ZooKeeper服務(wù)

405

13.4

使用ZooKeeper建立應(yīng)用程序

417

13.5

工業(yè)界中的ZooKeeper

428

案例研究

431

14.1

Hadoop在Last.fm的應(yīng)用

431

14.2

Hadoop和Hive在Facebook的應(yīng)用

441

14.3

Hadoop在Nutch搜索引擎

451

14.4

Hadoop用于Rackspace的日志處理

466

14.5

Cascading項(xiàng)目

474

14.6

ApacheHadoop的1TB排序

488

ApacheHadoop的安裝

491

Cloudera的Hadoop分發(fā)包

497

預(yù)備NCDC氣象資料

502第1章初識(shí)Hadoop古時(shí)候，人們用牛來(lái)拉重物，當(dāng)一頭牛拉不動(dòng)一根圓木的時(shí)候，他們不曾想過(guò)培育個(gè)頭更大的牛。同樣，我們也不需要嘗試更大的計(jì)算機(jī)，而是應(yīng)該開(kāi)發(fā)更多的計(jì)算系統(tǒng)。--格蕾斯·霍珀1.1

數(shù)據(jù)！數(shù)據(jù)我們生活在數(shù)據(jù)時(shí)代！很難估計(jì)全球存儲(chǔ)的電子數(shù)據(jù)總量是多少，但是據(jù)IDC估計(jì)2006年"數(shù)字全球"項(xiàng)目(digitaluniverse)的數(shù)據(jù)總量為0.18ZB，并且預(yù)測(cè)到2011年這個(gè)數(shù)字將達(dá)到1.8ZB，為2006年的10倍。1ZB相當(dāng)于10的21次方字節(jié)的數(shù)據(jù)，或者相當(dāng)于1000EB，1000000PB，或者大家更熟悉的10億TB的數(shù)據(jù)！這相當(dāng)于世界上每個(gè)人一個(gè)磁盤驅(qū)動(dòng)器的數(shù)量級(jí)。這一數(shù)據(jù)洪流有許多來(lái)源?？紤]下文：

紐約證券交易所每天產(chǎn)生1TB的交易數(shù)據(jù)。著名社交網(wǎng)站Facebook的主機(jī)存儲(chǔ)著約100億張照片，占據(jù)PB級(jí)存儲(chǔ)空間。A，一個(gè)家譜網(wǎng)站，存儲(chǔ)著2.5PB數(shù)據(jù)?；ヂ?lián)網(wǎng)檔案館(TheInternetArchive)存儲(chǔ)著約2PB數(shù)據(jù)，并以每月至少20TB的速度增長(zhǎng)。瑞士日內(nèi)瓦附近的大型強(qiáng)子對(duì)撞機(jī)每年產(chǎn)生約15PB的數(shù)據(jù)。此外還有大量數(shù)據(jù)。但是你可能會(huì)想它對(duì)自己有何影響。大部分?jǐn)?shù)據(jù)被鎖定在最大的網(wǎng)頁(yè)內(nèi)容里面(如搜索引擎)或者是金融和科學(xué)機(jī)構(gòu)，對(duì)不對(duì)？是不是所謂的"大數(shù)據(jù)"的出現(xiàn)會(huì)影響到較小的組織或個(gè)人？我認(rèn)為是這樣的。以照片為例，我妻子的祖父是一個(gè)狂熱的攝影愛(ài)好者，并且他成人之后，幾乎一直都在拍照片。他的所有照片(中等格式、幻燈片和35mm膠片)，在掃描成高解析度照片時(shí)，占了大約10GB的空間。相比之下，我家去年一年用數(shù)碼相機(jī)拍攝的照片就占用了5GB的空間。我家產(chǎn)生照片數(shù)據(jù)的速度是我妻子祖父的35倍！并且，隨著拍攝更多的照片變得越來(lái)越容易，這個(gè)速度還在增加中。更常見(jiàn)的情況是，個(gè)人數(shù)據(jù)的產(chǎn)生量正在快速地增長(zhǎng)。微軟研究院的MyLifeBits項(xiàng)目()顯示，在不久的將來(lái)，個(gè)人信息檔案將可能成為普遍現(xiàn)象。MyLifeBits是這樣的一個(gè)實(shí)驗(yàn)：一個(gè)人與外界的聯(lián)系(電話、郵件和文件)被抓取和存儲(chǔ)供以后訪問(wèn)。收集的數(shù)據(jù)包括每分鐘拍攝的照片等，導(dǎo)致整個(gè)數(shù)據(jù)量達(dá)到每月1GB的大小。當(dāng)存儲(chǔ)成本下降到使其可以存儲(chǔ)連續(xù)的音頻和視頻時(shí)，服務(wù)于未來(lái)MyLifeBits項(xiàng)目的數(shù)據(jù)量將是現(xiàn)在的許多倍。個(gè)人數(shù)據(jù)的增長(zhǎng)的確是大勢(shì)所趨，但更重要的是，計(jì)算機(jī)所產(chǎn)生的數(shù)據(jù)可能比人所產(chǎn)生的數(shù)據(jù)更大。機(jī)器日志、RFID讀取器、傳感器網(wǎng)絡(luò)、車載GPS和零售交易數(shù)據(jù)等，這些都會(huì)促使"數(shù)據(jù)之山越來(lái)越高"。公開(kāi)發(fā)布的數(shù)據(jù)量也在逐年增加。作為組織或企業(yè)，再也不能只管理自己的數(shù)據(jù)，未來(lái)的成功在很大程度上取決于它是否能從其他組織的數(shù)據(jù)中提取出價(jià)值。這方面的先鋒(如亞馬遜網(wǎng)絡(luò)服務(wù)器、I或者)的公共數(shù)據(jù)集，它們的存在就在于促進(jìn)"信息共享"，任何人都可以共享并自由(或以AWS平臺(tái)的形式，或以適度的價(jià)格)下載和分析這些數(shù)據(jù)。不同來(lái)源的信息混合處理后會(huì)帶來(lái)意外的效果和至今難以想像的應(yīng)用。以A項(xiàng)目為例，這是一個(gè)研究Flickr網(wǎng)站上天體愛(ài)好者群中新照片的項(xiàng)目。它分析每一張上傳的照片，并確定它是天空的哪一部分，或者是否是有趣的天體，如恒星或者星系。雖然這只是一個(gè)帶實(shí)驗(yàn)性質(zhì)的新服務(wù)，但是它顯示了數(shù)據(jù)(這里特指攝影照片)的可用性并且被用來(lái)進(jìn)行某些活動(dòng)(圖像分析)，而這些活動(dòng)很多時(shí)候并不是數(shù)據(jù)創(chuàng)建者預(yù)先能夠想像到的。有句話是這么說(shuō)的："算法再好，通常也難敵更多的數(shù)據(jù)。"意思是說(shuō)對(duì)于某些問(wèn)題(譬如基于既往偏好生成的電影和音樂(lè)推薦)，不論你的算法有多么猛，它們總是會(huì)在更多的數(shù)據(jù)面前無(wú)能為力(更不用說(shuō)沒(méi)有優(yōu)化過(guò)的算法了)?，F(xiàn)在，我們有一個(gè)好消息和一個(gè)壞消息。好消息是有海量數(shù)據(jù)！壞消息是我們正在為存儲(chǔ)和分析這些數(shù)據(jù)而奮斗不息。1.2

數(shù)據(jù)的存儲(chǔ)和分析問(wèn)題很簡(jiǎn)單：多年來(lái)硬盤存儲(chǔ)容量快速增加的同時(shí)，訪問(wèn)速度--數(shù)據(jù)從硬盤讀取的速度--卻未能與時(shí)俱進(jìn)。1990年，一個(gè)普通的硬盤驅(qū)動(dòng)器可存儲(chǔ)1370MB的數(shù)據(jù)并擁有4.4MB/s的傳輸速度，所以，只需五分鐘的時(shí)間就可以讀取整個(gè)磁盤的數(shù)據(jù)。20年過(guò)去了，1TB級(jí)別的磁盤驅(qū)動(dòng)器是很正常的，但是數(shù)據(jù)傳輸?shù)乃俣葏s在100MB/s左右。所以它需要花兩個(gè)半小時(shí)以上的時(shí)間讀取整個(gè)驅(qū)動(dòng)器的數(shù)據(jù)。從一個(gè)驅(qū)動(dòng)器上讀取所有的數(shù)據(jù)需要很長(zhǎng)的時(shí)間，寫甚至更慢。一個(gè)很簡(jiǎn)單的減少讀取時(shí)間的辦法是同時(shí)從多個(gè)磁盤上讀取數(shù)據(jù)。試想一下，我們擁有100個(gè)磁盤，每個(gè)存儲(chǔ)百分之一的數(shù)據(jù)。如果它們并行運(yùn)行，那么不到兩分鐘我們就可以讀完所有的數(shù)據(jù)。只使用一個(gè)磁盤的百分之一似乎很浪費(fèi)。但是我們可以存儲(chǔ)100個(gè)數(shù)據(jù)集，每個(gè)1TB，并讓它們共享磁盤的訪問(wèn)。我們可以想像，此類系統(tǒng)的用戶會(huì)很高興看到共享訪問(wèn)可以縮短分析時(shí)間，并且，從統(tǒng)計(jì)角度來(lái)看，他們的分析工作會(huì)分散到不同的時(shí)間點(diǎn)，所以互相之間不會(huì)有太多干擾。盡管如此，現(xiàn)在更可行的是從多個(gè)磁盤并行讀寫數(shù)據(jù)。第一個(gè)需要解決的問(wèn)題是硬件故障。一旦開(kāi)始使用多個(gè)硬件設(shè)施，其中一個(gè)會(huì)出故障的概率是非常高的。避免數(shù)據(jù)丟失的常見(jiàn)做法是復(fù)制：通過(guò)系統(tǒng)保存數(shù)據(jù)的冗余副本，在故障發(fā)生時(shí)，可以使用數(shù)據(jù)的另一份副本。這就是冗余磁盤陣列的工作方式。Hadoop的文件系統(tǒng)HDFS(HadoopDistributedFilesystem)也是一個(gè)例子，雖然它采取的是另一種稍有不同的方法，詳見(jiàn)后文描述。第二個(gè)問(wèn)題是大部分分析任務(wù)需要通過(guò)某種方式把數(shù)據(jù)合并起來(lái)，即從一個(gè)磁盤讀取的數(shù)據(jù)可能需要和另外99個(gè)磁盤中讀取的數(shù)據(jù)合并起來(lái)才能使用。各種不同的分布式系統(tǒng)能夠組合多個(gè)來(lái)源的數(shù)據(jù)，但是如何保證正確性是一個(gè)非常難的挑戰(zhàn)。MapReduce提供了一個(gè)編程模型，其抽象出上述磁盤讀寫的問(wèn)題，將其轉(zhuǎn)換為計(jì)算一個(gè)由成對(duì)鍵/值組成的數(shù)據(jù)集。這種模型的具體細(xì)節(jié)將在后面的章節(jié)討論。但是目前討論的重點(diǎn)是，這個(gè)計(jì)算由兩部分組成：Map和Reduce。這兩者的接口就是"整合"之地。就像HDFS一樣，MapReduce是內(nèi)建可靠性這個(gè)功能的。簡(jiǎn)而言之，Hadoop提供了一個(gè)穩(wěn)定的共享存儲(chǔ)和分析系統(tǒng)。存儲(chǔ)由HDFS實(shí)現(xiàn)，分析由MapReduce實(shí)現(xiàn)?？v然Hadoop還有其他功能，但這些功能是它的核心所在。1.3

相較于其他系統(tǒng)MapReduce似乎采用的是一種蠻力方法。即，針對(duì)每個(gè)查詢，每一個(gè)數(shù)據(jù)集--至少是很大一部分--都會(huì)被處理。但這正是它的能力。MapReduce可以處理一批查詢，并且它針對(duì)整個(gè)數(shù)據(jù)集處理即席查詢并在合理時(shí)間內(nèi)獲得結(jié)果的能力也是具有突破性的。它改變了我們對(duì)數(shù)據(jù)的看法，并且解放了以前存儲(chǔ)在磁帶和磁盤上的數(shù)據(jù)。它賦予我們對(duì)數(shù)據(jù)進(jìn)行創(chuàng)新的機(jī)會(huì)。那些以前需要很長(zhǎng)時(shí)間才能獲得答案的問(wèn)題現(xiàn)在已經(jīng)迎刃而解，但反過(guò)來(lái)，這又帶來(lái)了新的問(wèn)題和見(jiàn)解。例如，Rackspace的郵件部門Mailtrust，用Hadoop處理郵件的日志。他們寫的一個(gè)查詢是找到其用戶的地理分布。他們是這樣說(shuō)的："隨著我們的壯大，這些數(shù)據(jù)非常有用，我們每月運(yùn)行一次MapReduce任務(wù)來(lái)幫助我們決定哪些Rackspace數(shù)據(jù)中心需要添加新的郵件服務(wù)器。"通過(guò)將數(shù)百GB的數(shù)據(jù)整合，借助于分析工具，Rackspace的工程師得以了解這些數(shù)據(jù)，否則他們永遠(yuǎn)都不會(huì)了解，并且他們可以運(yùn)用這些信息去改善他們?yōu)橛脩籼峁┑姆?wù)。第14章將詳細(xì)介紹Rackspace公司是如何運(yùn)用Hadoop的。1.3.1

關(guān)系型數(shù)據(jù)庫(kù)管理系統(tǒng)為什么我們不能使用數(shù)據(jù)庫(kù)加上更多磁盤來(lái)做大規(guī)模的批量分析？為什么我們需要MapReduce？這個(gè)問(wèn)題的答案來(lái)自于磁盤驅(qū)動(dòng)器的另一個(gè)發(fā)展趨勢(shì)：尋址時(shí)間的提高速度遠(yuǎn)遠(yuǎn)慢于傳輸速率的提高速度。尋址就是將磁頭移動(dòng)到特定位置進(jìn)行讀寫操作的工序。它的特點(diǎn)是磁盤操作有延遲，而傳輸速率對(duì)應(yīng)于磁盤的帶寬。如果數(shù)據(jù)的訪問(wèn)模式受限于磁盤的尋址，勢(shì)必會(huì)導(dǎo)致它花更長(zhǎng)時(shí)間(相較于流)來(lái)讀或?qū)懘蟛糠謹(jǐn)?shù)據(jù)。另一方面，在更新一小部分?jǐn)?shù)據(jù)庫(kù)記錄的時(shí)候，傳統(tǒng)的B樹(shù)(關(guān)系型數(shù)據(jù)庫(kù)中使用的一種數(shù)據(jù)結(jié)構(gòu)，受限于執(zhí)行查找的速度)效果很好。但在更新大部分?jǐn)?shù)據(jù)庫(kù)數(shù)據(jù)的時(shí)候，B樹(shù)的效率就沒(méi)有MapReduce的效率高，因?yàn)樗枰褂门判?合并來(lái)重建數(shù)據(jù)庫(kù)。在許多情況下，MapReduce能夠被視為一種RDBMS(關(guān)系型數(shù)據(jù)庫(kù)管理系統(tǒng))的補(bǔ)充。(兩個(gè)系統(tǒng)之間的差異見(jiàn)表1-1)。MapReduce很適合處理那些需要分析整個(gè)數(shù)據(jù)集的問(wèn)題，以批處理的方式，尤其是AdHoc(自主或即時(shí))分析。RDBMS適用于點(diǎn)查詢和更新(其中，數(shù)據(jù)集已經(jīng)被索引以提供低延遲的檢索和短時(shí)間的少量數(shù)據(jù)更新。MapReduce適合數(shù)據(jù)被一次寫入和多次讀取的應(yīng)用，而關(guān)系型數(shù)據(jù)庫(kù)更適合持續(xù)更新的數(shù)據(jù)集。表1-1：關(guān)系型數(shù)據(jù)庫(kù)和MapReduce的比較傳統(tǒng)關(guān)系型數(shù)據(jù)庫(kù)MapReduce數(shù)據(jù)大小GBPB訪問(wèn)交互型和批處理批處理更新多次讀寫一次寫入多次讀取結(jié)構(gòu)靜態(tài)模式動(dòng)態(tài)模式集成度高低伸縮性非線性線性

MapReduce和關(guān)系型數(shù)據(jù)庫(kù)之間的另一個(gè)區(qū)別是它們操作的數(shù)據(jù)集中的結(jié)構(gòu)化數(shù)據(jù)的數(shù)量。結(jié)構(gòu)化數(shù)據(jù)是擁有準(zhǔn)確定義的實(shí)體化數(shù)據(jù)，具有諸如XML文檔或數(shù)據(jù)庫(kù)表定義的格式，符合特定的預(yù)定義模式。這就是RDBMS包括的內(nèi)容。另一方面，半結(jié)構(gòu)化數(shù)據(jù)比較寬松，雖然可能有模式，但經(jīng)常被忽略，所以它只能用作數(shù)據(jù)結(jié)構(gòu)指南。例如，一張電子表格，其中的結(jié)構(gòu)便是單元格組成的網(wǎng)格，盡管其本身可能保存任何形式的數(shù)據(jù)。非結(jié)構(gòu)化數(shù)據(jù)沒(méi)有什么特別的內(nèi)部結(jié)構(gòu)，例如純文本或圖像數(shù)據(jù)。MapReduce對(duì)于非結(jié)構(gòu)化或半結(jié)構(gòu)化數(shù)據(jù)非常有效，因?yàn)樗辉O(shè)計(jì)為在處理時(shí)間內(nèi)解釋數(shù)據(jù)。換句話說(shuō)：MapReduce輸入的鍵和值并不是數(shù)據(jù)固有的屬性，它們是由分析數(shù)據(jù)的人來(lái)選擇的。關(guān)系型數(shù)據(jù)往往是規(guī)范的，以保持其完整性和刪除冗余。規(guī)范化為MapReduce帶來(lái)問(wèn)題，因?yàn)樗棺x取記錄成為一個(gè)非本地操作，并且MapReduce的核心假設(shè)之一就是，它可以進(jìn)行(高速)流的讀寫。Web服務(wù)器日志是記錄集的一個(gè)很好的非規(guī)范化例子(例如，客戶端主機(jī)名每次都以全名來(lái)指定，即使同一客戶端可能會(huì)出現(xiàn)很多次)，這也是MapReduce非常適合用于分析各種日志文件的原因之一。MapReduce是一種線性的可伸縮的編程模型。程序員編寫兩個(gè)函數(shù)--

map函數(shù)和Reduce函數(shù)--每一個(gè)都定義一個(gè)鍵/值對(duì)集映射到另一個(gè)。這些函數(shù)無(wú)視數(shù)據(jù)的大小或者它們正在使用的集群的特性，這樣它們就可以原封不動(dòng)地應(yīng)用到小規(guī)模數(shù)據(jù)集或者大的數(shù)據(jù)集上。更重要的是，如果放入兩倍的數(shù)據(jù)量，運(yùn)行的時(shí)間會(huì)少于兩倍。但是如果是兩倍大小的集群，一個(gè)任務(wù)任然只是和原來(lái)的一樣快。這不是一般的SQL查詢的效果。隨著時(shí)間的推移，關(guān)系型數(shù)據(jù)庫(kù)和MapReduce之間的差異很可能變得模糊。關(guān)系型數(shù)據(jù)庫(kù)都開(kāi)始吸收MapReduce的一些思路(如ASTERDATA的和GreenPlum的數(shù)據(jù)庫(kù))，另一方面，基于MapReduce的高級(jí)查詢語(yǔ)言(如Pig和Hive)使MapReduce的系統(tǒng)更接近傳統(tǒng)的數(shù)據(jù)庫(kù)編程人員。1.3.2

網(wǎng)格計(jì)算高性能計(jì)算(HighPerformanceComputing，HPC)和網(wǎng)格計(jì)算社區(qū)多年來(lái)一直在做大規(guī)模的數(shù)據(jù)處理，它們使用的是消息傳遞接口(MessagePassingInterface，MPI)這樣的API。從廣義上講，高性能計(jì)算的方法是將作業(yè)分配給一個(gè)機(jī)器集群，這些機(jī)器訪問(wèn)共享文件系統(tǒng)，由一個(gè)存儲(chǔ)區(qū)域網(wǎng)絡(luò)(StorageAreaNetwork，SAN)進(jìn)行管理。這非常適用于以主計(jì)算密集型為主的作業(yè)，但當(dāng)節(jié)點(diǎn)需要訪問(wèn)的大數(shù)據(jù)量(數(shù)百GB的數(shù)據(jù)，這是MapReduce實(shí)際開(kāi)始"發(fā)光"的起點(diǎn))時(shí)，這會(huì)成為一個(gè)問(wèn)題，因?yàn)榫W(wǎng)絡(luò)帶寬成為"瓶頸"，所以計(jì)算節(jié)點(diǎn)閑置下來(lái)了。MapReduce嘗試在計(jì)算節(jié)點(diǎn)本地存儲(chǔ)數(shù)據(jù)，因此數(shù)據(jù)訪問(wèn)速度會(huì)因?yàn)樗潜镜財(cái)?shù)據(jù)而比較快。這項(xiàng)"數(shù)據(jù)本地化"功能，成為MapReduce的核心功能并且也是它擁有良好性能的原因之一。意識(shí)到網(wǎng)絡(luò)帶寬在數(shù)據(jù)中心環(huán)境是最有價(jià)值的資源(到處復(fù)制數(shù)據(jù)會(huì)很容易的把網(wǎng)絡(luò)帶寬飽和)之后，MapReduce便通過(guò)顯式網(wǎng)絡(luò)拓?fù)浣Y(jié)構(gòu)不遺余力地加以保護(hù)。請(qǐng)注意，這種安排不會(huì)排除MapReduce中的高CPU使用分析。MPI賦予程序員很大的控制，但也要求顯式控制數(shù)據(jù)流機(jī)制，需要使用傳統(tǒng)的C語(yǔ)言的功能模塊完成(例如socket)，以及更高級(jí)的算法來(lái)進(jìn)行分析。而MapReduce卻是在更高層面上完成任務(wù)，即程序員從鍵/值對(duì)函數(shù)的角度來(lái)考慮，同時(shí)數(shù)據(jù)流是隱含的。在一個(gè)大規(guī)模分布式計(jì)算平臺(tái)上協(xié)調(diào)進(jìn)程是一個(gè)很大的挑戰(zhàn)。最困難的部分是恰當(dāng)?shù)奶幚硎c錯(cuò)誤--在不知道一個(gè)遠(yuǎn)程進(jìn)程是否已經(jīng)失敗的時(shí)候--仍然需要繼續(xù)整個(gè)計(jì)算。MapReduce將程序員從必須考慮失敗任務(wù)的情況中解放出來(lái)，它檢測(cè)失敗的map或者reduce任務(wù)，在健康的機(jī)器上重新安排任務(wù)。MapReduce能夠做到這一點(diǎn)，因?yàn)樗且粋€(gè)無(wú)共享的架構(gòu)，這意味著各個(gè)任務(wù)之間彼此并不依賴。(這里講得稍微簡(jiǎn)單了一些，因?yàn)閙apper的輸出是反饋給reducer的，但這由MapReduce系統(tǒng)控制。在這種情況下，相對(duì)于返回失敗的map，應(yīng)該對(duì)返回reducer給予更多關(guān)注，因?yàn)樗仨毚_保它可以檢索到必要的map輸出，如果不行，必須重新運(yùn)行相關(guān)的map從而生成必要的這些輸出。)因此，從程序員的角度來(lái)看，執(zhí)行任務(wù)的順序是無(wú)關(guān)緊要的。相比之下，MPI程序必須顯式地管理自己的檢查點(diǎn)和恢復(fù)機(jī)制，從而把更多控制權(quán)交給程序員，但這樣會(huì)加大編程的難度。MapReduce聽(tīng)起來(lái)似乎是一個(gè)相當(dāng)嚴(yán)格的編程模型，而且在某種意義上看的確如此：我們被限定于鍵/值對(duì)的類型(它們按照指定的方式關(guān)聯(lián)在一起)，mapper和reducer彼此間的協(xié)作有限，一個(gè)接一個(gè)地運(yùn)行(mapper傳輸鍵/值對(duì)給reducer)。對(duì)此，一個(gè)很自然的問(wèn)題是：你是否能用它做點(diǎn)兒有用或普通的事情？答案是肯定的。MapReduce作為一個(gè)建立搜索索引產(chǎn)品系統(tǒng)，是由Google的工程師們開(kāi)發(fā)出來(lái)的，因?yàn)樗麄儼l(fā)現(xiàn)自己一遍又一遍地解決相同的問(wèn)題(MapReduce的靈感來(lái)自傳統(tǒng)的函數(shù)式編程、分布式計(jì)算和數(shù)據(jù)庫(kù)社區(qū))，但它后來(lái)被應(yīng)用于其他行業(yè)的其他許多應(yīng)用。我們驚喜地看到許多算法的變體在MapReduce中得以表示，從圖像圖形分析，到基于圖表的問(wèn)題，再到機(jī)器學(xué)習(xí)算法。它當(dāng)然不能解決所有問(wèn)題，但它是一個(gè)很普遍的數(shù)據(jù)處理工具。第14章將介紹一些Hadoop應(yīng)用范例。1.3.3

志愿計(jì)算人們第一次聽(tīng)說(shuō)Hadoop和MapReduce的時(shí)候，經(jīng)常會(huì)問(wèn)："和SETI@home有什么區(qū)別？"SETI，全稱為SearchforExtra-TerrestrialIntelligence(搜尋外星人)，運(yùn)行著一個(gè)稱為SETI@home的項(xiàng)目()。在此項(xiàng)目中，志愿者把自己計(jì)算機(jī)CPU的空閑時(shí)間貢獻(xiàn)出來(lái)分析無(wú)線天文望遠(yuǎn)鏡的數(shù)據(jù)借此尋外星智慧生命信號(hào)。SETI@home是最有名的擁有許多志愿者的項(xiàng)目，其他的還有GreatInternetMersennePrimeSearch(搜索大素?cái)?shù))與Folding@home項(xiàng)目(了解蛋白質(zhì)構(gòu)成及其與疾病之間的關(guān)系)。志愿計(jì)算項(xiàng)目通過(guò)將他們?cè)噲D解決的問(wèn)題分為幾個(gè)他們成為工作單元的塊來(lái)工作，并將它們送到世界各地的電腦上進(jìn)行分析。例如，SETI@home的工作單元大約是0.35MB的無(wú)線電望遠(yuǎn)鏡數(shù)據(jù)，并且一個(gè)典型的計(jì)算機(jī)需要數(shù)小時(shí)或數(shù)天來(lái)分析。完成分析后，結(jié)果發(fā)送回服務(wù)器，客戶端獲得的另一項(xiàng)工作單元。作為防止欺騙的預(yù)防措施，每個(gè)工作單元必須送到三臺(tái)機(jī)器上并且需要有至少兩個(gè)結(jié)果相同才會(huì)被接受。雖然SETI@home在表面上可能類似于MapReduce(將問(wèn)題分為獨(dú)立的塊，然后進(jìn)行并行計(jì)算)，但差異還是顯著的。SETI@home問(wèn)題是CPU高度密集型的，使其適合運(yùn)行于世界各地成千上萬(wàn)臺(tái)計(jì)算機(jī)上，因?yàn)橄鄬?duì)于其計(jì)算時(shí)間而言，傳輸工作單元的時(shí)間微不足道。志愿者捐獻(xiàn)的是CPU周期，而不是帶寬。MapReduce被設(shè)計(jì)為用來(lái)運(yùn)行那些需要數(shù)分鐘或數(shù)小時(shí)的作業(yè)，這些作業(yè)在一個(gè)聚集帶寬很高的數(shù)據(jù)中心中可信任的專用硬件設(shè)備上運(yùn)行。相比之下，SETI@home項(xiàng)目是在接入互聯(lián)網(wǎng)的不可信的計(jì)算機(jī)上運(yùn)行，這些計(jì)算機(jī)的網(wǎng)速不同，而且數(shù)據(jù)也不在本地。1.4

Hadoop發(fā)展簡(jiǎn)史Hadoop是DougCutting--ApacheLucene創(chuàng)始人--開(kāi)發(fā)的使用廣泛的文本搜索庫(kù)。Hadoop起源于ApacheNutch，后者是一個(gè)開(kāi)源的網(wǎng)絡(luò)搜索引擎，本身也是由Lucene項(xiàng)目的一部分。Hadoop名字的起源Hadoop這個(gè)名字不是一個(gè)縮寫，它是一個(gè)虛構(gòu)的名字。該項(xiàng)目的創(chuàng)建者，DougCutting如此解釋Hadoop的得名："這個(gè)名字是我孩子給一頭吃飽了的棕黃色大象命名的。我的命名標(biāo)準(zhǔn)就是簡(jiǎn)短，容易發(fā)音和拼寫，沒(méi)有太多的意義，并且不會(huì)被用于別處。小孩子是這方面的高手。Googol就是由小孩命名的。"Hadoop及其子項(xiàng)目和后繼模塊所使用的名字往往也與其功能不相關(guān)，經(jīng)常用一頭大象或其他動(dòng)物主題(例如："Pig")。較小的各個(gè)組成部分給與更多描述性(因此也更俗)的名稱。這是一個(gè)很好的原則，因?yàn)樗馕吨梢源笾聫钠涿植聹y(cè)其功能，例如，jobtracker的任務(wù)就是跟蹤MapReduce作業(yè)。從頭開(kāi)始構(gòu)建一個(gè)網(wǎng)絡(luò)搜索引擎是一個(gè)雄心勃勃的目標(biāo)，不只是要編寫一個(gè)復(fù)雜的、能夠抓取和索引網(wǎng)站的軟件，還需要面臨著沒(méi)有專有運(yùn)行團(tuán)隊(duì)支持運(yùn)行它的挑戰(zhàn)，因?yàn)樗心敲炊嗒?dú)立部件。同樣昂貴的還有：據(jù)MikeCafarella和DougCutting估計(jì)，一個(gè)支持此10億頁(yè)的索引需要價(jià)值約50萬(wàn)美元的硬件投入，每月運(yùn)行費(fèi)用還需要3萬(wàn)美元。不過(guò)，他們相信這是一個(gè)有價(jià)值的目標(biāo)，因?yàn)檫@會(huì)開(kāi)放并最終使搜索引擎算法普及化。Nutch項(xiàng)目開(kāi)始于2002年，一個(gè)可工作的抓取工具和搜索系統(tǒng)很快浮出水面。但他們意識(shí)到，他們的架構(gòu)將無(wú)法擴(kuò)展到擁有數(shù)十億網(wǎng)頁(yè)的網(wǎng)絡(luò)。在2003年發(fā)表的一篇描述Google分布式文件系統(tǒng)(簡(jiǎn)稱GFS)的論文為他們提供了及時(shí)的幫助，文中稱Google正在使用此文件系統(tǒng)。GFS或類似的東西，可以解決他們?cè)诰W(wǎng)絡(luò)抓取和索引過(guò)程中產(chǎn)生的大量的文件的存儲(chǔ)需求。具體而言，GFS會(huì)省掉管理所花的時(shí)間，如管理存儲(chǔ)節(jié)點(diǎn)。在2004年，他們開(kāi)始寫一個(gè)開(kāi)放源碼的應(yīng)用，即Nutch的分布式文件系統(tǒng)(NDFS)。2004年，Google發(fā)表了論文，向全世界介紹了MapReduce。2005年初，Nutch的開(kāi)發(fā)者在Nutch上有了一個(gè)可工作的MapReduce應(yīng)用，到當(dāng)年年中，所有主要的Nutch算法被移植到使用MapReduce和NDFS來(lái)運(yùn)行。Nutch中的NDFS和MapReduce實(shí)現(xiàn)的應(yīng)用遠(yuǎn)不只是搜索領(lǐng)域，在2006年2月，他們從Nutch轉(zhuǎn)移出來(lái)成為一個(gè)獨(dú)立的Lucene子項(xiàng)目，稱為Hadoop。大約在同一時(shí)間，DougCutting加入雅虎，Yahoo提供一個(gè)專門的團(tuán)隊(duì)和資源將Hadoop發(fā)展成一個(gè)可在網(wǎng)絡(luò)上運(yùn)行的系統(tǒng)(見(jiàn)后文的補(bǔ)充材料)。在2008年2月，雅虎宣布其搜索引擎產(chǎn)品部署在一個(gè)擁有1萬(wàn)個(gè)內(nèi)核的Hadoop集群上。2008年1月，Hadoop已成為Apache頂級(jí)項(xiàng)目，證明它是成功的，是一個(gè)多樣化、活躍的社區(qū)。通過(guò)這次機(jī)會(huì)，Hadoop成功地被雅虎之外的很多公司應(yīng)用，如Last.fm、Facebook和《紐約時(shí)報(bào)》。(一些應(yīng)用在第14章的案例研究和Hadoop維基有介紹，Hadoop維基的網(wǎng)址為。)有一個(gè)良好的宣傳范例，《紐約時(shí)報(bào)》使用亞馬遜的EC2云計(jì)算將4TB的報(bào)紙掃描文檔壓縮，轉(zhuǎn)換為用于Web的PDF文件。這個(gè)過(guò)程歷時(shí)不到24小時(shí)，使用100臺(tái)機(jī)器運(yùn)行，如果不結(jié)合亞馬遜的按小時(shí)付費(fèi)的模式(即允許《紐約時(shí)報(bào)》在很短的一段時(shí)間內(nèi)訪問(wèn)大量機(jī)器)和Hadoop易于使用的并行程序設(shè)計(jì)模型，該項(xiàng)目很可能不會(huì)這么快開(kāi)始啟動(dòng)。2008年4月，Hadoop打破世界紀(jì)錄，成為最快排序1TB數(shù)據(jù)的系統(tǒng)。運(yùn)行在一個(gè)910節(jié)點(diǎn)的群集，Hadoop在209秒內(nèi)排序了1TB的數(shù)據(jù)(還不到三分半鐘)，擊敗了前一年的297秒冠軍。同年11月，谷歌在報(bào)告中聲稱，它的MapReduce實(shí)現(xiàn)執(zhí)行1TB數(shù)據(jù)的排序只用了68秒。在2009年5月，有報(bào)道宣稱Yahoo的團(tuán)隊(duì)使用Hadoop對(duì)1TB的數(shù)據(jù)進(jìn)行排序只花了62秒時(shí)間。Hadoop@Yahoo！構(gòu)建互聯(lián)網(wǎng)規(guī)模的搜索引擎需要大量的數(shù)據(jù)，因此需要大量的機(jī)器來(lái)進(jìn)行處理。Yahoo！Search包括四個(gè)主要組成部分：Crawler，從因特網(wǎng)下載網(wǎng)頁(yè)；WebMap，構(gòu)建一個(gè)網(wǎng)絡(luò)地圖；Indexer，為最佳頁(yè)面構(gòu)建一個(gè)反向索引；Runtime(運(yùn)行時(shí))，回答用戶的查詢。WebMap是一幅圖，大約包括一萬(wàn)億條邊(每條代表一個(gè)網(wǎng)絡(luò)鏈接)和一千億個(gè)節(jié)點(diǎn)(每個(gè)節(jié)點(diǎn)代表不同的網(wǎng)址)。創(chuàng)建和分析此類大圖需要大量計(jì)算機(jī)運(yùn)行若干天。在2005年初，WebMap所用的基礎(chǔ)設(shè)施名為Dreadnaught，需要重新設(shè)計(jì)以適應(yīng)更多節(jié)點(diǎn)的需求。Dreadnaught成功地從20個(gè)節(jié)點(diǎn)擴(kuò)展到600個(gè)，但需要一個(gè)完全重新的設(shè)計(jì)，以進(jìn)一步擴(kuò)大。Dreadnaught與MapReduce有許多相似的地方，但靈活性更強(qiáng)，結(jié)構(gòu)更少。具體說(shuō)來(lái)，每一個(gè)分段(fragment)，Dreadnaught作業(yè)可以將輸出發(fā)送到此作業(yè)下一階段中的每一個(gè)分段，但排序是在庫(kù)函數(shù)中完成的。在實(shí)際情形中，大多數(shù)WebMap階段都是成對(duì)存在的，對(duì)應(yīng)于MapReduce。因此，WebMap應(yīng)用并不需要為了適應(yīng)MapReduce而進(jìn)行大量重構(gòu)。EricBaldeschwieler(Eric14)組建了一個(gè)小團(tuán)隊(duì)，我們開(kāi)始設(shè)計(jì)并原型化一個(gè)新的框架(原型為GFS和MapReduce，用C++語(yǔ)言編寫)，打算用它來(lái)替換Dreadnaught。盡管當(dāng)務(wù)之急是我們需要一個(gè)WebMap新框架，但顯然，標(biāo)準(zhǔn)化對(duì)于整個(gè)Yahoo!Search平臺(tái)至關(guān)重要，并且通過(guò)使這個(gè)框架泛化，足以支持其他用戶，我們才能夠充分運(yùn)用對(duì)整個(gè)平臺(tái)的投資。與此同時(shí)，我們?cè)陉P(guān)注Hadoop(當(dāng)時(shí)還是Nutch的一部分)及其進(jìn)展情況。2006年1月，雅虎聘請(qǐng)了DougCutting，一個(gè)月后，我們決定放棄我們的原型，轉(zhuǎn)而使用Hadoop。相較于我們的原型和設(shè)計(jì)，Hadoop的優(yōu)勢(shì)在于它已經(jīng)在20個(gè)節(jié)點(diǎn)上實(shí)際應(yīng)用過(guò)。這樣一來(lái)，我們便能在兩個(gè)月內(nèi)搭建一個(gè)研究集群，并著手幫助真正的客戶使用這個(gè)新的框架，速度比原來(lái)預(yù)計(jì)的快許多。另一個(gè)明顯的優(yōu)點(diǎn)是Hadoop已經(jīng)開(kāi)源，較容易(雖然遠(yuǎn)沒(méi)有那么容易！)從雅虎法務(wù)部門獲得許可在開(kāi)源方面進(jìn)行工作。因此，我們?cè)?006年初設(shè)立了一個(gè)200個(gè)節(jié)點(diǎn)的研究集群，我們將WebMap的計(jì)劃暫時(shí)擱置，轉(zhuǎn)而為研究用戶支持和發(fā)展Hadoop。Hadoop大事記2004年--最初的版本(現(xiàn)在稱為HDFS和MapReduce)由DougCutting和MikeCafarella開(kāi)始實(shí)施。2005年12月--Nutch移植到新的框架，Hadoop在20個(gè)節(jié)點(diǎn)上穩(wěn)定運(yùn)行。2006年1月--DougCutting加入雅虎。2006年2月--ApacheHadoop項(xiàng)目正式啟動(dòng)以支持MapReduce和HDFS的獨(dú)立發(fā)展。2006年2月--雅虎的網(wǎng)格計(jì)算團(tuán)隊(duì)采用Hadoop。2006年4月--標(biāo)準(zhǔn)排序(10GB每個(gè)節(jié)點(diǎn))在188個(gè)節(jié)點(diǎn)上運(yùn)行47.9個(gè)小時(shí)。2006年5月--雅虎建立了一個(gè)300個(gè)節(jié)點(diǎn)的Hadoop研究集群。2006年5月--標(biāo)準(zhǔn)排序在500個(gè)節(jié)點(diǎn)上運(yùn)行42個(gè)小時(shí)(硬件配置比4月的更好)。06年11月--研究集群增加到600個(gè)節(jié)點(diǎn)。06年12月--標(biāo)準(zhǔn)排序在20個(gè)節(jié)點(diǎn)上運(yùn)行1.8個(gè)小時(shí)，100個(gè)節(jié)點(diǎn)3.3小時(shí)，500個(gè)節(jié)點(diǎn)5.2小時(shí)，900個(gè)節(jié)點(diǎn)7.8個(gè)小時(shí)。07年1月--研究集群到達(dá)900個(gè)節(jié)點(diǎn)。07年4月--研究集群達(dá)到兩個(gè)1000個(gè)節(jié)點(diǎn)的集群。08年4月--贏得世界最快1TB數(shù)據(jù)排序在900個(gè)節(jié)點(diǎn)上用時(shí)209秒。08年10月--研究集群每天裝載10TB的數(shù)據(jù)。09年3月--17個(gè)集群總共24000臺(tái)機(jī)器。09年4月--贏得每分鐘排序，59秒內(nèi)排序500GB(在1400個(gè)節(jié)點(diǎn)上)和173分鐘內(nèi)排序100TB數(shù)據(jù)(在3400個(gè)節(jié)點(diǎn)上)。1.5

ApacheHadoop項(xiàng)目今天，Hadoop是一個(gè)分布式計(jì)算基礎(chǔ)架構(gòu)這把"大傘"下的相關(guān)子項(xiàng)目的集合。這些項(xiàng)目屬于Apache軟件基金會(huì)()，后者為開(kāi)源軟件項(xiàng)目社區(qū)提供支持。雖然Hadoop最出名的是MapReduce及其分布式文件系統(tǒng)(HDFS，從NDFS改名而來(lái))，但還有其他子項(xiàng)目提供配套服務(wù)，其他子項(xiàng)目提供補(bǔ)充性服務(wù)。這些子項(xiàng)目的簡(jiǎn)要描述如下，其技術(shù)棧如圖1-1所示。

圖1-1：Hadoop的子項(xiàng)目Core一系列分布式文件系統(tǒng)和通用I/O的組件和接口(序列化、JavaRPC和持久化數(shù)據(jù)結(jié)構(gòu))。Avro一種提供高效、跨語(yǔ)言RPC的數(shù)據(jù)序列系統(tǒng)，持久化數(shù)據(jù)存儲(chǔ)。(在本書寫作期間，Avro只是被當(dāng)作一個(gè)新的子項(xiàng)目創(chuàng)建，而且尚未有其他Hadoop子項(xiàng)目在使用它。)MapReduce分布式數(shù)據(jù)處理模式和執(zhí)行環(huán)境，運(yùn)行于大型商用機(jī)集群。HDFS分布式文件系統(tǒng)，運(yùn)行于大型商用機(jī)集群。Pig一種數(shù)據(jù)流語(yǔ)言和運(yùn)行環(huán)境，用以檢索非常大的數(shù)據(jù)集。Pig運(yùn)行在MapReduce和HDFS的集群上。Hbase一個(gè)分布式的、列存儲(chǔ)數(shù)據(jù)庫(kù)。HBase使用HDFS作為底層存儲(chǔ)，同時(shí)支持MapReduce的批量式計(jì)算和點(diǎn)查詢(隨機(jī)讀取)。ZooKeeper一個(gè)分布式的、高可用性的協(xié)調(diào)服務(wù)。ZooKeeper提供分布式鎖之類的基本服務(wù)用于構(gòu)建分布式應(yīng)用。

Hive分布式數(shù)據(jù)倉(cāng)庫(kù)。Hive管理HDFS中存儲(chǔ)的數(shù)據(jù)，并提供基于SQL的查詢語(yǔ)言(由運(yùn)行時(shí)引擎翻譯成MapReduce作業(yè))用以查詢數(shù)據(jù)。Chukwa分布式數(shù)據(jù)收集和分析系統(tǒng)。Chukwa運(yùn)行HDFS中存儲(chǔ)數(shù)據(jù)的收集器，它使用MapReduce來(lái)生成報(bào)告。(在寫作本書期間，Chukwa剛剛從Core中的"contrib"模塊分離出來(lái)獨(dú)立成為一個(gè)獨(dú)立的子項(xiàng)目。)第2章MapReduce簡(jiǎn)介MapReduce是一種用于數(shù)據(jù)處理的編程模型。該模型非常簡(jiǎn)單。同一個(gè)程序Hadoop可以運(yùn)行用各種語(yǔ)言編寫的MapReduce程序。在本章中，我們將看到用Java，Ruby，Python和C++這些不同語(yǔ)言編寫的不同版本。最重要的是，MapReduce程序本質(zhì)上是并行的，因此可以將大規(guī)模的數(shù)據(jù)分析交給任何一個(gè)擁有足夠多機(jī)器的運(yùn)營(yíng)商。MapReduce的優(yōu)勢(shì)在于處理大型數(shù)據(jù)集，所以下面首先來(lái)看一個(gè)例子。2.1

一個(gè)氣象數(shù)據(jù)集在我們這個(gè)例子里，要編寫一個(gè)挖掘氣象數(shù)據(jù)的程序。分布在全球各地的氣象傳感器每隔一小時(shí)便收集當(dāng)?shù)氐臍庀髷?shù)據(jù)，從而積累了大量的日志數(shù)據(jù)。它們是適合用MapReduce進(jìn)行分析的最佳候選，因?yàn)樗鼈兪前虢Y(jié)構(gòu)化且面向記錄的數(shù)據(jù)。數(shù)據(jù)的格式我們將使用NationalClimaticDataCenter(國(guó)家氣候數(shù)據(jù)中心，NCDC，網(wǎng)址為)提供的數(shù)據(jù)。數(shù)據(jù)是以面向行的ASCII格式存儲(chǔ)的，每一行便是一個(gè)記錄。該格式支持許多氣象元素，其中許多數(shù)據(jù)是可選的或長(zhǎng)度可變的。為簡(jiǎn)單起見(jiàn)，我們將重點(diǎn)討論基本元素(如氣溫)，這些數(shù)據(jù)是始終都有且有固定寬度的。例2-1顯示了一個(gè)簡(jiǎn)單的示例行，其中一些重要字段加粗顯示。該行已被分成多行以顯示出每個(gè)字段，在實(shí)際文件中，字段被整合成一行且沒(méi)有任何分隔符。例2-1：國(guó)家氣候數(shù)據(jù)中心數(shù)據(jù)記錄的格式0057

332130

USAF

weather

station

identifier

99999

WBAN

weather

station

identifier

19500101

observation

date

0300

observation

time

+51317

latitude

(degrees

1000)

+028783

longitude

(degrees

1000)

FM-12

+0171

elevation

(meters)

99999

V020

320

wind

direction

(degrees)

quality

code

0072

00450

sky

ceiling

height

(meters)

quality

code

010000

visibility

distance

(meters)

quality

code

-0128

air

temperature

(degrees

Celsius

10)

quality

code

-0139

dew

point

temperature

(degrees

Celsius

10)

quality

code

10268

atmospheric

pressure

(hectopascals

10)

quality

code

數(shù)據(jù)文件按照日期和氣象站進(jìn)行組織。從1901年到2001年，每一年都有一個(gè)目錄，每一個(gè)目錄都包含一個(gè)打包文件，文件中的每一個(gè)氣象站都帶有當(dāng)年的數(shù)據(jù)。例如，1990年的前面的數(shù)據(jù)項(xiàng)如下：%

raw/1990

head

010010-99999-1990.gz

010014-99999-1990.gz

010015-99999-1990.gz

010016-99999-1990.gz

010017-99999-1990.gz

010030-99999-1990.gz

010040-99999-1990.gz

010080-99999-1990.gz

010100-99999-1990.gz

010150-99999-1990.gz

因?yàn)閷?shí)際生活中有成千上萬(wàn)個(gè)氣象臺(tái)，所以整個(gè)數(shù)據(jù)集由大量較小的文件組成。通常情況下，我們更容易、更有效地處理數(shù)量少的大型文件，因此，數(shù)據(jù)會(huì)被預(yù)先處理而使每年記錄的讀數(shù)連接到一個(gè)單獨(dú)的文件中。(具體做法請(qǐng)參見(jiàn)附錄C)2.2

使用UnixTools來(lái)分析數(shù)據(jù)在全球氣溫?cái)?shù)據(jù)中每年記錄的最高氣溫是多少？我們先不用Hadoop來(lái)回答這一問(wèn)題，因?yàn)榇鸢钢行枰峁┮粋€(gè)性能標(biāo)準(zhǔn)(baseline)和一種檢查結(jié)果的有效工具。對(duì)于面向行的數(shù)據(jù)，傳統(tǒng)的處理工具是awk。例2-2是一個(gè)小的程序腳本，用于計(jì)算每年的最高氣溫。例2-2：一個(gè)用于從NCDC氣象記錄中找出每年最高氣溫的程序#!/usr/bin/env

bash

for

year

all/*

echo

-ne

'basename

$year

.gz'"\t"

gunzip

-c

$year

awk

temp

substr($0,

88,

substr($0,

93,

1);

(temp

!=9999

/[01459]/

temp

max)

max

temp

}

END

{

max

done

該腳本循環(huán)遍歷壓縮文件，首先顯示年份，然后使用awk處理每個(gè)文件。awk腳本從數(shù)據(jù)中提取兩個(gè)字段：氣溫和質(zhì)量代碼。氣溫值通過(guò)加上一個(gè)0變成一個(gè)整數(shù)。接下來(lái)，執(zhí)行測(cè)試，從而判斷氣溫值是否有效(值9999代表在NCDC數(shù)據(jù)集缺少值)，質(zhì)量代碼顯示的讀數(shù)是有疑問(wèn)還是根本就是錯(cuò)誤的。如果讀數(shù)是正確的，那么該值將與目前看到的最大值進(jìn)行比較，如果該值比原先的最大值大，就替換掉目前的最大值。當(dāng)文件中所有的行都已處理完并打印出最大值后，END塊中的代碼才會(huì)被執(zhí)行。下面是某次運(yùn)行結(jié)果的開(kāi)始部分：%./max_temperature.sh

1901

317

1902

244

1903

289

1904

256

1905

283

...

由于源文件中的氣溫值按比例增加到10倍，所以結(jié)果1901年的最高氣溫是31.7°C(在本世紀(jì)初幾乎沒(méi)有多少氣溫讀數(shù)會(huì)被記錄下來(lái)，所以這是可能的)。為完成對(duì)跨越一世紀(jì)這么長(zhǎng)時(shí)間的查找，程序在EC2High-CPUExtraLargeInstance機(jī)器上一共運(yùn)行了42分鐘。為加快處理，我們需要并行運(yùn)行部分程序。從理論上講，這很簡(jiǎn)單：我們可以通過(guò)使用計(jì)算機(jī)上所有可用的硬件線程來(lái)處理在不同線程中的各個(gè)年份的數(shù)據(jù)。但是這之中存在一些問(wèn)題。首先，劃分成大小相同的作業(yè)塊通常并不容易或明顯。在這種情況下，不同年份的文件，大小差異很大，所以一些線程會(huì)比其他線程更早完成。即使它們繼續(xù)下一步的工作，但是整個(gè)運(yùn)行中占主導(dǎo)地位的還是那些運(yùn)行時(shí)間很長(zhǎng)的文件。另一種方法是將輸入數(shù)據(jù)分成固定大小的塊，然后把每塊分配到各個(gè)進(jìn)程。其次，獨(dú)立線程運(yùn)行結(jié)果在合并后，可能還需要進(jìn)一步的處理。在這種情況下，每年的結(jié)果是獨(dú)立于其他年份，并可能通過(guò)連接所有結(jié)果和按年份排序這兩種方式來(lái)合并它們。如果使用固定大小的塊這種方法，則此類合并會(huì)更緊湊。對(duì)于這個(gè)例子，某年的數(shù)據(jù)通常被分割成幾個(gè)塊，每個(gè)進(jìn)行獨(dú)立處理。我們將最終獲得每個(gè)數(shù)據(jù)塊的最高氣溫，所以最后一步是尋找這些每年氣溫值中的最大值。最后，我們?nèi)匀皇芟抻谝慌_(tái)計(jì)算機(jī)的處理能力。如果手中所有的處理器都使用上都至少需要20分鐘，那就只能這樣了。我們不能使它更快。另外，一些數(shù)據(jù)集的增長(zhǎng)會(huì)超出一臺(tái)計(jì)算機(jī)的處理能力。當(dāng)我們開(kāi)始使用多臺(tái)計(jì)算機(jī)時(shí)，整個(gè)大環(huán)境中的許多其他因素將發(fā)揮作用，可能由于協(xié)調(diào)性和可靠性的問(wèn)題而出現(xiàn)當(dāng)機(jī)等錯(cuò)誤。誰(shuí)運(yùn)行整個(gè)作業(yè)？我們?nèi)绾翁幚硎〉倪M(jìn)程？因此，盡管并行處理可行，但實(shí)際上它非常復(fù)雜。使用Hadoop之類的框架非常有助于處理這些問(wèn)題。2.3

使用Hadoop進(jìn)行數(shù)據(jù)分析為了更好地發(fā)揮Hadoop提供的并行處理機(jī)制的優(yōu)勢(shì)，我們必須把查詢表示成MapReduce作業(yè)。經(jīng)過(guò)一些本地的小規(guī)模測(cè)試，我們將能夠在機(jī)器集群上運(yùn)行它。2.3.1

map和reduceMapReduce的工作過(guò)程分為兩個(gè)階段：map階段和reduce階段。每個(gè)階段都有鍵/值對(duì)作為輸入和輸出，并且它們的類型可由程序員選擇。程序員還具體定義了兩個(gè)函數(shù)：map函數(shù)和reduce函數(shù)。我們?cè)趍ap階段輸入的是原始的NCDC數(shù)據(jù)。我們選擇的是一種文本輸入格式，以便數(shù)據(jù)集的每一行都會(huì)是一個(gè)文本值。鍵是在文件開(kāi)頭部分文本行起始處的偏移量，但我們沒(méi)有這方面的需要，所以將其忽略。map函數(shù)很簡(jiǎn)單。我們使用map函數(shù)來(lái)找出年份和氣溫，因?yàn)槲覀冎粚?duì)它們有興趣。在本例中，map函數(shù)只是一個(gè)數(shù)據(jù)準(zhǔn)備階段，通過(guò)這種方式來(lái)建立數(shù)據(jù)，使得reducer函數(shù)能在此基礎(chǔ)上進(jìn)行工作：找出每年的最高氣溫。map函數(shù)也是很適合去除已損記錄的地方：在這里，我們將篩選掉缺失的、不可靠的或錯(cuò)誤的氣溫

數(shù)據(jù)。為了全面了解map的工作方式，我們思考下面幾行示例的輸入數(shù)據(jù)(考慮到頁(yè)面篇幅，一些未使用的列已被去除，用省略號(hào)表示)：這些行以鍵/值對(duì)的方式來(lái)表示map函數(shù)：鍵是文件中的行偏移量，而這往往是我們?cè)趍ap函數(shù)中所忽視的。map函數(shù)的功能僅僅提取年份和氣溫(以粗體顯示)，并將其作為輸出被發(fā)送。(氣溫值已被解釋為整數(shù))(1950,

(1950,

22)

(1950,

?11)

(1949,

111)

(1949,

78)

map函數(shù)的輸出先由MapReduce框架處理，然后再被發(fā)送到reduce函數(shù)。這一處理過(guò)程根據(jù)鍵來(lái)對(duì)鍵/值對(duì)進(jìn)行排序和分組。因此，繼續(xù)我們的示例，reduce函數(shù)會(huì)看到如下輸入：(1949,

[111,

78])

(1950,

[0,

22,

?11])

每年的年份后都有一系列氣溫讀數(shù)。所有reduce函數(shù)現(xiàn)在必須重復(fù)這個(gè)列表并從中找出最大的讀數(shù)：(1949,

111)

(1950,

22)

這是最后的輸出：全球氣溫記錄中每年的最高氣溫。

整個(gè)數(shù)據(jù)流如圖2-1所示。在圖的底部是Unix的管道，模擬整個(gè)MapReduce的流程，其中的內(nèi)容我們將在以后討論Hadoop數(shù)據(jù)流時(shí)再次提到。

（點(diǎn)擊查看大圖）圖2-1：MapReduce的邏輯數(shù)據(jù)流2.3.2

JavaMapReduce（1）在明白MapReduce程序的工作原理之后，下一步就是要用代碼來(lái)實(shí)現(xiàn)它。我們需要三樣?xùn)|西：一個(gè)map函數(shù)、一個(gè)reduce函數(shù)和一些來(lái)運(yùn)行作業(yè)的代碼。map函數(shù)是由一個(gè)Mapper接口來(lái)實(shí)現(xiàn)的，其中聲明了一個(gè)map()方法。例2-3顯示了我們的map函數(shù)的實(shí)現(xiàn)。例2-3：最高氣溫示例的Mapper接口import

java.io.IOException;

import

org.apache.hadoop.io.IntWritable;

import

org.apache.hadoop.io.LongWritable;

import

org.apache.hadoop.io.Text;

import

org.apache.hadoop.mapred.MapReduceBase;

import

org.apache.hadoop.mapred.Mapper;

import

org.apache.hadoop.mapred.OutputCollector;

import

org.apache.hadoop.mapred.Reporter;

public

class

MaxTemperatureMapper

extends

MapReduceBase

implements

Mapper<LongWritable,

Text,

IntWritable>

{

private

static

final

int

MISSING

9999;

public

void

map(LongWritable

key,

Text

value,

OutputCollector<Text,

IntWritable>

output,

Reporter

reporter)

throws

IOException

{

String

line

value.toString();

String

year

line.substring(15,

19);

int

airTemperature;

(line.charAt(87)

'+')

{

parseInt

doesn't

leading

plus

signs

airTemperature

Integer.parseInt(line.substring(88,

92));

}

else

{

airTemperature

Integer.parseInt(line.substring(87,

92));

}

String

quality

line.substring(92,

93);

(airTemperature

MISSING

quality.matches("[01459]"))

{

output.collect(new

Text(year),

new

IntWritable(airTemperature));

}

該Mapper接口是一個(gè)泛型類型，它有4個(gè)形式參數(shù)類型，由它們來(lái)指定map函數(shù)的輸入鍵、輸入值、輸出鍵和輸出值的類型。就目前的示例來(lái)說(shuō)，輸入鍵是一個(gè)長(zhǎng)整數(shù)偏移量，輸入的值是一行文本，輸出的鍵是年份，輸出的值是氣溫(整數(shù))。Hadoop規(guī)定了自己的一套可用于網(wǎng)絡(luò)序列優(yōu)化的基本類型，而不是使用內(nèi)置的Java類型。這些都可以在org.apache.hadoop.io包中找到?，F(xiàn)在我們使用的是LongWritable類型(相當(dāng)于Java的Long類型)、Text類型(相當(dāng)于Java的String類型)和IntWritable類型(相當(dāng)于Java的Integer類型)。map()方法需要傳入一個(gè)鍵和一個(gè)值。我們將一個(gè)包含Java字符串輸入行的Text值轉(zhuǎn)換成Java的String類型，然后利用其substring()方法提取我們感興趣的列。map()方法還提供了一個(gè)OutputCollector實(shí)例來(lái)寫入輸出內(nèi)容。在這種情況下，我們寫入年份作為一個(gè)Text對(duì)象(因?yàn)槲覀冎皇褂靡粋€(gè)鍵)，用IntWritable類型包裝氣溫值。我們只有在氣溫顯示出來(lái)后并且它的質(zhì)量代碼表示的是正確的氣溫讀數(shù)時(shí)才寫入輸出記錄。reduce函數(shù)同樣在使用Reducer時(shí)被定義，如例2-4所示。例2-4：最高氣溫示例的Reducerimport

java.io.IOException;

import

java.util.Iterator;

import

org.apache.hadoop.io.IntWritable;

import

org.apache.hadoop.io.Text;

import

org.apache.hadoop.mapred.MapReduceBase;

import

org.apache.hadoop.mapred.OutputCollector;

import

org.apache.hadoop.mapred.Reducer;

import

org.apache.hadoop.mapred.Reporter;

public

class

MaxTemperatureReducer

extends

MapReduceBase

implements

Reducer<Text,

IntWritable,

Text,

IntWritable>

{

public

void

reduce(Text

key,

Iterator<IntWritable>

values,

OutputCollector<Text,

IntWritable>

output,

Reporter

reporter)

throws

IOException

{

int

maxValue

Integer.MIN_VALUE;

while

(values.hasNext())

{

maxValue

Math.max(maxValue,

values.next().get());

}

output.collect(key,

new

IntWritable(maxValue));

}

同樣，四個(gè)形式參數(shù)類型用于指定reduce函數(shù)的輸入和輸出類型。reduce函數(shù)的輸入類型必須與map函數(shù)的輸出類型相匹配：Text類型和IntWritable類型。在這種情況下，reduce函數(shù)的輸出類型是Text和IntWritable這兩種類型，前者是年份的類型而后者是最高氣溫的類型，在這些輸入類型之中，我們遍歷所有氣溫，并把每個(gè)記錄進(jìn)行比較直到找到一個(gè)最高的為止。第三部分代碼運(yùn)行的是MapReduce作業(yè)(請(qǐng)參見(jiàn)例2-5)。2.3.2

JavaMapReduce（2）例2-5：在氣象數(shù)據(jù)集中找出最高氣溫的應(yīng)用程序import

java.io.IOException;

import

org.apache.hadoop.fs.Path;

import

org.apache.hadoop.io.IntWritable;

import

org.apache.hadoop.io.Text;

import

org.apache.hadoop.mapred.FileInputFormat;

import

org.apache.hadoop.mapred.FileOutputFormat;

import

org.apache.hadoop.mapred.JobClient;

import

org.apache.hadoop.mapred.JobConf;

public

class

MaxTemperature

{

public

static

void

main(String[]

args)

throws

IOException

{

(args.length

{

System.err.println("Usage:

MaxTemperature

<input

path>

<output

path>");

System.exit(-1);

}

JobConf

conf

new

JobConf(MaxTemperature.class);

conf.setJobName("Max

temperature");

FileInputFormat.addInputPath(conf,

new

Path(args[0]));

FileOutputFormat.setOutputPath(conf,

new

Path(args[1]));

conf.setMapperClass(MaxTemperatureMapper.class);

conf.setReducerClass(MaxTemperatureReducer.class);

conf.setOutputKeyClass(Text.class);

conf.setOutputValueClass(IntWritable.class);

JobClient.runJob(conf);

}

JobConf對(duì)象指定了作業(yè)的各種參數(shù)。它授予你對(duì)整個(gè)作業(yè)如何運(yùn)行的控制權(quán)。當(dāng)我們?cè)贖adoop集群上運(yùn)行這個(gè)作業(yè)時(shí)，我們把代碼打包成一個(gè)JAR文件(Hadoop會(huì)在集群分發(fā)這個(gè)包)。我們沒(méi)有明確指定JAR文件的名稱，而是在JobConf構(gòu)造函數(shù)中傳送一個(gè)類，Hadoop會(huì)找到這個(gè)包含此類的JAR文件。在創(chuàng)建JobConf對(duì)象后，我們將指定輸入和輸出的路徑。通過(guò)調(diào)用FileInputFormat內(nèi)的靜態(tài)方法addInputPath()來(lái)定義輸入的路徑，它可以是單個(gè)文件、目錄(本例中，輸入的內(nèi)容組成此目錄下所有文件)或文件模式的路徑。同時(shí)，addInputPath()可被調(diào)用多次從而實(shí)現(xiàn)使用多路徑輸入。輸出路徑(其中只有一個(gè))是在FileOutputFormat內(nèi)的靜態(tài)方法setOutputPath()來(lái)指定的。它指定了reduce函數(shù)輸出文件寫入的目錄。在運(yùn)行作業(yè)前該目錄不應(yīng)該存在，否則Hadoop會(huì)報(bào)錯(cuò)并且拒絕運(yùn)行任務(wù)。這種預(yù)防措施是為了防止數(shù)據(jù)丟失(一個(gè)長(zhǎng)時(shí)間的任務(wù)可能非常惱人地被另一個(gè)意外覆蓋)。接下來(lái)，通過(guò)setMapperClass()和setReducerClass()這兩個(gè)方法來(lái)指定要使用的map和reduce類型。setOutputKeyClass()和setOutputValueClass()方法控制map和reduce函數(shù)的輸出類型，正如本例所示，這兩個(gè)方法往往是相同的。如果它們不同，那么map的輸出類型可設(shè)置成使用setMapOutputKeyClass()和setMapOutputValue-Class()方法。輸入的類型通過(guò)輸入格式來(lái)控制，我們沒(méi)有設(shè)置，因?yàn)槲覀兪褂玫氖悄J(rèn)的TextInputFormat(文本輸入格式)。在設(shè)置了定義map和reduce函數(shù)的類之后，運(yùn)行作業(yè)的準(zhǔn)備工作就算完成了。JobClient內(nèi)的靜態(tài)方法runJob()會(huì)提交作業(yè)并等待它完成，把進(jìn)展情況寫入控制臺(tái)。運(yùn)行測(cè)試寫完MapReduce作業(yè)之后，拿一個(gè)小型的數(shù)據(jù)集進(jìn)行測(cè)試以排除與代碼直接有關(guān)的問(wèn)題，這是常規(guī)做法。首先，以獨(dú)立模式安裝Hadoop(詳細(xì)說(shuō)明請(qǐng)參見(jiàn)附錄A)。在這種模式下，Hadoop運(yùn)行中使用本地帶jobrunner(作業(yè)運(yùn)行程序)的文件系統(tǒng)。讓我們用前面討論過(guò)的五行代碼的例子來(lái)測(cè)試它(考慮到頁(yè)面，這里已經(jīng)對(duì)輸出稍做修改和重新排版)：%

export

HADOOP_CLASSPATH=build/classes

hadoop

MaxTemperature

input/ncdc/sample.txt

output

09/04/07

12:34:35

INFO

jvm.JvmMetrics:

Initializing

JVM

Metrics

with

processName=Job

Tracker,

sessionId=

09/04/07

12:34:35

WARN

mapred.JobClient:

Use

GenericOptionsParser

for

parsing

the

arguments.

Applications

should

implement

Tool

for

the

same.

09/04/07

12:34:35

WARN

mapred.JobClient:

job

jar

file

set.

User

classes

may

not

found.

See

JobConf(Class)

JobConf#setJar(String).

Format:

Total

input

paths

process

09/04/07

12:34:35

INFO

mapred.JobClient:

Running

job:

job_local_0001

Format:

Total

input

paths

process

09/04/07

12:34:35

INFO

mapred.MapTask:

numReduceTasks:

09/04/07

12:34:35

INFO

mapred.MapTask:

100

09/04/07

12:34:35

INFO

mapred.MapTask:

data

buffer

79691776/99614720

09/04/07

12:34:35

INFO

mapred.MapTask:

record

buffer

262144/327680

09/04/07

12:34:35

INFO

mapred.MapTask:

Starting

flush

map

output

09/04/07

12:34:36

INFO

mapred.MapTask:

Finished

spill

09/04/07

12:34:36

INFO

mapred.TaskRunner:

Task:attempt_local_0001_m_000000_0

done.

And

the

process

commiting

09/04/07

12:34:36

INFO

mapred.LocalJobRunner:

file:/Users/tom/workspace/htdg/input/n

cdc/sample.txt:0+529

09/04/07

12:34:36

INFO

mapred.TaskRunner:

Task

attempt_local_0001_m_000000_0'

done.

09/04/07

12:34:36

INFO

mapred.LocalJobRunner:

09/04/07

12:34:36

INFO

mapred.Merger:

Merging

sorted

segments

09/04/07

12:34:36

INFO

mapred.Merger:

Down

the

last

merge-pass,

with

segments

left

total

size:

bytes

09/04/07

12:34:36

INFO

mapred.LocalJobRunner:

09/04/07

12:34:36

INFO

mapred.TaskRunner:

Task:attempt_local_0001_r_000000_0

done

And

the

process

commiting

09/04/07

12:34:36

INFO

mapred.LocalJobRunner:

09/04/07

12:34:36

INFO

mapred.TaskRunner:

Task

attempt_local_0001_r_000000_0

allowed

commit

now

09/04/07

12:34:36

INFO

mapred.

FileOutputCommitter:

Saved

output

task

'attempt_local_0001_r_000000_0'

file:/

Users/tom/workspace/htdg/output

09/04/07

12:34:36

INFO

mapred.

LocalJobRunner:

reduce

09/04/07

12:34:36

INFO

mapred.TaskRunner:

Task

'attempt_local_0001_r_000000_0'

done.

09/04/07

12:34:36

INFO

mapred.JobClient:

map

100%

reduce

100%

09/04/07

12:34:36

INFO

mapred.JobClient:

Job

complete:

job_local_0001

09/04/07