目录
目录 I
初识Hadoop 1
1.1 数据！数据 1
1.2 数据的存储和分析 3
1.3 相较于其他系统 4
1.4 Hadoop发展简史 9
1.5 Apache Hadoop项目 12
MapReduce简介 15
2.1 一个气象数据集 15
2.2 使用Unix Tools来分析数据 17
2.3 使用Hadoop进行数据分析 19
2.4 分布化 30
2.5 Hadoop流 35
2.6 Hadoop管道 40
Hadoop分布式文件系统 44
3.1 HDFS的设计 44
3.2 HDFS的概念 45
3.3 命令行接口 48
3.4 Hadoop文件系统 50
3.5 Java接口 54
3.6 数据流 68
3.7 通过distcp进行并行复制 75
3.8 Hadoop归档文件 77
Hadoop的I/O 80
4.1 数据完整性 80
4.2 压缩 83
4.3 序列化 92
4.4 基于文件的数据结构 111
MapReduce应用开发 125
5.1 API的配置 126
5.2 配置开发环境 128
5.3 编写单元测试 134
5.4 本地运行测试数据 138
5.5 在集上运行 144
5.6 作业调优 159
5.7 MapReduce的工作流 162
MapReduce的工作原理 166
6.1 运行MapReduce作业 166
6.2 失败 172
6.3 作业的调度 174
6.4 shuffle和排序 175
6.6 任务的执行 181
MapReduce的类型与格式 188
7.1 MapReduce类型 188
7.3 输出格式 217
MapReduce 特性 227
8.1 计数器 227
8.2 排序 235
8.3 联接 252
8.4 次要数据的分布 258
8.5 MapReduce的类库 263
Hadoop集的安装 264
9.1 集说明 264
9.2 集的建立和安装 268
9.3 SSH配置 270
9.4 Hadoop配置 271
9.5 安装之后 286
9.6 Hadoop集基准测试 286
9.7 云计算中的Hadoop 290
Hadoop的管理 293
10.1 HDFS 293
10.2 监控 306
10.3 维护 313
Pig简介 321
11.1 安装和运行Pig 322
11.2 实例 325
11.3 与数据库比较 329
11.4 Pig Latin 330
11.5 用户定义函数 343
11.6 数据处理操作符 353
11.7 Pig实践提示与技巧 363
Hbase简介 366
12.1 HBase基础 366
12.2 概念 367
12.3 安装 371
12.4 客户端 374
12.5 示例 377
12.6 HBase与RDBMS的比较 385
12.7 实践 390
ZooKeeper简介 394
13.1 ZooKeeper的安装和运行 395
13.2 范例 396
13.3 ZooKeeper服务 405
13.4 使用ZooKeeper建立应用程序 417
13.5 工业界中的ZooKeeper 428
案例研究 431
14.1 Hadoop在的应用 431
14.2 Hadoop和Hive在Facebook的应用 441
14.3 Hadoop在Nutch搜索引擎 451
14.4 Hadoop用于Rackspace的日志处理 466
14.5 Cascading项目 474
14.6 Apache Hadoop的1 TB排序 488
Apache Hadoop的安装 491
Cloudera的Hadoop分发包 497
预备NCDC气象资料 502
第1章 初识Hadoop
古时候，人们用牛来拉重物，当一头牛拉不动一根圆木的时候，他们不曾想过培育个头更大的牛。同样，我们也不需要尝试更大的计算机，而是应该开发更多的计算系统。
--格蕾斯·霍珀
1.1 数据！数据
我们生活在数据时代！很难估计全球存储的电子数据总量是多少，但是据IDC估计2006年"数字全球"项目(digital universe)的数据总量为0.18 ZB，并且预测到2011年这个数字将达到1.8 ZB，为2006年的10倍 。1 ZB相当于10的21次方字节的数据，或者相当于1000 EB，1 000 000 PB，或者大家更熟悉的10亿TB的数据！这相当于世界上每个人一个磁盘驱动器的数量级。
这一数据洪流有许多来源。考虑下文：
纽约证券交易所每天产生1 TB的交易数据。
著名社交网站Facebook的主机存储着约100亿张照片，占据PB级存储空间。
，一个家谱网站，存储着2.5 PB数据。
互联网档案馆(The Internet Archive)存储着约2 PB数据，并以每月至少20 TB的速度增长。
瑞士日内瓦附近的大型强子对撞机每年产生约15 PB的数据。
此外还有大量数据。但是你可能会想它对自己有何影响。大部分数据被锁定在最大的网页内容里面(如搜索引擎)或者是金融和科学机构，对不对？是不是所谓的"大数据"的出现会影响到较小的组织或个人？
我认为是这样的。以照片为例，我妻子的祖父是一个狂热的摄影爱好者，并且他成人之后，几乎一直都在拍照片。他的所有照片(中等格式、幻灯片和35 mm胶片)，在扫描成高解析度照片时，占了大约10 GB的空间。相比之下，我家去年一年用数码相机拍摄的照片就占用了5 GB的空间。我家产生照片数据的速度是我妻子祖父的35倍！并且，随着拍摄更多的照片变得越来越容易，这个速度还在增加中。
更常见的情况是，个人数据的产生量正在快速地增长。微软研究院的MyLifeBits项目(/en-us/projects/mylifebits/)显示，在不久的将来，个人信息档案将可能成为普遍现象。MyLifeBits是这样的一个实验：一个人与外界的联系(电话、邮件和文件)被抓取和存储供以后访问。收集的数据包括每分钟拍摄的照片等，导致整个数
据量达到每月1 GB的大小。当存储成本下降到使其可以存储连续的音频和视频时，服务于未来MyLifeBits项目的数据量将是现在的许多倍。
个人数据的增长的确是大势所趋，但更重要的是，计算机所产生的数据可能比人所产生的数据更大。机器日志、RFID读取器、传感器网络、车载GPS和零售交易数据等，这些都会促使"数据之山越来越高"。
公开发布的数据量也在逐年增加。作为组织或企业，再也不能只管理自己的数据，未来的成功在很大程度上取决于它是否能从其他组织的数据中提取出价值。
这方面的先锋(如亚马逊网络服务器、或者)的公共数据集，它们的存在就在于促进"信息共享"，任何人都可以共享并自由(或以AWS平台的形式，或以适度的价格)下载和分析这些数据。不同来源的信息混合处理后会带来意外的效果和至今难以想像的应用。
以项目为例，这是一个研究Flickr网站上天体爱好者中新照片的项目。它分析每一张上传的照片，并确定它是天空的哪一部分，或者是否是有趣的天体，如恒星或者星系。虽然这只是一个带实验性质的新服务，但是它显示了数据(这里特指摄影照片)的可用性并且被用来进行某些活动(图像分析)，而这些活动很多时候并不是数据创建者预先能够想像到的。
有句话是这么说的："算法再好，通常也难敌更多的数据。"意思是说对于某些问题(譬如基于既往偏好生成的电影和音乐推荐)，不论你的算法有多么猛，它们总是会在更多的数据面前无能为力(更不用说没有优化过的算法了)。
现在，我们有一个好消息和一个坏消息。好消息是有海量数据！坏消息是我们正在为存储和分析这些数据而奋斗不息。
1.2 数据的存储和分析
问题很简单：多年来硬盘存储容量快速增加的同时，访问速度-- 数据从硬盘读取的速度-- 却未能与时俱进。1990年，一个普通的硬盘驱动器可存储1370 MB的数据并拥有4.4
MB/s的传输速度 ，所以，只需五分钟的时间就可以读取整个磁盘的数据。20年过去了，1
TB级别的磁盘驱动器是很正常的，但是数据传输的速度却在100 MB/s左右。所以它需要花两个半小时以上的时间读取整个驱动器的数据。
从一个驱动器上读取所有的数据需要很长的时间，写甚至更慢。一个很简单的减少读取时间的办法是同时从多个磁盘上读取数据。试想一下，我们拥有100个磁盘，每个存储百分之一的数据。如果它们并行运行，那么不到两分钟我们就可以读完所有的数据。
只使用一个磁盘的百分之一似乎很浪费。但是我们可以存储100个数据集，每个1 TB，并让它们共享磁盘的访问。我们可以想像，此类系统的用户会很高兴看到共享访问可以缩短分析时间，并且，从统计角度来看，他们的分析工作会分散到不同的时间点，所以互相之间不会有太多干扰。
尽管如此，现在更可行的是从多个磁盘并行读写数据。
第一个需要解决的问题是硬件故障。一旦开始使用多个硬件设施，其中一个会出故障的概率是非常高的。避免数据丢失的常见做法是复制：通过系统保存数据的冗余副本，在故障发生时，可以使用数据的另一份副本。这就是冗余磁盘阵列的工作方式。Hadoop的文件系统HDFS(Hadoop Distributed Filesystem)也是一个例子，虽然它采取的是另一种稍有不同的方法，详见后文描述。
第二个问题是大部分分析任务需要通过某种方式把数据合并起来，即从一个磁盘读取的数据可能需要和另外99个磁盘中读取的数据合并起来才能使用。各种不同的分布式系统能够组合多个来源的数据，但是如何保证正确性是一个非常难的挑战。MapReduce提供了一个编程模型，其抽象出上述磁盘读写的问题，将其转换为计算一个由成对键/值组成的数据集。这种模型的具体细节将在后面的章节讨论。但是目前讨论的重点是，这个计算由两部分组成：Map和Reduce。这两者的接口就是"整合"之地。就像HDFS一样，MapReduce是内建可靠性这个功能的。
简而言之，Hadoop提供了一个稳定的共享存储和分析系统。存储由HDFS实现，分析由MapReduce实现。纵然Hadoop还有其他功能，但这些功能是它的核心所在。
1.3 相较于其他系统
MapReduce似乎采用的是一种蛮力方法。即，针对每个查询，每一个数据集-- 至少是很大一部分-- 都会被处理。但这正是它的能力。MapReduce可以处理一批查询，并且它针对整个数据集处理即席查询并在合理时间内获得结果的能力也是具有突破性的。它改变了我们对数据的看法，并且解放了以前存储在磁带和磁盘上的数据。它赋予我们对数据进行创新的机会。那些以前需要很长时间才能获得答案的问题现在已经迎刃而解，但反过来，这又带来了新的问题和见解。
例如，Rackspace的邮件部门Mailtrust，用Hadoop处理邮件的日志。他们写的一个查询是到其用户的地理分布。他们是这样说的：
"随着我们的壮大，这些数据非常有用，我们每月运行一次MapReduce任务来帮助我们决定哪些Rackspace数据中心需要添加新的邮件服务器。"
通过将数百GB的数据整合，借助于分析工具，Rackspace的工程师得以了解这些数据，否则他们永远都不会了解，并且他们可以运用这些信息去改善他们为用户提供的服务。第14章将详细介绍Rackspace公司是如何运用Hadoop的。
1.3.1 关系型数据库管理系统
为什么我们不能使用数据库加上更多磁盘来做大规模的批量分析？为什么我们需要MapReduce？
这个问题的答案来自于磁盘驱动器的另一个发展趋势：寻址时间的提高速度远远慢于传输速率的提高速度。寻址就是将磁头移动到特定位置进行读写操作的工序。它的特点是磁盘操作有延迟，而传输速率对应于磁盘的带宽。
如果数据的访问模式受限于磁盘的寻址，势必会导致它花更长时间(相较于流)来读或写大部分数据。另一方面，在更新一小部分数据库记录的时候，传统的B树(关系型数据库中使用的一种数据结构，受限于执行查的速度)效果很好。但在更新大部分数据库数据的时候，B树的效率就没有MapReduce的效率高，因为它需要使用排序/合并来重建数据库。
在许多情况下，MapReduce能够被视为一种RDBMS(关系型数据库管理系统)的补充。(两个系统之间的差异见表1-1)。MapReduce很适合处理那些需要分析整个数据集的问题，以批处理的方式，尤其是Ad Hoc(自主或即时)分析。RDBMS适用于点查询和更新(其中，数据集已经被索引以提供低延迟的检索和短时间的少量数据更新。MapReduce适合数据被一次写入和多次读取的应用，而关系型数据库更适合持续更新的数据集。
表1-1：关系型数据库和MapReduce的比较
数据大小 GB
访问
更新
结构
集成度
伸缩性
交互型和批处理
多次读写
静态模式
高
非线性
传统关系型数据库
PB
批处理
一次写入多次读取
动态模式
低
线性
MapReduce
MapReduce和关系型数据库之间的另一个区别是它们操作的数据集中的结构化数据的数量。结构化数据是拥有准确定义的实体化数据，具有诸如XML文档或数据库表定义的格式，符合特定的预定义模式。这就是RDBMS包括的内容。另一方面，半结构化数据比较宽松，虽然可能有模式，但经常被忽略，所以它只能用作数据结构指南。例如，一张电子表格，其中的结构便是单元格组成的网格，尽管其本身可能保存任何形式的数据。非结构化数据没有什么特别的内部结构，例如纯文本或图像数据。MapReduce对于非结构化或半结构化数据非
常有效，因为它被设计为在处理时间内解释数据。换句话说：MapReduce输入的键和值并不是数据固有的属性，它们是由分析数据的人来选择的。
关系型数据往往是规范的，以保持其完整性和删除冗余。规范化为MapReduce带来问题，因为它使读取记录成为一个非本地操作，并且MapReduce的核心假设之一就是，它可以进行(高速)流的读写。
Web服务器日志是记录集的一个很好的非规范化例子(例如，客户端主机名每次都以全名来指定，即使同一客户端可能会出现很多次)，这也是MapReduce非常适合用于分析各种日志文件的原因之一。
MapReduce是一种线性的可伸缩的编程模型。程序员编写两个函数-- map函数和Reduce函数-- 每一个都定义一个键/值对集映射到另一个。这些函数无视数据的大小或者它们正在使用的集的特性，这样它们就可以原封不动地应用到小规模数据集或者大的数据集上。更重要的是，如果放入两倍的数据量，运行的时间会少于两倍。但是如果是两倍大小的集，一个任务任然只是和原来的一样快。这不是一般的SQL查询的效果。
随着时间的推移，关系型数据库和MapReduce之间的差异很可能变得模糊。关系型数据库都开始吸收MapReduce的一些思路(如ASTER DATA的和GreenPlum的数据库)，另一方面，基于MapReduce的高级查询语言(如Pig和Hive)使MapReduce的系统更接近传统的数据库编程人员。
1.3.2 网格计算
高性能计算(High Performance Computing，HPC)和网格计算社区多年来一直在做大规模的数据处理，它们使用的是消息传递接口(Message Passing Interface，MPI)这样的API。从广义上讲，高性能计算的方法是将作业分配给一个机器集，这些机器访问共享文件系统，由一个存储区域网络(Storage Area Network，SAN)进行管理。这非常适用于以主计算密集型为主的作业，但当节点需要访问的大数据量(数百GB的数据，这是MapReduce实际开始"发光"的起点)时，这会成为一个问题，因为网络带宽成为"瓶颈"，所以计算节点闲置下来了。
MapReduce尝试在计算节点本地存储数据，因此数据访问速度会因为它是本地数据而比较快。 这项"数据本地化"功能，成为MapReduce的核心功能并且也是它拥有良好性能的原因之一。意识到网络带宽在数据中心环境是最有价值的资源(到处复制数据会很容易的把网络带宽饱和)之后，MapReduce便通过显式网络拓扑结构不遗余力地加以保护。请注意，这种安排不会排除MapReduce中的高CPU使用分析。
MPI赋予程序员很大的控制，但也要求显式控制数据流机制，需要使用传统的C语言的功能模块完成(例如socket)，以及更高级的算法来进行分析。而MapReduce却是在更高层面上完成任务，即程序员从键/值对函数的角度来考虑，同时数据流是隐含的。
在一个大规模分布式计算平台上协调进程是一个很大的挑战。最困难的部分是恰当的处理失效与错误-- 在不知道一个远程进程是否已经失败的时候-- 仍然需要继续整个计算。MapReduce将程序员从必须考虑失败任务的情况中解放出来，它检测失败的map或者reduce任务，在健康的机器上重新安排任务。MapReduce能够做到这一点，因为它是一个无共享的架构，这意味着各个任务之间彼此并不依赖。(这里讲得稍微简单了一些，因为mapper的输出是反馈给reducer的，但这由MapReduce系统控制。在这种情况下，相对于返回失败的map，应该对返回reducer给予更多关注，因为它必须确保它可以检索到必要的map输出，如果不行，必须重新运行相关的map从而生成必要的这些输出。)因此，从程序员的角度来看，执行任务的顺序是无关紧要的。相比之下，MPI程序必须显式地管理自己的检查点和恢复机制，从而把更多控制权交给程序员，但这样会加大编程的难度。
MapReduce听起来似乎是一个相当严格的编程模型，而且在某种意义上看的确如此：我们被限定于键/值对的类型(它们按照指定的方式关联在一起)，mapper和reducer彼此间的协作有限，一个接一个地运行(mapper传输键/值对给reducer)。对此，一个很自然的问题是：你是否能用它做点儿有用或普通的事情？
答案是肯定的。MapReduce作为一个建立搜索索引产品系统，是由Google的工程师们开发出来的，因为他们发现自己一遍又一遍地解决相同的问题(MapReduce的灵感来自传统的函数式编程、分布式计算和数据库社区)，但它后来被应用于其他行业的其他许多应用。我们惊喜地看到许多算法的变体在MapReduce中得以表示，从图像图形分析，到基于图表的问题，再到机器学习算法 。它当然不能解决所有问题，但它是一个很普遍的数据处理工具。
第14章将介绍一些Hadoop应用范例。
1.3.3 志愿计算
人们第一次听说Hadoop和MapReduce的时候，经常会问："和SETI@home有什么区别？"SETI，全称为Search for Extra-Terrestrial Intelligence(搜寻外星人)，运行着一个称为SETI@home的项目()。在此项目中，志愿者把自己计算机CPU的空闲时间贡献出来分析无线天文望远镜的数据借此寻外星智慧生命信号。SETI@home是最有名的拥有许多志愿者的项目，其他的还有Great Internet Mersenne Prime Search(搜索大素数)与Folding@home项目(了解蛋白质构成及其与疾病之间的关系)。
志愿计算项目通过将他们试图解决的问题分为几个他们成为工作单元的块来工作，并将它们送到世界各地的电脑上进行分析。例如，SETI@home的工作单元大约是0.35 MB的无线电望远镜数据，并且一个典型的计算机需要数小时或数天来分析。完成分析后，结果发送回服务器，客户端获得的另一项工作单元。作为防止欺骗的预防措施，每个工作单元必须送到三台机器上并且需要有至少两个结果相同才会被接受。
虽然SETI@home在表面上可能类似于MapReduce(将问题分为独立的块，然后进行并行计算)，但差异还是显著的。SETI@home问题是CPU高度密集型的，使其适合运行于世界各地成千上万台计算机上，因为相对于其计算时间而言，传输工作单元的时间微不足道 。志愿者捐献的是CPU周期，而不是带宽。
MapReduce被设计为用来运行那些需要数分钟或数小时的作业，这些作业在一个聚集带宽很高的数据中心中可信任的专用硬件设备上运行。相比之下，SETI@home项目是在接入互联网的不可信的计算机上运行，这些计算机的网速不同，而且数据也不在本地。
1.4 Hadoop发展简史
Hadoop是Doug Cutting-- Apache Lucene创始人-- 开发的使用广泛的文本搜索库。Hadoop起源于Apache Nutch，后者是一个开源的网络搜索引擎，本身也是由Lucene项目的一部分。
Hadoop名字的起源
Hadoop这个名字不是一个缩写，它是一个虚构的名字。该项目的创建者，Doug Cutting如此解释Hadoop的得名："这个名字是我孩子给一头吃饱了的棕黄大象命名的。我的命名标准就是简短，容易发音和拼写，没有太多的意义，并且不会被用于别处。小孩子是这方面的高手。Googol就是由小孩命名的。"
Hadoop及其子项目和后继模块所使用的名字往往也与其功能不相关，经常用一头大象或其他动物主题(例如："Pig")。较小的各个组成部分给与更多描述性(因此也更俗)的名称。这是一个很好的原则，因为它意味着可以大致从其名字猜测其功能，例如，jobtracker 的任务就是跟踪MapReduce作业。
从头开始构建一个网络搜索引擎是一个雄心勃勃的目标，不只是要编写一个复杂的、能够抓取和索引网站的软件，还需要面临着没有专有运行团队支持运行它的挑战，因为它有那么多独立部件。同样昂贵的还有：据Mike Cafarella和Doug Cutting估计，一个支持此10亿页的索引需要价值约50万美元的硬件投入，每月运行费用还需要3万美元。 不过，他们相信这是一个有价值的目标，因为这会开放并最终使搜索引擎算法普及化。
Nutch项目开始于2002年，一个可工作的抓取工具和搜索系统很快浮出水面。但他们意识到，他们的架构将无法扩展到拥有数十亿网页的网络。在2003年发表的一篇描述Google分布式文件系统(简称GFS)的论文为他们提供了及时的帮助，文中称Google正在使用此文件系统。 GFS或类似的东西，可以解决他们在网络抓取和索引过程中产生的大量的文件的存储需求。具体而言，GFS会省掉管理所花的时间，如管理存储节点。在2004年，他们开始写一个开放源码的应用，即Nutch的分布式文件系统(NDFS)。
2004年，Google发表了论文，向全世界介绍了MapReduce。 2005年初，Nutch的开发者在Nutch上有了一个可工作的MapReduce应用，到当年年中，所有主要的Nutch算法被移植到使用MapReduce和NDFS来运行。
Nutch中的NDFS和MapReduce实现的应用远不只是搜索领域，在2006年2月，他们从Nutch转移出来成为一个独立的Lucene子项目，称为Hadoop。大约在同一时间，Doug
Cutting加入雅虎，Yahoo提供一个专门的团队和资源将Hadoop发展成一个可在网络上运行的系统(见后文的补充材料)。在2008年2月，雅虎宣布其搜索引擎产品部署在一个拥有1万个内核的Hadoop集上。
2008年1月，Hadoop已成为Apache顶级项目，证明它是成功的，是一个多样化、活跃的社区。通过这次机会，Hadoop成功地被雅虎之外的很多公司应用，如、Facebook和《纽约时报》。(一些应用在第14章的案例研究和Hadoop维基有介绍，Hadoop维基的网址为/hadoop/PoweredBy。)
有一个良好的宣传范例，《纽约时报》使用亚马逊的EC2云计算将4 TB的报纸扫描文档压缩，转换为用于Web的PDF文件。 这个过程历时不到24小时，使用100台机器运行，如果不结合亚马逊的按小时付费的模式(即允许《纽约时报》在很短的一段时间内访问大量机器)和Hadoop易于使用的并行程序设计模型，该项目很可能不会这么快开始启动。
2008年4月，Hadoop打破世界纪录，成为最快排序1TB数据的系统。运行在一个910节点的集，Hadoop在209秒内排序了1 TB的数据(还不到三分半钟)，击败了前一年的297秒冠军。同年11月，谷歌在报告中声称，它的MapReduce实现执行1TB数据的排序只用了68秒。 在2009年5月，有报道宣称Yahoo的团队使用Hadoop对1 TB的数据进行排序只花了62秒时间。
Hadoop@Yahoo！
构建互联网规模的搜索引擎需要大量的数据，因此需要大量的机器来进行处理。Yahoo！Search包括四个主要组成部分：Crawler，从因特网下载网页；WebMap，构建一个网络地图；Indexer，为最佳页面构建一个反向索引；Runtime(运行时)，回答用户的查询。WebMap是一幅图，大约包括一万亿条边(每条代表一个网络链接)和一千亿个节点(每个节点代表不同的网址)。创建和分析此类大图需要大量计算机运行若干天。在2005年初，WebMap所用的基础设施名为Dreadnaught，需要重新设计以适应更多节点的需求。Dreadnaught成功地从20个节点扩展到600个，但需要一个完全重新的设计，以进一步扩大。Dreadnaught与MapReduce有许多相似的地方，但灵活性更强，结构更少。具体说来，每一个分段(fragment)，Dreadnaught作业可以将输出发送到此作业下一阶段中的每一个分段，但排序是在库函数中完成的。在实际情形中，大多数WebMap阶段都是成对存在的，对应于MapReduce。因此，WebMap应用并不需要为了适应MapReduce而进行大量重构。
Eric Baldeschwieler(Eric14)组建了一个小团队，我们开始设计并原型化一个新的框架(原型为GFS和MapReduce，用C++语言编写)，打算用它来替换Dreadnaught。尽管当务之急是我们需要一个WebMap新框架，但显然，标准化对于整个Yahoo! Search平台至关重要，并且通过使这个框架泛化，足以支持其他用户，我们才能够充分运用对整个平台的投资。
与此同时，我们在关注Hadoop(当时还是Nutch的一部分)及其进展情况。2006年1月，雅虎聘请了Doug Cutting，一个月后，我们决定放弃我们的原型，转而使用Hadoop。相较于我们的原型和设计，Hadoop的优势在于它已经在20个节点上实际应用过。这样一来，我们便能在两个月内搭建一个研究集，并着手帮助真正的客户使用这个新的框架，速度比原来预计的快许多。另一个明显的优点是Hadoop已经开源，较容易(虽然远没有那么容易！)从雅虎法务部门获得许可在开源方面进行工作。因此，我们在2006年初设立了一个200个节点的研究集，我们将WebMap的计划暂时搁置，转而为研究用户支持和发展Hadoop。
Hadoop大事记
2004年-- 最初的版本(现在称为HDFS和MapReduce)由Doug Cutting和Mike Cafarella开始实施。
2005年12月-- Nutch移植到新的框架，Hadoop在20个节点上稳定运行。
2006年1月-- Doug Cutting加入雅虎。
2006年2月-- Apache Hadoop项目正式启动以支持MapReduce和HDFS的独立发展。
2006年2月-- 雅虎的网格计算团队采用Hadoop。
2006年4月-- 标准排序(10 GB每个节点)在188个节点上运行47.9个小时。
2006年5月-- 雅虎建立了一个300个节点的Hadoop研究集。
2006年5月-- 标准排序在500个节点上运行42个小时(硬件配置比4月的更好)。
06年11月-- 研究集增加到600个节点。
06年12月-- 标准排序在20个节点上运行1.8个小时，100个节点3.3小时，500个节点5.2小时，900个节点7.8个小时。
07年1月-- 研究集到达900个节点。
07年4月-- 研究集达到两个1000个节点的集。
08年4月-- 赢得世界最快1 TB数据排序在900个节点上用时209秒。
08年10月-- 研究集每天装载10 TB的数据。
09年3月-- 17个集总共24 000台机器。
09年4月-- 赢得每分钟排序，59秒内排序500 GB(在1400个节点上)和173分钟内排序100 TB数据(在3400个节点上)。
1.5 Apache Hadoop项目
今天，Hadoop是一个分布式计算基础架构这把"大伞"下的相关子项目的集合。这些项目属于Apache软件基金会()，后者为开源软件项目社区提供支持。虽然Hadoop最出名的是MapReduce及其分布式文件系统(HDFS，从NDFS改名而来)，但还有其他子项目提供配套服务，其他子项目提供补充性服务。这些子项目的简要描述如下，其技术栈如图1-1所示。
图1-1：Hadoop的子项目
Core
一系列分布式文件系统和通用I/O的组件和接口(序列化、Java RPC和持久化数据结构)。
Avro
一种提供高效、跨语言RPC的数据序列系统，持久化数据存储。(在本书写作期间，Avro只是被当作一个新的子项目创建，而且尚未有其他Hadoop子项目在使用它。)
MapReduce
分布式数据处理模式和执行环境，运行于大型商用机集。
HDFS
分布式文件系统，运行于大型商用机集。
Pig
一种数据流语言和运行环境，用以检索非常大的数据集。Pig运行在MapReduce和HDFS的集上。
Hbase
一个分布式的、列存储数据库。HBase使用HDFS作为底层存储，同时支持MapReduce的批量式计算和点查询(随机读取)。
ZooKeeper
一个分布式的、高可用性的协调服务。ZooKeeper提供分布式锁之类的基本服务用于构建分布式应用。
Hive
分布式数据仓库。Hive管理HDFS中存储的数据，并提供基于SQL的查询语言(由运行时引擎翻译成MapReduce作业)用以查询数据。
Chukwa
分布式数据收集和分析系统。Chukwa运行HDFS中存储数据的收集器，它使用MapReduce来生成报告。(在写作本书期间，Chukwa刚刚从Core中的"contrib"模块分离出来独立成为一个独立的子项目。)
第2章 MapReduce简介
MapReduce是一种用于数据处理的编程模型。该模型非常简单。同一个程序Hadoop可以运行用各种语言编写的MapReduce程序。在本章中，我们将看到用Java，Ruby，Python和C++这些不同语言编写的不同版本。最重要的是，MapReduce程序本质上是并行的，因此可以将大规模的数据分析交给任何一个拥有足够多机器的运营商。MapReduce的优势在于处理大型数据集，所以下面首先来看一个例子。
2.1 一个气象数据集
在我们这个例子里，要编写一个挖掘气象数据的程序。分布在全球各地的气象传感器每隔一小时便收集当地的气象数据，从而积累了大量的日志数据。它们是适合用MapReduce进行分析的最佳候选，因为它们是半结构化且面向记录的数据。
数据的格式
我们将使用National Climatic Data Center(国家气候数据中心，NCDC，网址为/)提供的数据。数据是以面向行的ASCII格式存储的，每一行便是一个记录。该格式支持许多气象元素，其中许多数据是可选的或长度可变的。为简单起见，我们将重点讨论基本元素(如气温)，这些数据是始终都有且有固定宽度的。
例2-1显示了一个简单的示例行，其中一些重要字段加粗显示。该行已被分成多行以显示出每个字段，在实际文件中，字段被整合成一行且没有任何分隔符。
例2-1：国家气候数据中心数据记录的格式
1. 0057
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
332130 # USAF weather station identifier
99999 # WBAN weather station identifier
19500101 # observation date
0300 # observation time
4
+51317 # latitude (degrees × 1000)
+028783 # longitude (degrees × 1000)
FM-12
+0171 # elevation (meters)
99999
V020
320 # wind direction (degrees)
1 # quality code
N
0072
1
00450 # sky ceiling height (meters)
1 # quality code
C
N
010000 # visibility distance (meters)
1 # quality code
N
9
-0128 # air temperature (degrees Celsius × 10)
1 # quality code
-0139 # dew point temperature (degrees Celsius × 10)
1 # quality code
10268 # atmospheric pressure (hectopascals × 10)
1 # quality code
数据文件按照日期和气象站进行组织。从1901年到2001 年，每一年都有一个目录，每一个目录都包含一个打包文件，文件中的每一个气象站都带有当年的数据。例如，1990年的前面的数据项如下：
1.
2.
3.
4.
% ls raw/1990 | head
5.
6.
7.
8.
9.
10.
11.
因为实际生活中有成千上万个气象台，所以整个数据集由大量较小的文件组成。通常情况下，我们更容易、更有效地处理数量少的大型文件，因此，数据会被预先处理而使每年记录的读数连接到一个单独的文件中。(具体做法请参见附录C)
2.2 使用Unix Tools来分析数据
在全球气温数据中每年记录的最高气温是多少？我们先不用Hadoop来回答这一问题，因为答案中需要提供一个性能标准(baseline)和一种检查结果的有效工具。
对于面向行的数据，传统的处理工具是awk。例2-2是一个小的程序脚本，用于计算每年的最高气温。
例2-2：一个用于从NCDC气象记录中出每年最高气温的程序
1.
2.
3.
4.
5.
6.
7.
8.
#!/usr/bin/env bash
for year in all/*
do
echo -ne 'basename $year .gz'"t"
gunzip -c $year |
awk '{ temp = substr($0, 88, 5) + 0;
q = substr($0, 93, 1);
if (temp !=9999 && q ~ /[01459]/
&& temp > max) max = temp }
9.
10.
END { print max }'
done
该脚本循环遍历压缩文件，首先显示年份，然后使用awk处理每个文件。awk脚本从数据中提取两个字段：气温和质量代码。气温值通过加上一个0变成一个整数。接下来，执行测试，从而判断气温值是否有效(值9999代表在NCDC数据集缺少值)，质量代码显示的读数是有疑问还是根本就是错误的。如果读数是正确的，那么该值将与目前看到的最大值进行比较，如果该值比原先的最大值大，就替换掉目前的最大值。当文件中所有的行都已处理完并打印出最大值后，END块中的代码才会被执行。
下面是某次运行结果的开始部分：
1.
2.
3.
4.
5.
6.
7.
%./max_
1901 317
1902 244
1903 289
1904 256
1905 283
...
由于源文件中的气温值按比例增加到10倍，所以结果1901年的最高气温是31.7°C(在本世纪初几乎没有多少气温读数会被记录下来，所以这是可能的)。为完成对跨越一世纪这么长时间的查，程序在EC2 High-CPU Extra Large Instance机器上一共运行了42分钟。
为加快处理，我们需要并行运行部分程序。从理论上讲，这很简单：我们可以通过使用计算机上所有可用的硬件线程来处理在不同线程中的各个年份的数据。但是这之中存在一些问题。
首先，划分成大小相同的作业块通常并不容易或明显。在这种情况下，不同年份的文件，大小差异很大，所以一些线程会比其他线程更早完成。即使它们继续下一步的工作，但是整个运行中占主导地位的还是那些运行时间很长的文件。另一种方法是将输入数据分成固定大小的块，然后把每块分配到各个进程。
其次，独立线程运行结果在合并后，可能还需要进一步的处理。在这种情况下，每年的结果是独立于其他年份，并可能通过连接所有结果和按年份排序这两种方式来合并它们。如果使用固定大小的块这种方法，则此类合并会更紧凑。对于这个例子，某年的数据通常被分割成几个块，每个进行独立处理。我们将最终获得每个数据块的最高气温，所以最后一步是寻这些每年气温值中的最大值。
最后，我们仍然受限于一台计算机的处理能力。如果手中所有的处理器都使用上都至少需要20分钟，那就只能这样了。我们不能使它更快。另外，一些数据集的增长会超出一台计算机的处理能力。当我们开始使用多台计算机时，整个大环境中的许多其他因素将发挥作用，可能由于协调性和可靠性的问题而出现当机等错误。谁运行整个作业？我们如何处理失败的进程？
因此，尽管并行处理可行，但实际上它非常复杂。使用Hadoop之类的框架非常有助于处理这些问题。
2.3 使用Hadoop进行数据分析
为了更好地发挥Hadoop提供的并行处理机制的优势，我们必须把查询表示成MapReduce作业。经过一些本地的小规模测试，我们将能够在机器集上运 行它。
2.3.1 map和reduce
MapReduce的工作过程分为两个阶段：map阶段和reduce阶段。每个阶段都有键/值对作为输入和输出，并且它们的类型可由程序员选择。程序员还具体定义了两个函数：map函数和reduce函数。
我们在map阶段输入的是原始的NCDC数据。我们选择的是一种文本输入格式，以便数据集的每一行都会是一个文本值。键是在文件开头部分文本行起始处的偏移量，但我们没有这方面的需要，所以将其忽略。
map函数很简单。我们使用map函数来出年份和气温，因为我们只对它们有兴趣。在本例中，map函数只是一个数据准备阶段，通过这种方式来建立数据，使得reducer函数能在此基础上进行工作：出每年的最高气温。map函数也是很适合去除已损记录的地方：在这里，我们将筛选掉缺失的、不可靠的或错误的气温 数据。
为了全面了解map的工作方式，我们思考下面几行示例的输入数据(考虑到页面篇幅，一些未使用的列已被去除，用省略号表示)：
1.
2.
3.
4.
5.
9999999N9+
9999999N9+
0500001N9+
0500001N9+
这些行以键/值对的方式来表示map函数：
1.
2.
(0, 9999999N9+)
(106, 9999999N9+)
3. (212, )
4. (318, 0500001N9+)
5. (424, 0500001N9+)
键是文件中的行偏移量，而这往往是我们在map函数中所忽视的。map函数的功能仅仅提取年份和气温(以粗体显示)，并将其作为输出被发送。(气温值已被解释为整数)
1.
2.
3.
(1950, 0)
(1950, 22)
(1950, ?11)
4.
5.
(1949, 111)
(1949, 78)
map函数的输出先由MapReduce框架处理，然后再被发送到reduce函数。这一处理过程根据键来对键/值对进行排序和分组。因此，继续我们的示例，reduce 函数会看到如下输入：
1.
2.
(1949, [111, 78])
(1950, [0, 22, ?11])
每年的年份后都有一系列气温读数。所有reduce函数现在必须重复这个列表并从中出最大的读数：
1.
2.
(1949, 111)
(1950, 22)
这是最后的输出：全球气温记录中每年的最高气温。
整个数据流如图2-1所示。在图的底部是Unix的管道，模拟整个MapReduce的流程，其中的内容我们将在以后讨论Hadoop 数据流时再次提到。
（点击查看大图）图2-1：MapReduce的逻辑数据流
2.3.2 Java MapReduce（1）
在明白MapReduce程序的工作原理之后，下一步就是要用代码来实现它。我们需要三样东西：一个map函数、一个reduce函数和一些来运行作业的代码。map函数是由一个Mapper接口来实现的，其中声明了一个map()方法。例2-3显示了我们的map函数的实现。
例2-3：最高气温示例的Mapper接口
1.
2.
3.
4.
import ption;
import table;
import itable;
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
import ;
import uceBase;
import ;
import Collector;
import er;
public class MaxTemperatureMapper extends MapReduceBase
implements Mapper
private static final int MISSING = 9999;
public void map(LongWritable key, Text value,
OutputCollector
throws IOException {
String line = ng();
String year = ing(15, 19);
int airTemperature;
if ((87) == '+') { // parseInt
doesn't like leading plus signs
24.
25.
26.
27.
28.
29.
30.
airTemperature = nt(ing(88, 92));
} else {
airTemperature = nt(ing(87, 92));
}
String quality = ing(92, 93);
if (airTemperature != MISSING && s("[01459]")) {
t(new Text(year), new IntWritable(airTemperature));
31.
32.
33.
}
}
}
该Mapper接口是一个泛型类型，它有4个形式参数类型，由它们来指定map函数的输入键、输入值、输出键和输出值的类型。就目前的示例来说，输入键是一个长整数偏移量，输入的值是一行文本，输出的键是年份，输出的值是气温(整数)。Hadoop规定了自己的一套可用于网络序列优化的基本类型，而不是使用内置的Java类型。这些都可以在包中到。现在我们使用的是LongWritable类型(相当于Java的Long类型)、Text类型(相当于Java的String类型)和IntWritable类型(相当于Java的Integer类型)。
map()方法需要传入一个键和一个值。我们将一个包含Java字符串输入行的Text值转换成Java的String类型，然后利用其substring()方法提取我们感兴趣的列。
map()方法还提供了一个OutputCollector实例来写入输出内容。在这种情况下，我们写入年份作为一个Text对象(因为我们只使用一个键)，用IntWritable类型包装气温值。我们只有在气温显示出来后并且它的质量代码表示的是正确的气温读数时才写入输出记录。
reduce函数同样在使用Reducer时被定义，如例2-4所示。
例2-4：最高气温示例的Reducer
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
import ption;
import or;
import table;
import ;
import uceBase;
import Collector;
import r;
import er;
public class MaxTemperatureReducer extends MapReduceBase
implements Reducer
public void reduce(Text key, Iterator
OutputCollector
16.
17.
18.
19.
20.
21.
22.
23.
24.
throws IOException {
int maxValue = _VALUE;
while (t()) {
maxValue = (maxValue, ().get());
}
t(key, new IntWritable(maxValue));
}
}
同样，四个形式参数类型用于指定reduce函数的输入和输出类型。reduce函数的输入类型必须与map函数的输出类型相匹配：Text类型和IntWritable类型。在这种情况下，reduce函数的输出类型是Text和IntWritable这两种类型，前者是年份的类型而后者是最高气温的
类型，在这些输入类型之中，我们遍历所有气温，并把每个记录进行比较直到到一个最高的为止。
第三部分代码运行的是MapReduce作业(请参见例2-5)。
2.3.2 Java MapReduce（2）
例2-5：在气象数据集中出最高气温的应用程序
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
13.
14.
15.
16.
17.
18.
19.
20.
21.
22.
23.
24.
25.
26.
27.
28.
29.
30.
31.
import ption;
import ;
import table;
import ;
import putFormat;
import tputFormat;
import ent;
import f;
public class MaxTemperature {
public static void main(String[] args) throws IOException {
if ( != 2) {
n("Usage: MaxTemperature
本文发布于:2024-09-23 13:18:49，感谢您对本站的认可！
本文链接：https://www.17tex.com/fanyi/16002.html
版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系，我们将在24小时内删除。
| 留言与评论（共有 0 条评论） |
豫公网安备41160202000603
站长QQ:729038198
关于我们
投诉建议