[转] 分布式计算开源框架Hadoop入门实践
Author :岑文初<?xml:namespace prefix = o ns = "urn:schemas-microsoft-com:office:office" />
Email: [email protected]c.com
msn: [email protected]
blog: http://blog.****.net/cenwenchu79/
引
在SIP项目设计的过程中，对于它庞大的日志在早先就考虑使用任务分解的多线程处理模式来分析统计，在前面有一篇Blog中提到了那部分的设计，但是由于统计的内容暂时还是十分简单，所以就采用Memcache作为计数器结合Mysql完成了访问控制以及统计的工作。但未来，对于海量日志分析的工作，还是需要有所准备。现在最火的技术词汇莫过于“云计算”，在Open API日益盛行的今天，互联网应用的数据将会越来越有价值，如何去分析这些数据，挖掘其内在价值，就需要分布式计算来支撑起海量数据的分析工作。
回过头来看，早先那种多线程，多任务分解的日志分析设计，其实是分布式计算的一个单机版缩略，如何将这种单机的工作分拆，变成集群工作协同，其实就是分布式计算框架设计所涉及的。在去年参加BEA的大会时候，BEA和VMWare合作采用虚拟机来构建集群，无非就是希望使得计算机硬件能够类似于应用程序中的资源池中的资源，使用者无需关心资源的分配情况，最大化了硬件资源的使用价值。分布式计算也是如此，具体的计算任务交由哪一台机器执行，执行后由谁来汇总，这都由分布式框架的Master来抉择，而使用者只需简单的将待分析内容的提供给分布式计算系统作为输入，就可以得到分布式计算后的结果。 Hadoop是Apache开源组织的一个分布式计算开源框架，在很多大型网站上都已经得到了应用，亚马逊，Facebook,Yahoo等等。对于我来说，最近的一个使用点就是服务集成平台的日志分析，服务集成平台的日志量将会很大，这也正好符合了分布式计算的适用场景（日志分析，索引建立就是两大应用场景）。
当前没有正式确定使用，所以也是自己业余摸索，后续所写的相关内容，都是一个新手的学习过程，难免会有一些错误，只是希望记录下来可以分享给更多志同道合的朋友。
What is Hadoop
搞什么东西之前，第一步是要知道What，然后是Why,最后才是How，但很多开发的朋友在做了多年项目以后，都习惯是先How，然后What，最后才是Why，这样只会变得浮躁，同时往往会将技术误用不适合的场景。
Hadoop框架中最核心设计就是：MapReduce和HDFS。MapReduce的思想是由Google的一篇论文所提及而被广为流传的，简单的一句话解释MapReduce就是任务的分解与结果的汇总。HDFS是Hadoop分布式文件系统的缩写，为分布式计算存储提供了底层支持。
MapReduce从它名字上来看就大致可以看出个缘由，两个动词Map,Reduce，Map（展开）就是将一个任务分解成为多个任务，Reduce就是将分解后多任务处理的结果汇总起来，得出最后的分析结果。这不是什么新思想，其实在前面提到了多线程，多任务的设计就可以找到这种思想的影子。不论是现实社会，还是在程序设计中，一项工作往往可以被拆分成为多个任务，任务之间的关系可以分为两种：一种是不相关的任务，可以并行执行；另一种是任务之间有相互的依赖，先后顺序不能够颠倒，这类任务是无法并行处理的。回到过去，大学老师上课时让大家去分析关键路径，无非就是找最省时的任务分解执行方式。在分布式系统中，机器集群就可以看作硬件资源池，将并行的任务拆分交由每一个空闲机器资源去处理，能够极大地提高计算效率，同时这种资源无关性，对于计算集群的扩展无疑提供了最好的设计保证。（其实我一直认为Hadoop的卡通图标不应该是一个小象，应该是蚂蚁，分布式计算就好比蚂蚁吃大象，廉价的机器群可以匹敌任何高性能的计算机，纵向扩展的曲线始终敌不过横向扩展的斜线）。任务分解处理以后，那就需要将处理以后的结果在汇总起来，这就是Reduce要做的工作。
<?xml:namespace prefix = v ns = "urn:schemas-microsoft-com:vml" /><shapetype id="_x0000_t75" coordsize="21600,21600" o:spt="75" o:preferrelative="t" path="[email protected]@[email protected]@[email protected]@[email protected]@5xe" filled="f" stroked="f"><stroke joinstyle="miter"></stroke><formulas><f eqn="if lineDrawn pixelLineWidth 0"></f><f eqn="sum @0 1 0"></f><f eqn="sum 0 0 @1"></f><f eqn="prod @2 1 2"></f><f eqn="prod @3 21600 pixelWidth"></f><f eqn="prod @3 21600 pixelHeight"></f><f eqn="sum @0 0 1"></f><f eqn="prod @6 1 2"></f><f eqn="prod @7 21600 pixelWidth"></f><f eqn="sum @8 21600 0"></f><f eqn="prod @7 21600 pixelHeight"></f><f eqn="sum @10 21600 0"></f></formulas><path o:extrusionok="f" gradientshapeok="t" o:connecttype="rect"></path><lock v:ext="edit" aspectratio="t"></lock></shapetype><shape id="_x0000_i1025" style="WIDTH: 359.25pt; HEIGHT: 202.5pt" type="#_x0000_t75"><imagedata src="file:///C:%5CDOCUME~1%5CWENCHU~1.CEN%5CLOCALS~1%5CTemp%5Cmsohtmlclip1%5C01%5Cclip_image001.emz" o:title=""></imagedata></shape>
图 1 MapReduce
上图就是MapReduce大致的结构图，在Map前还可能会对输入的数据有split的过程，保证任务并行效率，在Map之后还会有shuffle的过程，对于提高Reduce的效率以及减小数据传输的压力有很大的帮助。后面会具体提及这些部分的细节。
HDFS是分布式计算的存储基石，Hadoop的分布式文件系统和其他分布式文件系统有很多类似的特质。
分布式文件系统基本的几个特点：
1. 对于整个集群有单一的命名空间。
2. 数据一致性。适合一次写入多次读取的模型，客户端在文件没有被成功创建之前是无法看到文件存在。
3. 文件会被分割成多个文件块，每个文件块被分配存储到数据节点上，而且根据配置会有复制文件块来保证数据的安全性。
<shape id="_x0000_i1026" style="WIDTH: 309.75pt; HEIGHT: 219.75pt" type="#_x0000_t75"><imagedata src="file:///C:%5CDOCUME~1%5CWENCHU~1.CEN%5CLOCALS~1%5CTemp%5Cmsohtmlclip1%5C01%5Cclip_image003.emz" o:title=""><font color="#000000" size="3"></font></imagedata></shape>
图 2 HDFS
上图中展现了整个HDFS三个重要角色：NameNode,DataNode,Client。
NameNode可以看作是分布式文件系统中的管理者，主要负责管理文件系统的命名空间，集群配置信息，存储块的复制。NameNode会存储文件系统的Meta-data在内存中，这些信息主要包括了文件信息，每一个文件对应的文件块的信息，每一个文件块在DataNode的信息。
DataNode是文件存储的基本单元。它存储Block在本地文件系统中，保存了Block的Meta-data，同时周期性的发送所有存在的block的报告给NameNode。
Client就是需要获取分布式文件系统文件的应用程序。
这里通过三个操作来说明他们之间的交互关系。
文件写入：
1. Client向NameNode发起文件写入的请求。
2. NameNode根据文件大小和文件块配置情况，返回给Client它所管理部分DataNode的信息。
3. Client将文件划分为多个Block，根据DataNode的地址信息，按顺序写入到每一个DataNode块中。
文件读取：
1. Client向NameNode发起文件读取的请求。
2. NameNode返回文件存储的DataNode的信息。
3. Client读取文件信息。
文件Block复制：
1. NameNode发现部分文件的block不符合最小复制数或者部分DataNode失效。
2. 通知DataNode相互复制Block。
3. DataNode开始直接相互复制。
最后在说一下HDFS的几个设计特点：（对于框架设计值得借鉴）
1． Block的放置
默认不配置，一个Block会有三份备份。一份放在NameNode指定的DataNode，另一份放在与指定DataNode非同一Rack上的DataNode，最后一份放在与指定DataNode同一Rack上的DataNode上。备份无非就是为了数据安全，考虑同一Rack的失败情况以及不同Rack之间数据拷贝性能问题就采用这种配置方式。
2． 心跳检测DataNode的健康状况，如果发现问题就采取数据备份的方式来保证数据的安全性。
3． 数据复制。（DataNode失败的时候，需要平衡DataNode的存储利用率的时候，需要平衡DataNode数据交互压力的时候）
这里先说一下，使用HDFS的balancer命令，可以配置一个Threshold来平衡每一个DataNode磁盘利用率。例如设置了Threshold为10%，那么执行balancer命令的时候，首先统计所有DataNode的磁盘利用率的均值，然后判断如果某一个DataNode的磁盘利用率超过这个均值Threshold以上，那么将会把这个DataNode的block转移到磁盘利用率低的DataNode，这对于新节点的加入来说十分有用。
4． 数据交验。采用CRC32作数据交验。在文件Block写入的时候除了写入数据还会写入交验信息，在读取的时候需要交验后再读入。
5． NameNode是单点。如果失败的话，任务处理信息将会纪录在本地文件系统和远端的文件系统中。
6． 数据管道性的写入。
当客户端要写入文件到DataNode上，首先客户端读取一个Block然后写到第一个DataNode上，然后由第一个DataNode传递到备份的DataNode上，一直到所有需要写入这个Block的NataNode都成功写入，客户端才会继续开始写下一个Block。
7． 安全模式。
在分布式文件系统启动的时候，开始的时候会有安全模式，当分布式文件系统处于安全模式的情况下，文件系统中的内容不允许修改也不允许删除，直到安全模式结束。安全模式主要是为了系统启动的时候检查各个DataNode上数据块的有效性，同时根据策略必要的复制或者删除部分数据块。运行期通过命令也可以进入安全模式。在实践过程中，系统启动的时候去修改和删除文件也会有安全模式不允许修改的出错提示，只需要等待一会儿即可。
综合MapReduce和HDFS来看Hadoop的结构：
<shape id="_x0000_i1027" style="WIDTH: 415.5pt; HEIGHT: 207.75pt" type="#_x0000_t75"><imagedata src="file:///C:%5CDOCUME~1%5CWENCHU~1.CEN%5CLOCALS~1%5CTemp%5Cmsohtmlclip1%5C01%5Cclip_image005.emz" o:title=""><font color="#000000" size="3"></font></imagedata></shape>
图 3 Hadoop
在Hadoop的系统中，会有一台Master，主要负责NameNode的工作以及JobTracker的工作。JobTracker是的主要职责就是启动，跟踪，调度各个Slave的任务执行。还会有多台Slave，每一台Slave通常具有DataNode的功能以及TaskTracker的工作。TaskTracker根据应用要求来结合本地数据执行Map任务以及Reduce任务。
说到这里，就要提到分布式计算的最重要的一个设计点：Moving Computation is Cheaper than Moving Data。就是在分布式处理中，移动数据的代价总是高于转移计算的代价。简单来说就是分而治之的工作，需要将数据也分而存储，本地任务处理本地数据然后归总，这样才会保证分布式计算的高效性。
Why is hadoop
说完了What，简单的说一下Why。官方网站已经给了很多的说明，这里就大致说一下其优点及使用的场景（没有不好的工具，只用不适用的工具，因此选择好场景才能够真正发挥分布式计算的作用）
1． 可扩展。不论是存储的可扩展还是计算的可扩展都是Hadoop的设计根本。
2． 经济。框架可以运行在任何普通的PC上。
3． 可靠。分布式文件系统的备份恢复机制以及MapReduce的任务监控保证了分布式处理的可靠性。
4． 高效。分布式文件系统的高效数据交互实现以及MapReduce结合Local Data处理的模式，为高效处理海量的信息作了基础准备。
使用场景：个人觉得最适合的就是海量数据的分析，其实Google最早提出MapReduce也就是为了海量数据分析。同时HDFS最早是为了搜索引擎实现而开发的，后来才被用于分布式计算框架中。
海量数据被分割于多个节点，然后由每一个节点并行计算，将得出结果归并到输出。同时第一阶段的输出又可以作为下一阶段计算的输入，因此可以想象到一个树状结构的分布式计算图，在不同阶段都有不同产出，同时并行和串行结合的计算也可以很好的在分布式集群的资源下得以高效的处理。
How to Use Hadoop & Tips
其实参看Hadoop官方文档已经能够很容易配置分布式框架运行环境了，不过这里既然写了就再多写一点，同时有一些细节需要注意的也说一下，其实也就是这些细节会让人摸索半天。
Hadoop可以单机跑，也可以配置集群跑，单机跑就不需要多说了，只需要按照Demo的运行说明直接执行命令即可。这里主要重点说一下集群配置运行的过程。
环境：
7台普通的机器，操作系统都是linux。内存和CPU就不说了，反正Hadoop一大特点就是机器在多不在精。JDK必须是1.5以上的，这个切记。7台机器的机器名务必不同，后续会谈到机器名对于MapReduce有很大的影响。
部署考虑：
正如上面我描述的，对于Hadoop的集群来说，可以分成两大类角色，Master和Slave，前者主要配置NameNode和JobTracker的角色，负责总管分布式数据和分解任务的执行，后者配置DataNode和TaskTracker的角色，负责分布式数据存储以及任务的执行。本来打算一台机器是否可以配置成为Master同时也是Slave，不过发现在NameNode初始化的过程中以及TaskTracker执行过程中机器名配置好像有冲突（NameNode和TaskTracker对于Hosts的配置有些冲突，究竟是把机器名对应IP放在配置前面还是把Localhost对应IP放在前面有点问题，不过可能也是我自己的问题吧，这个大家可以根据实施情况给我反馈）。最后反正决定一台Master，六台Slave，后续复杂的应用开发和测试结果的比对会增加机器配置。
实施步骤：
1． 在所有的机器上都建立相同的目录，也可以就建立相同的用户，以该用户的home路径来做hadoop的安装路径。例如我在所有的机器上都建立了/home/wenchu。
2． 下载Hadoop，先解压到Master上。这里我是下载的<?xml:namespace prefix = st1 ns = "urn:schemas-microsoft-com:office:smarttags" /><chsdate isrocdate="False" islunardate="False" day="30" month="12" year="1899" w:st="on"><span lang="EN-US"><font size="3">0.17.1</font></span></chsdate>的版本。此时Hadoop的安装路径就是/home/wenchu/hadoop-0.17.1。
3． 解压后进入conf目录，主要需要修改以下文件：hadoop-env.sh，hadoop-site.xml,masters,slaves。
Hadoop的基础配置文件是hadoop-default.xml，看Hadoop的代码可以知道，默认建立一个Job的时候会建立Job的Config，Config首先读入hadoop-default.xml的配置，然后再读入hadoop-site.xml的配置（这个文件初始的时候配置为空），hadoop-site.xml中主要配置你需要覆盖的hadoop-default.xml的系统级配置，以及你需要在你的MapReduce过程中使用的自定义配置（具体的一些使用例如final等参考文档）。
以下是一个简单的hadoop-site.xml的配置：
<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!-- Put site-specific property overrides in this file. -->
<configuration>
<property>
<name>fs.default.name</name>//你的namenode的配置，机器名加端口
<value>hdfs://10.2.224.46:54310/</value>
</property>
<property>
<name>mapred.job.tracker</name>//你的JobTracker的配置，机器名加端口
<value>hdfs://10.2.224.46:54311/</value>
</property>
<property>
<name>dfs.replication</name>//数据需要备份的数量，默认是三
<value>1</value>
</property>
<property>
<name>hadoop.tmp.dir</name>//Hadoop的默认临时路径，这个最好配置，然后在新增节点或者其他情况下莫名其妙的DataNode启动不了，就删除此文件中的tmp目录即可。不过如果删除了NameNode机器的此目录，那么就需要重新执行NameNode格式化的命令了。
<value>/home/wenchu/hadoop/tmp/</value>
</property>
<property>
<name>mapred.child.java.opts</name>//java虚拟机的一些参数可以参照配置
<value>-Xmx<chmetcnv w:st="on" tcsc="0" numbertype="1" negative="False" hasspace="False" sourcevalue="512" unitname="m">512m</chmetcnv></value>
</property>
<property>
<name>dfs.block.size</name>//block的大小，单位字节，后面会提到用处，必须是512的倍数，因为采用crc作文件完整性交验，默认配置512是checksum的最小单元。
<value>5120000</value>
<description>The default block size for new files.</description>
</property>
</configuration>
hadoop-env.sh文件只需要修改一个参数：
# The java implementation to use. Required.
export JAVA_HOME=/usr/ali/jdk<chsdate isrocdate="False" islunardate="False" day="30" month="12" year="1899" w:st="on">1.5.0</chsdate>_10
配置你的Java路径，记住一定要1.5版本以上，免得莫名其妙出现问题。
Masters中配置Masters的ip或者机器名，如果是机器名那么需要在/etc/hosts中有所设置。
Slaves中配置的是Slaves的ip或者机器名，同样如果是机器名需要在/etc/hosts中有所设置。
范例如下：我这里配置的都是ip.
Masters:
10.2.224.46
Slaves:
10.2.226.40
10.2.226.39
10.2.226.38
10.2.226.37
10.2.226.41
10.2.224.36
4． 建立Master到每一台Slave的ssh受信证书。由于Master将会通过SSH启动所有的Slave的Hadoop，所以需要建立单向或者双向证书保证命令执行时不需要再输入密码。Master和所有的Slave机器上执行：ssh-****** -t rsa。执行此命令的时候，看到提示只需要回车。然后就会在