数据开发系列篇：大数据处理和编程实践Hadoop

Hadoop是Apache开源组织的一个分布式计算开源框架，在很多大型网站上都已经得到了应用，如亚马逊、Facebook和Yahoo等等。对于我来说，最近的一个使用点就是服务集成平台的日志分析。服务集成平台的日志量将会很大，而这也正好符合了分布式计算的适用场景(日志分析和索引建立就是两大应用场景)。

当前没有正式确定使用，所以也是自己业余摸索，后续所写的相关内容，都是一个新手的学习过程，难免会有一些错误，只是希望记录下来可以分享给更多志同道合的朋友。

什么是Hadoop

搞什么东西之前，第一步是要知道What(是什么)，然后是Why(为什么)，最后才是How(怎么做)。但很多开发的朋友在做了多年项目以后，都习惯是先How，然后What，最后才是Why，这样只会让自己变得浮躁，同时往往会将技术误用于不适合的场景。

Hadoop框架中最核心的设计就是：MapReduce和HDFS。MapReduce的思想是由Google的一篇论文所提及而被广为流传的，简单的一句话解释MapReduce就是“任务的分解与结果的汇总”。HDFS是Hadoop分布式文件系统(Hadoop Distributed File System)的缩写，为分布式计算存储提供了底层支持。

MapReduce从它名字上来看就大致可以看出个缘由，两个动词Map和Reduce，“Map(展开)”就是将一个任务分解成为多个任务，“Reduce”就是将分解后多任务处理的结果汇总起来，得出最后的分析结果。这不是什么新思想，其实在前面提到的多线程，多任务的设计就可以找到这种思想的影子。不论是现实社会，还是在程序设计中，一项工作往往可以被拆分成为多个任务，任务之间的关系可以分为两种：一种是不相关的任务，可以并行执行;另一种是任务之间有相互的依赖，先后顺序不能够颠倒，这类任务是无法并行处理的。回到大学时期，教授上课时让大家去分析关键路径，无非就是找最省时的任务分解执行方式。在分布式系统中，机器集群就可以看作硬件资源池，将并行的任务拆分，然后交由每一个空闲机器资源去处理，能够极大地提高计算效率，同时这种资源无关性，对于计算集群的扩展无疑提供了最好的设计保证。(其实我一直认为Hadoop的卡通图标不应该是一个小象，应该是蚂蚁，分布式计算就好比蚂蚁吃大象，廉价的机器群可以匹敌任何高性能的计算机，纵向扩展的曲线始终敌不过横向扩展的斜线)。任务分解处理以后，那就需要将处理以后的结果再汇总起来，这就是Reduce要做的工作。

图1：MapReduce结构示意图

上图就是MapReduce大致的结构图，在Map前还可能会对输入的数据有Split(分割)的过程，保证任务并行效率，在Map之后还会有Shuffle(混合)的过程，对于提高Reduce的效率以及减小数据传输的压力有很大的帮助。后面会具体提及这些部分的细节。

HDFS是分布式计算的存储基石，Hadoop的分布式文件系统和其他分布式文件系统有很多类似的特质。分布式文件系统基本的几个特点：

• 对于整个集群有单一的命名空间。
• 数据一致性。适合一次写入多次读取的模型，客户端在文件没有被成功创建之前无法看到文件存在。
• 文件会被分割成多个文件块，每个文件块被分配存储到数据节点上，而且根据配置会由复制文件块来保证数据的安全性。

图2：HDFS结构示意图

上图中展现了整个HDFS三个重要角色：NameNode、DataNode和Client。NameNode可以看作是分布式文件系统中的管理者，主要负责管理文件系统的命名空间、集群配置信息和存储块的复制等。NameNode会将文件系统的Meta-data存储在内存中，这些信息主要包括了文件信息、每一个文件对应的文件块的信息和每一个文件块在DataNode的信息等。DataNode是文件存储的基本单元，它将Block存储在本地文件系统中，保存了Block的Meta-data，同时周期性地将所有存在的Block信息发送给NameNode。Client就是需要获取分布式文件系统文件的应用程序。这里通过三个操作来说明他们之间的交互关系。

文件写入：

• Client向NameNode发起文件写入的请求。
• NameNode根据文件大小和文件块配置情况，返回给Client它所管理部分DataNode的信息。
• Client将文件划分为多个Block，根据DataNode的地址信息，按顺序写入到每一个DataNode块中。

文件读取：

• Client向NameNode发起文件读取的请求。
• NameNode返回文件存储的DataNode的信息。
• Client读取文件信息。

文件Block复制：

• NameNode发现部分文件的Block不符合最小复制数或者部分DataNode失效。
• 通知DataNode相互复制Block。
• DataNode开始直接相互复制。

最后再说一下HDFS的几个设计特点(对于框架设计值得借鉴)：

Block的放置：默认不配置。一个Block会有三份备份，一份放在NameNode指定的DataNode，另一份放在与指定DataNode非同一Rack上的DataNode，最后一份放在与指定DataNode同一Rack上的DataNode上。备份无非就是为了数据安全，考虑同一Rack的失败情况以及不同Rack之间数据拷贝性能问题就采用这种配置方式。

心跳检测DataNode的健康状况，如果发现问题就采取数据备份的方式来保证数据的安全性。

数据复制(场景为DataNode失败、需要平衡DataNode的存储利用率和需要平衡DataNode数据交互压力等情况)：这里先说一下，使用HDFS的balancer命令，可以配置一个Threshold来平衡每一个DataNode磁盘利用率。例如设置了Threshold为10%，那么执行balancer命令的时候，首先统计所有DataNode的磁盘利用率的均值，然后判断如果某一个DataNode的磁盘利用率超过这个均值Threshold以上，那么将会把这个DataNode的block转移到磁盘利用率低的DataNode，这对于新节点的加入来说十分有用。

数据交验：采用CRC32作数据交验。在文件Block写入的时候除了写入数据还会写入交验信息，在读取的时候需要交验后再读入。

NameNode是单点：如果失败的话，任务处理信息将会纪录在本地文件系统和远端的文件系统中。

数据管道性的写入：当客户端要写入文件到DataNode上，首先客户端读取一个Block然后写到第一个DataNode上，然后由第一个DataNode传递到备份的DataNode上，一直到所有需要写入这个Block的NataNode都成功写入，客户端才会继续开始写下一个Block。

安全模式：在分布式文件系统启动的时候，开始的时候会有安全模式，当分布式文件系统处于安全模式的情况下，文件系统中的内容不允许修改也不允许删除，直到安全模式结束。安全模式主要是为了系统启动的时候检查各个DataNode上数据块的有效性，同时根据策略必要的复制或者删除部分数据块。运行期通过命令也可以进入安全模式。在实践过程中，系统启动的时候去修改和删除文件也会有安全模式不允许修改的出错提示，只需要等待一会儿即可。

下面综合MapReduce和HDFS来看Hadoop的结构：

图3：Hadoop结构示意图

在Hadoop的系统中，会有一台Master，主要负责NameNode的工作以及JobTracker的工作。JobTracker的主要职责就是启动、跟踪和调度各个Slave的任务执行。还会有多台Slave，每一台Slave通常具有DataNode的功能并负责TaskTracker的工作。TaskTracker根据应用要求来结合本地数据执行Map任务以及Reduce任务。

说到这里，就要提到分布式计算最重要的一个设计点：Moving Computation is Cheaper than Moving Data。就是在分布式处理中，移动数据的代价总是高于转移计算的代价。简单来说就是分而治之的工作，需要将数据也分而存储，本地任务处理本地数据然后归总，这样才会保证分布式计算的高效性。

为什么要选择Hadoop

说完了What，简单地说一下Why。官方网站已经给了很多的说明，这里就大致说一下其优点及使用的场景(没有不好的工具，只用不适用的工具，因此选择好场景才能够真正发挥分布式计算的作用)：

可扩展：不论是存储的可扩展还是计算的可扩展都是Hadoop的设计根本。

经济：框架可以运行在任何普通的PC上。

可靠：分布式文件系统的备份恢复机制以及MapReduce的任务监控保证了分布式处理的可靠性。

高效：分布式文件系统的高效数据交互实现以及MapReduce结合Local Data处理的模式，为高效处理海量的信息作了基础准备。

使用场景：个人觉得最适合的就是海量数据的分析，其实Google最早提出MapReduce也就是为了海量数据分析。同时HDFS最早是为了搜索引擎实现而开发的，后来才被用于分布式计算框架中。海量数据被分割于多个节点，然后由每一个节点并行计算，将得出的结果归并到输出。同时第一阶段的输出又可以作为下一阶段计算的输入，因此可以想象到一个树状结构的分布式计算图，在不同阶段都有不同产出，同时并行和串行结合的计算也可以很好地在分布式集群的资源下得以高效的处理。

Hadoop的基本组成和生态系统

Hadoop系统运行于一个由普通商用服务器组成的计算集群上，该服务器集群在提供大规模分布式数据存储资源的同时，也提供大规模的并行化计算资源。

在大数据处理软件系统上，随着Apache Hadoop系统开源化的发展，在最初包含HDFS、MapReduce、HBase等基本子系统的基础上，至今Hadoop平台已经演进为一个包含很多相关子系统的完整的大数据处理生态系统。(图1-15展示了Hadoop平台的基本组成与生态系统)

1.MapReduce并行计算框架

MapReduce并行计算框架是一个并行化程序执行系统。它提供了一个包含Map和Reduce两阶段的并行处理模型和过程，提供一个并行化编程模型和接口，让程序员可以方便快速地编写出大数据并行处理程序。MapReduce以键值对数据输入方式来处理数据，并能自动完成数据的划分和调度管理。在程序执行时，MapReduce并行计算框架将负责调度和分配计算资源，划分和输入输出数据，调度程序的执行，监控程序的执行状态，并负责程序执行时各计算节点的同步以及中间结果的收集整理。MapReduce框架提供了一组完整的供程序员开发MapReduce应用程序的编程接口。

2.分布式文件系统HDFS

HDFS(Hadoop Distributed File System)是一个类似于Google GFS的开源的分布式文件系统。它提供了一个可扩展、高可靠、高可用的大规模数据分布式存储管理系统，基于物理上分布在各个数据存储节点的本地Linux 系统的文件系统，为上层应用程序提供了一个逻辑上成为整体的大规模数据存储文件系统。与GFS类似，HDFS采用多副本(默认为3个副本)数据冗余存储机制，并提供了有效的数据出错检测和数据恢复机制，大大提高了数据存储的可靠性。

3.分布式数据库管理系统HBase

为了克服HDFS难以管理结构化/半结构化海量数据的缺点，Hadoop提供了一个大规模分布式数据库管理和查询系统HBase。HBase是一个建立在HDFS之上的分布式数据库，它是一个分布式可扩展的NoSQL数据库，提供了对结构化、半结构化甚至非结构化大数据的实时读写和随机访问能力。 HBase提供了一个基于行、列和时间戳的三维数据管理模型，HBase中每张表的记录数(行数)可以多达几十亿条甚至更多，每条记录可以拥有多达上百万的字段。

4.公共服务模块Common

Common是一套为整个Hadoop系统提供底层支撑服务和常用工具的类库和API编程接口，这些底层服务包括Hadoop抽象文件系统 FileSystem、远程过程调用RPC、系统配置工具Configuration以及序列化机制。在0.20及以前的版本中，Common包含 HDFS、MapReduce和其他公共的项目内容;从0.21版本开始，HDFS和MapReduce被分离为独立的子项目，其余部分内容构成 Hadoop Common。

5.数据序列化系统Avro

Avro是一个数据序列化系统，用于将数据结构或数据对象转换成便于数据存储和网络传输的格式。Avro提供了丰富的数据结构类型，快速可压缩的二进制数据格式，存储持久性数据的文件集，远程调用RPC和简单动态语言集成等功能。

6.分布式协调服务框架Zookeeper

Zookeeper是一个分布式协调服务框架，主要用于解决分布式环境中的一致性问题。Zookeeper主要用于提供分布式应用中经常需要的系统可靠性维护、数据状态同步、统一命名服务、分布式应用配置项管理等功能。Zookeeper可用来在分布式环境下维护系统运行管理中的一些数据量不大的重要状态数据，并提供监测数据状态变化的机制，以此配合其他Hadoop子系统(如HBase、Hama等)或者用户开发的应用系统，解决分布式环境下系统可靠性管理和数据状态维护等问题。

7.分布式数据仓库处理工具Hive

Hive是一个建立在Hadoop之上的数据仓库，用于管理存储于HDFS或HBase中的结构化/半结构化数据。它最早由Facebook开发并用于处理并分析大量的用户及日志数据，2008年Facebook将其贡献给Apache成为Hadoop开源项目。为了便于熟悉SQL的传统数据库使用者使用Hadoop系统进行数据查询分析，Hive允许直接用类似SQL的HiveQL查询语言作为编程接口编写数据查询分析程序，并提供数据仓库所需要的数据抽取转换、存储管理和查询分析功能，而HiveQL语句在底层实现时被转换为相应的MapReduce程序加以执行。

8.数据流处理工具Pig

Pig是一个用来处理大规模数据集的平台，由Yahoo!贡献给Apache成为开源项目。它简化了使用Hadoop进行数据分析处理的难度，提供一个面向领域的高层抽象语言Pig Latin，通过该语言，程序员可以将复杂的数据分析任务实现为Pig操作上的数据流脚本，这些脚本最终执行时将被系统自动转换为MapReduce任务链，在Hadoop上加以执行。Yahoo!有大量的MapReduce作业是通过Pig实现的。

9.键值对数据库系统Cassandra

Cassandra是一套分布式的K-V型的数据库系统，最初由Facebook开发，用于存储邮箱等比较简单的格式化数据，后Facebook将 Cassandra贡献出来成为Hadoop开源项目。Cassandra以Amazon专有的完全分布式Dynamo为基础，结合了Google BigTable基于列族(Column Family)的数据模型，提供了一套高度可扩展、最终一致、分布式的结构化键值存储系统。它结合了Dynamo的分布技术和Google的 Bigtable数据模型，更好地满足了海量数据存储的需求。同时，Cassandra变更垂直扩展为水平扩展，相比其他典型的键值数据存储模型，Cassandra提供了更为丰富的功能。

10.日志数据处理系统Chukwa

Chukwa是一个由Yahoo!贡献的开源的数据收集系统，主要用于日志的收集和数据的监控，并与MapReduce协同处理数据。Chukwa 是一个基于Hadoop的大规模集群监控系统，继承了Hadoop系统的可靠性，具有良好的适应性和扩展性。它使用HDFS来存储数据，使用 MapReduce来处理数据，同时还提供灵活强大的辅助工具用以分析、显示、监视数据结果。

11.科学计算基础工具库Hama

Hama是一个基于BSP并行计算模型(Bulk Synchronous Parallel，大同步并行模型)的计算框架，主要提供一套支撑框架和工具，支持大规模科学计算或者具有复杂数据关联性的图计算。Hama类似 Google公司开发的Pregel，Google利用Pregel来实现图遍历(BFS)、最短路径(SSSP)、PageRank等计算。Hama可以与Hadoop的HDSF进行完美的整合，利用HDFS对需要运行的任务和数据进行持久化存储。由于BSP在并行化计算模型上的灵活性，Hama框架可在大规模科学计算和图计算方面得到较多应用，完成矩阵计算、排序计算、PageRank、BFS等不同的大数据计算和处理任务。

12.数据分析挖掘工具库Mahout

Mahout来源于Apache Lucene子项目，其主要目标是创建并提供经典的机器学习和数据挖掘并行化算法类库，以便减轻需要使用这些算法进行数据分析挖掘的程序员的编程负担，不需要自己再去实现这些算法。Mahout现在已经包含了聚类、分类、推荐引擎、频繁项集挖掘等广泛使用的机器学习和数据挖掘算法。此外，它还提供了包含数据输入输出工具，以及与其他数据存储管理系统进行数据集成的工具和构架。

13.关系数据交换工具Sqoop

Sqoop是SQL-to-Hadoop的缩写，是一个在关系数据库与Hadoop平台间进行快速批量数据交换的工具。它可以将一个关系数据库中的数据批量导入Hadoop的HDFS、HBase、Hive中，也可以反过来将Hadoop平台中的数据导入关系数据库中。Sqoop充分利用了 Hadoop MapReduce的并行化优点，整个数据交换过程基于MapReduce实现并行化的快速处理。

14.日志数据收集工具Flume

Flume是由Cloudera开发维护的一个分布式、高可靠、高可用、适合复杂环境下大规模日志数据采集的系统。它将数据从产生、传输、处理、输出的过程抽象为数据流，并允许在数据源中定义数据发送方，从而支持收集基于各种不同传输协议的数据，并提供对日志数据进行简单的数据过滤、格式转换等处理能力。输出时，Flume可支持将日志数据写往用户定制的输出目标。

像现在由于Hadoop系统和框架在不同大数据计算规模上支持能力不足，目前的spark系统也是比较引人关注，spark在广泛支持批处理、内存计算、流式计算、迭代计算、图数据计算等众多计算模式上比较有优势，目前像BAT这样的公司也在围绕Hadoop、spark、storm开发适合自己的平台。