毕业论文-基于Hadoop的大数据应用系统研发

HUNANUNIVERSITY毕业论文论文题目基于Hadoop的大数据应用系统研发学生姓名学生学号专业班级自动化1101学院名称电气与信息工程学院指导老师学院院长2015年 5月23日第一章绪论1.1课题背景及目的在计算机和互联网经过了几十年的快速发展，数字化已经开始在全球快速普及。文字，声音，图像全都转化为计算机可以存储和利用的数据，很难想像有什么不能转化为数据。特别是在移动互联网的爆炸性增长和物联网开始出现之时，数据以指数的形式增长，面对着如此海量的数据，过去的人们是难以想像的，而在今天，处理如此海量的数据也并非易事。面对海量数据，我们首先要考虑的便是数据的存储问题，分布式存储已经是一个很好的方案，其次便是数据的计算问题，计算必须能进行拆分，一个大的任务能由几个小的任务组合而成，而且不应该是串行组合而应该是并行组合，这样才能使得小任务能在不同计算机中并行完成。为了更加方便快捷的利用数据，最好便是能让计算实现本地化，最完美的情况下即让分布式网络中的计算机都能同时处理各自计算机中的数据，从而减少分布式网络集群中的带宽压力，而最终的结果由所有的计算机计算的结果组成。于是，云计算这个概念便脱颖而出，网络化快速发展已经为云计算提供了基础平台。Hadoop作为基础的云计算平台，在数据处理方面有很大的优势。首先，Hadoop是开源的，这允许我们根据自己的需要进行合适的扩展，Hadoop本身也提供了扩展的接口。其次，Hadoop的HDFS能很好地解决分布式数据存储的问题，有一系列容错和优化机制。再者，Hadoop提供的MapReduce编程模型简单又实用，有良好的可扩展性，最重要的是，MapReduce程序本质上是并行运行的。在数据之中，文本数据是不容忽视的重要组成部分之一，而以Hadoop为平台来处理海量数据有着巨大的优势。本设计基于这么原因，利用Hadoop来完成文本处理的目的。实现的主要功能：单词计数及单词以出现次数排序输出，Grep的实现，文本的倒排索引，单词共现。1.2国内外研究状况Hadoop起源于开源的网络搜索引擎ApacheNutch，以Google公司发表的GFS文件系统和MapReduce系统为蓝本构建了Hadoop的HNFS和MapReduce。2008年1月，Hadoop已经成为Apache的顶级项目，且已经在许多大公司得到了实际应用，例如Yahoo!和Facebook等。Hadoop开源项目自从推出以来，经历了数十个版本的演进，并逐步发展成熟，成为一个包含HDFS,MapReduce,HBase,Hive,Zookeeper,Avro,Pig,Sqoop等一系列相关子项目的大数据处理平台和生态系统。在Yahoo!的Hadoop研究集群中，Hadoop已经分别在188个节点，300个节点，500个节点，900个节点等进行扩展研究，且研究集群曾有过17个共24000个节点。在2009年4月，Yahoo!的Hadoop有每分钟排序中胜出，在59秒内GB（1400个节点）和173分钟内排序100TB数据（3400个节点）。对于节点数超出4000的大型集群，经典的MapReduce系统（Hadoop1.X版本）也开始面临着扩展性的瓶颈。在2012年，Yahoo!的一个团队开始设计下一代的MapReduce，即YARN（YetAnotherResourceNegotiator），并且演化为Hadoop2.X版本。作为新一代的Hadoop系统，YARN构架将主控节点的资源管理和作业管理功能实现分离设置，并且引入了全局资源管理器和针对每个作业的应用主控管理器，以此来减轻原主控节点的负担，并且YARN还可以基于Zookeeper实现资源管理的失效恢复，以此提高了Hadoop系统的高可用性。YARN还引入了资源容器的概念，将系统计算资源统一划分和封装为许多个资源单元，而不是像经典的MapReduce中那样区分Map和Reduce的计算资源，以此来提高计算资源的利用率。此外，YARN还能容纳MapReduce之外的其他并行计算模型和构架，以此来提高Hadoop构架的并行化编程的灵活性。YARN的设计使得不同的YARN应用可以在同一个集群上共存，大大提高了可管理性和集群的利用率。目前，Hadoop已经成为企业主流的部署系统，也是公认的大数据通用存储和分析平台，大量直接或间接使用Hadoop系统的产品也越来越多，许多公司也分布了自己Hadoop发行版本，例如EMC,IBM等。1.3设计和研究方法本文主要从平台搭建开始，逐步开始了解Hadoop，对HDFS和MapReduce工作机制进行深入了解。Hadoop是以Java进行开发的，所以本文的设计也是有Java语言进行开发，以文本处理为目的，对MapReduce的类型和格式进行进一步了解，从而对Hadoop的容错性和可扩展性有进一步的认识。本文中对Hadoop的一部分实现源代码进行分析，从而了解一些系统提供的接口的实现和自己对系统的扩展的实现过程。Hadoop和Java都提供了丰富的接口，所以在程序的实现过程中会大部分调用其提供的接口。1.4设计过程和研究内容本设计针对文本数据进行处理，Java为文本处理提供了大量的接口实现，例如在实现Grep程序过程中通过Java的Pattern类和Matcher进行正则表达示的创建和匹配操作。文本处理实现过程中，有过许多实现方式，例如，在倒排索引程序实现中，可以以不同的形式处理中间结果。Hadoop的版本有许多，本设计采用Hadoop1.2.1版本，以伪分布式方式，通过Cygwin模拟Linux环境在Windows7上搭建开发平台（Linux是Hadoop的产品开发和运行的平台，Win32平台是作为开发平台支持的，安装Cygwin是为了提供shell支持）。设计中的所有API均为新版API，MapReduce的执行机制为Mapreduce1（经典MapReduce）。在程序开发过程中，使用的Java版本为jdk1.8.0_25，开发环境使用Eclipse，安装HadoopEclipse插件（插件版本必须与Hadoop版本一致）后，使Hadoop的开发环境中嵌入Eclipse,实现开发环境的图形化，降低开发的复杂度。第二章Hadoop框架2.1Hadoop构造模块Hadoop实现了分布式存储和分布式计算的概念，而在一个集群上，“运行Hadoop”意味着在集群中的节点之上运行着一组守护进程来为分布式存储和分布式计算提供服务。这些守护过程有各自的特色和各自的任务，即NameNode，DataNode,SecondaryNameNode,JobTracker,TaskTracker。Hadoop在分布式计算和分布式存储中都采用主从结构，分布式存储系统被称为Hadoop文件系统。Hadoop有一个类似Linux虚拟文件系统的抽象文件系统概念并提供了很多接口来访问实际的文件系统，目前已经很多个具体实现，例如Hadoop的旗舰文件系统HDFS。HDFS有类似磁盘的块的概念，默认为64MB，HDFS上的文件以块大小划分，作为独立的存储单元。 HDFS集群有两类节点，即NameNode和DataNode，它们以管理者－工作者模式运行。NameNode是管理者，管理文件系统的命名空间，维护文件系统树及树中的文件和目录（这些信息以文件形式保存在本地磁盘中：命名空间镜像文件和编辑日志文件）。NameNode同时也记录每个文件中的各个数据块的节点信息，并保存在内存中，并会在系统启动时则数据节点创建。DataNode是文件系统的工作节点，存储着数据块信息，它需要定期向NameNode发送所存储的块的列表。在NameNode失效时，文件系统将无法使用，而SecondaryNameNode作为一个用于监测HDFS集群状态的辅助守护进程，其主要的用途就是用来保存NameNode中对HDFSmetadata的信息的备份，并减少namenode重启的时间（NameNode和SecondaryNameNode不应该处于同一节点），实现尽量防止数据丢失。（Hadoop的2.X发行版本中在HDFS中增加了高可用性的支持）Hadoop的分布式计算作业的控制主要由一个JobTracker和多个TaskTraker实现。JobTracker是应用程序和Hadoop之间的交互纽带，也是整个MapReduce计算框架中的主服务，负责整个作业的控制和资源的管理。JobTracker在启动之后，会一直监听并接收来自整个作业中的TaskTraker发送的“心跳”信息（节点资源使用情况和任务运行情况），根据这些信息，JobTraker实现作业控制和资源管理。TaskTraker运行在集群中各节点之上，接收并执行JobTracker的各种命令，如运行，提交，终止任务等，同时也需要将所处的节点上的各个任务状态信息周期性通过“心跳”信息的方式提交给JobTracker。2.2Hadoop平台搭建Hadoop有三种安装模式：单机模式，伪分布式模式，完全分布式模式。Hadoop的默认安装模式为单机模式（最小配置模式），可以通过配置文件（XML格式）来改变Hadoop的安装模式，并且可以通过配置文件实现Hadoop的性能调优。在本设计中，并没有采用完全分布式模式安装，而是将Hadoop配置为伪分布式工作模式来模拟分布式集群。在这种情况下，每一个Hadoop守护进程都作为一个独立的Java进程运行区别于单机模式下的无需运行任何守护进程，所有程序都在同一个JVM上运行。安装条件JDK(1.8.0_25),Cygwin，Hadoop1.2.1，Eclipse，Win32。安装JDK和Cygwin先安装JDK，配置JAVA_HOME环境变量，再在安装Cygwin时在SelectPackages界面里，Category展开net，选择如下openssh和openssl两项，如果要在Eclipe上编译Hadoop，需要安装Category为Base下的sed，安装完成后，将Cygwin安装文件夹中的bin目录加入到系统环境变量PATH中。安装sshd服务启动Cygwin，执行ssh-host-config-y命令，执行后，会提示输入密码，否则会退出该配置，此时输入密码和确认密码，回车。最后出现Hostconfigurationfinished.Havefun!表示安装成功。配置SSH免密码登录输入ssh-keygen-tdsa-P''-f~/.ssh/id_dsa(执行ssh-keygen命令生成密钥文件，-t表示指定生成的密钥类型(dsa,rsa），-P表示提供的密语,-f指定生成的密钥文件)。执行此命令后，在你的Cygwin\home\用户名路径下面会生成.ssh文件夹，可以通过命令ls-a/home/用户名查看，执行ssh-version命令ssh查看版本。执行完ssh-keygen命令后，再执行下面命令，就可以生成authorized_keys文件了。cd~/.ssh/cpid_dsa.pubauthorized_keys然后执行exit命令，退出Cygwin窗口。再次在桌面上双击Cygwin图标，打开Cygwin窗口，执行sshlocalhost命令，第一次执行该命令会有提示，输入yes后，回车即可。安装Hadoop把hadoop压缩包（本机选择Hadoop1.2.1版本）解压到/home/用户名目录下（在本设计中，用户名为lenovo），文件夹名称更改为hadoop（原解压后文件名过长，在shell环境下输入不太方便，仅仅是为方便起见，可以不更改）。配置伪分布式模式可以把伪分布模式看作是只有一个节点的集群，在这个集群中，这个节点既是Master，也是Slave，既是NameNode，也是DataNode，既是JobTracker，也是TaskTracker，这种模式下修改几个配置文件即可。配置hadoop-env.sh，以记事本打开该文件，设置JAVA_HOME的值为你的JDK安装路径，以本机配置为例：exportJAVA_HOME="H:\ProgramFiles\Java\jdk1.8.0_25"配置core-site.xml<configuration><property><name></name><value>hdfs://localhost:9000</value></property></configuration>配置hdfs-site.xml<configuration><property><name>dfs.replication</name><value>1</value></property></configuration>配置mapred-site.xml<configuration><property><name>mapred.job.tracker</name><value>localhost:9001</value></property></configuration>启动Hadoop打开Cgywin窗口，执行cd~/hadoop命令，进入hadoop文件夹，启动Hadoop前，需要先格式化Hadoop的文件系统HDFS，执行命令：bin/hadoopnamenode-format，输入命令bin/start-all.sh，启动所有进程。验证是否安装成功打开浏览器，分别输入下列网址，如果能够正常浏览，说明安装成功。http://localhost:50030，回车打开MapReduce的web页面。http://localhost:50070，回车打开HDFS的web页面。至此Hadoop环境已经配置完成，在Cygwin环境中也已经可以进行开发，为了能更加方便快捷地使用IDE开发Hadoop，选择利用Eclipse的Hadoop插件。在安装插件，配置Hadoop的相关信息之后，如果创建Hadoop程序，插件会自动导入Hadoop编程接口的JAR文件，这样用户就可以在Eclipse的图形化界面中编写、调试、运行Hadoop程序（包括单机程序和分布式程序），也可以在其中查看自己程序的实时状态、错误信息和运行结果，还可以查看、管理HDFS以及文件（上传本地文件和下载HDFS文件不再需要通过shell来进行，更加方便）。配置Eclipse开发环境首先在网上下载已经修改过Hadoop源码的hadoop-eclipse-1.2.1插件jar包（Windows文件权限问题会导致报错，Linux下不存在这个问题，也可以自行编译插件，但需要对hadoop\src\core\org\apache\hadoop\fs\下的FileUtil.java进行更改，注释掉checkReturnValue函数的中代码），放置在eclipse/plugins目录下，并重启eclipse，在eclipse的右上角应该出现了一只蓝色的大象logo，点击之后，在正下方的区域将会多出一项Map/ReduceLocations，点击NewHadoopLocation，需要填写以下内容：Locationname（当前创建的链接名字，可以任意指定）Map/ReduceMaster（执行MR的主机地址，本机host:localhost,port:9001）DFSMaster（DistributionFileSystem的主机地址，本机host:localhost,port:9000）（host和port与之前的配置文件相关）Username（链接至Hadoop的用户名，本机为lenovo）至此，Hadoop开发环境已经搭建完成。2.3HDFS对每一个系统来说，文件系统始终扮演着非常重要的角色。在Linux操作系统中，Linux可以支持许多种磁盘文件系统，原因就是Linux在实际文件系统之上又抽象出一层虚拟文件系统的概念，虚拟文件系统提供统一的接口，而实地的文件系统只需要实现各自的接口即可。Hadoop也有类似的一个综合性的文件系统抽象，而在众多的Hadoop文件系统中，HDFS是Hadoop的旗舰级文件系统，且被设计成与MapReduce结合使用时可以实现高性能。2.3.1HDFS的设计HDFS的设计理念是一个面向大规模数据密集型应用的、可伸缩的分布式文件系统。首先，HDFS面向的文件数据体积非常大，文件分布在整个集群的多台计算机中（也有可能扩展分布到多个集群中）。其次，HDFS假设数据失效为常态，例如应用程序或操作系统的bug，操作人员的失误，硬件的出错。HDFS存储超大文件是以流式数据访问的模式，即一次写入，多次读取是效率最高的访问模式。HDFS中的文件可能只有一个写入者，而且写操作只能只追加数据的方式添加在文件末尾，并不支持多个写入者或在任意位置进行修改。HDFS的设计主要是针对高数据吞吐量应用而进行优化，所以要求时间延迟低的数据访问方式的应用并不适合在HDFS上运行。Hadoop的设计之初便是面向商用硬件，因此，对于庞大的集群来说，节点故障的概率还是不容忽视，HDFS被设计成在遇到节点故障之时仍能继续运行肯不能让用户察觉到有明显的中断。还有一个重要的因素便是所存储的文件的大小，因为NameNode将文件系统的metadata存储在内存中，即在文件文件数量太多时，内存便成为存储数据的瓶颈。HDFS中的文件被划分为块，并以块为独立的存储单元（HDFS中小于块的文件不会占据整个块的空间）。块大小默认为64MB，主要是为了减小寻址开销，使用块而不是整个文件作为数据的基本存储单元在很大程度上简化了存储子系统的设计。在HDFS中，每个块的副本数默认为3个，这为HDFS提供了数据容错能力和提高了可用性。副本的存放是HDFS可靠性和提升性能的关键。HDFS采用机架感知策略来改进数据的可靠性，可用性和网络带宽的利用率，即一个副本存放在本地机架的节点了，另一个放在同一机架的另一个节点上，第三个放在不同机架的节点上。在读取数据，HDFS会尽量读取离客户端最近的副本，从而降低带宽延迟和节省网络带宽资源。另一方面，SecondaryNameNode机制也为NameNode宕机时的减少文件的损失作出替补作用（但这也必然会在一定程序上损失数据，SecondaryNameNode的同步备份总是有个滞后时间）。2.3.2HDFS的数据流Hadoop文件系统支持类似Linux文件系统的系统调用，客户端通过文件系统API实现对HDFS中的数据的读取和写入。客户端读取HDFS中的数据：图2.1客户端读取HDFS中的数据客户端将数据写入HDFS：图2.2客户端将数据写入HDFS2.3.3序列化与反序列化序列化指的将结构化对象转化为字节流以便在网络上传输或写入到磁盘的过程，反序列化为序列化的逆过程。序列化常用于分布式数据处理的进程间通信和永久存储中。在Hadoop中，集群中各个节点间进程间的通信是通过RPC实现的。RPC协议将消息序列化为二进制格式后发送至过程节点，远程节点反序列化后还原为原始数据。Hadoop使用自己的序列化格式Writable，Writable是Hadoop的核心，绝大多数的MapReuce程序的键和值使用Hadoop自带的Writable类型（已经经过很好的性能调优），但是，有时为了根据自己的需要，再好地调整结构，可以实现自己的Writable类型（通过Hadoop提供的接口实现）。在通用的几个Writable实现中，LongWritable对应long类型，Text类似于String，IntWritable对应于int型。2.4MapReduceMapReduce是Hadoop中的编程模型。MapReduce任务分为两个处理阶段：map阶段和reduce阶段。每个阶段都以键值对作为函数的输入和输出，类型可以自行选择。在通常的Hadoop程序开发中，我们往往只需要编写map函数和reduce函数即可，其他的可调用Hadoop提供的接口实现。2.4.1MapReduce工作机制在经典MapReduce中，一个作业的运行包含四个独立的实体。客户端：用于提交MapReduce作业。JobTracker：协调作业的运行。TaskTracker：运行作业划分之后任务。分布式文件系统：通常为HDFS，用于共享作业文件。整个MapRedece作业的执行过程可由以下几个部分组成：作业的提交过程向JobTracker请求新的作业ID，检查作业的输出说明，计算作业的输入分片，将运行作业所需要的资源（作业JAR文件，配置文件，输入分片）复制到以作业ID命名的目录下JobTracker的分布式文件系统中（通常为HDFS），通知JobTracker作业准备开始执行。作业的初始化初始化的主要工作是构造map任务和reduce任务并对它们进行初始化。JobTracker中的作业队列由作业调度器进行调度（一般为基于队列的FIFO调度器），作业调度器会从分布式文件系统中获取输入分片信息（每个分片对应一个Map任务，reduce任务的数量由配置文件中的参数决定，在伪分布式情况下，数量默认为1，reduce任务的数量也是性能调优的一个重要方面）。除此之外，一般还会创建两个任务：作业创建和作业清理，这两个任务在TaskTracker中执行。任务的分配TaskTracker通过循环方式定期向JobTracker发送“心跳”信息（节点资源使用情况和任务运行情况），根据“心跳”，JobTracker判断TaskTracker是否存活，并通过这个信息为它分配一个任务。任务的执行首先，需要从分布式文件系统中把作业JAR文件复制到TaskTracker所在的文件系统，实现JAR文件的本地化。然后，TaskTracker需要为任务新建一个本地工作目录，再把JAR文件解压到此文件夹中。最后，启动一个新的Java虚拟机来运行每个任务（一般情况下，可以在不同的任务之间重用JVM）。任务之间是并行执行的，而整个作业的执行必须依赖于各个并行执行的任务，只要其中有一个任务出错或者执行缓慢必将影响整个作业的执行进度，因此必须采取某种机制来处理这种情况的发生，Hadoop采用一种“推测执行”的机制来处理。在实际应用情况中，任务执行缓慢的原因可能有很多，例如硬件老化或者软件配置出错等等，想要检测出具体的原因十分困难，所以Hadoop采取的机制并不会尝试去诊断或者修复执行缓慢的任务，相反，在一个任务运行得较为缓慢的时候，它会尽量检测，同时会启动另一个相同的任务作为备份。进度和状态的更新在MapReuce作业运行期间，必须对任务的进度进行追踪，以确保该任务还在运行，如果出现意外已经停止运行，则必须重新启动一个新的任务。任务中有一组计数器，负责对任务运行过程中的各个阶段进行计数，以此为来构成进度。作业的完成JobTracker在接收到作业的最后一个任务已经完成的通知后，便把作业的状态设置为成功，作业的统计信息和计数值随后输出到控制台。最后是一些清理工作，如清空作业的工作状态，删除中间的输出等。MapReduce框架的输入和输出都是键值对，map任务的输出作为reduce任务的输入，但是map的输出并不是直接传递给reduce，而是经过按键排序，整个流程大致如下：图2.3MapReduce数据输入输出过程mapper输出先写入到内存缓冲区中（缓冲区默认为100MB），当缓冲区中的数据达到一定的数量时（默认为80%），后台线程便会把数据写到磁盘中。在写入到磁盘之前，数据将根据要被传送到的reducer划分成相对应的分区，在每个分区中，数据按照键来进行排序。如果有combiner时，它会在排序的输出后运行（combiner函数处理map函数的输出，产生的结果作为reduce函数的输入，目的是为了减少map函数输出的数据以减少写入到磁盘的数据同时也节约网络带宽资源，一般情况下combiner函数可以等同于reduce函数，也可以自定义，但要保证在有combiner的情况下，最终结果不会有不同），reducer通过HTTP协议得到输出文件的分区信息。在map阶段，对输入分片中的每条记录循环调用map函数。在reduce部分，reduce任务需要多个map任务的输出作为它的分区文件，它并不会等所有的map任务都完成，一旦有一个任务完成，它就开始复制其输出。分区的选择主要依靠mapper输出的键，可以自定义partitioner（默认的分区方式是哈希方式），以达到自己的要求。在reduce阶段，对已排序输出的每个键调用reduce函数。2.4.2MapReduce中的类型与格式关于输入输出，reduce函数的数据输入类型必须等于map函数的输出类型。一个map任务只处理一个输入分片，在默认情况下，输入分片的大小等同与文件块的大小，每个输入分片都被划分为若干条记录，每条记录对就一个键值对，map函数每次处理一条记录，直至输入分片处理完成。输入分片和记录仅仅是逻辑概念，在数据的存储上，还是以块的方式。对于记录中的键和值，不同的输入类型有不同的解释方式，对于默认的文本输入而言，每条记录为输入的一行，键是该在整个文件中的字节偏移量，值是该行的文本内容（不包括终止符）。输出类型与输入类型类似，可以自定义输入输出类型和键值。2.5Hadoop性能调优在整个Hadoop系统中，我们有很大的选择权，有很多的配置可以由用户自行设置，针对不同的情况选择合适的配置是十分关键的一步，也充分体现了Hadoop的适应性。Hadoop的性能与很多方面有关，不仅与自身参数配置有关，还涉及底层硬件，操作系统和Java虚拟机等，在这里的性能调优是针对Hadoop系统本身而言。在应用中，应该考虑实际情况，对参数进行配置，例如map任务缓冲区占内存的大小，缓冲区内存“溢出”比例多少，选择是否对map任务的输出进行压缩（如果采用压缩时，应该选择哪一种压缩策略），还有一些有关内存使用的参数和设置线程数目的参数，可以根据应用情况，动态调整。1.资源的规划Hadoop的计算资源以槽（slot）来表示，分为map槽和reduce槽，通过为TaskTracker配置槽数目，从而控制TaskTracker上并发执行的map任务和reduce任务数目。2.调整“心跳”配置“心跳”的间隔对性能有着不小的影响，时间间隔过小时，JobTracker需要处理高并发的“心跳”信息，给JobTracker造成不小的压力，间隔过大时，空闲的资源信息不能及时汇报给JobTracke，造成资源的浪费，同时也降低了系统整体的吞吐率，影响系统性能。在实际情况中，根据集群中节点数的不同，“心跳”的间隔有很大的差异，一般来说，小规模的集群间隔也设置的小。3.磁盘块配置这里的磁盘块并不是指文件系统中的块，而是本地磁盘，map任务的输出要写入到磁盘中，而对于输入输出密集型的任务来说，这会给本地磁盘造成很大的压力，而这里的配置指的就是在有多个磁盘存在的情况下，Hadoop将采用轮询的方式将map任务的输出结果写入到多个磁盘中，以负载均衡的方式提高系统性能。4.任务调度优化作为Hadoop的核心组件之一，调度器负责将系统中空闲的资源分配给各个任务，调度器的调度效率直接影响系统的吞吐率，每个作业都有分配有优先级，默认的调度器采用的是FIFO调度器（并不支持抢占），Hadoop中还有两个多用户调度器：公平调试器（支持抢占）和容量调试器。在只有一个作业运行的情况下，作业会得到整个集群的所有资源，在多个用户的情况下，每个用户有自己的作业池，因此作业多的用户也并不会获得更多的集群资源。为了将所有的空闲资源尽可能得分配给任务，三种调试器都支持批量任务调试功能（FIFO调试器只支持批量调试），而不是一次只分配一个，通过配置参数可以改变任务调试方式。第三章文本处理Hadoop非常适合于处理非结构化的文本数据，Hadoop本身提供了大量的文本处理的接口和实现，所以，在实现文本处理的过程中，大部分都调用接口实现，只有在没有提供合适的接口时，才自己实现。在Hadoop的编程模型中，一般来说，我们只需实现map函数和reduce函数，在新版API中，使用虚类来代替旧版中的接口，所以在实际的程序中，只需要继承Mapper类与Reducer类，在中实现map函数和reduce函数即可。3.1输入文件内容及输入输出文件路径文件输入路径：hdfs://localhost:9000/user/lenovo/input，其中input目录下有两个文件，分别为word1.txt，word2.txt，文件内容如下。word1.txt:ApacheHadoop

isan

open-source

softwareframework

writtenin

Java

for

distributedstorage

and

distributedprocessingofverylargedatasetson

computerclusters

builtfrom

commodityhardware.AllthemodulesinHadooparedesignedwithafundamentalassumptionthathardwarefailures(ofindividualmachines,orracksofmachines)arecommonplaceandthusshouldbeautomaticallyhandledinsoftwarebytheframework.ThecoreofApacheHadoopconsistsofastoragepart(HadoopDistributedFileSystem(HDFS))andaprocessingpart(MapReduce).Hadoopsplitsfilesintolargeblocksanddistributesthemamongstthenodesinthecluster.Toprocessthedata,HadoopMapReducetransfers

packagedcode

fornodestoprocessinparallel,basedonthedataeachnodeneedstoprocess.Thisapproachtakesadvantageofdatalocality[3]—nodesmanipulatingthedatathattheyhaveonhand—toallowthedatatobe

processed

fasterandmoreefficientlythanitwouldbeinamoreconventional

supercomputerarchitecturethatreliesona

parallelfilesystem

wherecomputationanddataareconnectedviahigh-speednetworking.[4]ThebaseApacheHadoopframeworkiscomposedofthefollowingmodules:HadoopCommon

–containslibrariesandutilitiesneededbyotherHadoopmodules;HadoopDistributedFileSystem(HDFS)

–adistributedfile-systemthatstoresdataoncommoditymachines,providingveryhighaggregatebandwidthacrossthecluster;HadoopYARN

–aresource-managementplatformresponsibleformanagingcomputingresourcesinclustersandusingthemforschedulingofusers'applications;[5][6]

andHadoopMapReduce

–aprogrammingmodelforlargescaledataprocessing.Theterm"Hadoop"hascometorefernotjusttothebasemodulesabove,butalsotothe"ecosystem",[7]

orcollectionofadditionalsoftwarepackagesthatcanbeinstalledontopoforalongsideHadoop,suchas

ApachePig,

ApacheHive,

ApacheHBase,

ApacheSpark,andothers.[8][9]ApacheHadoop'sMapReduceandHDFScomponentswereinspiredby

Google

papersontheir

MapReduce

and

GoogleFileSystem.[10]TheHadoopframeworkitselfismostlywrittenintheJavaprogramminglanguage,withsomenativecodein

C

andcommandlineutilitieswrittenas

Shellscript.Forend-users,thoughMapReduceJavacodeiscommon,anyprogramminglanguagecanbeusedwith"HadoopStreaming"toimplementthe"map"and"reduce"partsoftheuser'sprogram.[11]

Otherrelatedprojectsexposeotherhigher-leveluserinterfaces.ProminentcorporateusersofHadoopincludeFacebookandYahoo.Itcanbedeployedintraditionalon-sitedatacentersbuthasalsobeenimplementedinpubliccloudspacessuchas

MicrosoftAzure,

AmazonWebServices,

GoogleAppEngine

andIBM

Bluemix.word2.txt:Java

isageneral-purpose

computerprogramminglanguage

thatis

concurrent,

class-based,

object-oriented,[12]

andspecificallydesignedtohaveasfewimplementationdependenciesaspossible.Itisintendedtoletapplicationdevelopers"writeonce,runanywhere"(WORA),[13]

meaningthat

compiled

JavacodecanrunonallplatformsthatsupportJavawithouttheneedforrecompilation.[14]

Javaapplicationsaretypicallycompiledto

bytecode

thatcanrunonanyJavavirtualmachine

(JVM)regardlessof

computerarchitecture.Asof2015,Javaisoneofthemost

popularprogramminglanguagesinuse,[15][16][17][18]

particularlyforclient-serverwebapplications,withareported9milliondevelopers.[citationneeded]

Javawasoriginallydevelopedby

JamesGosling

at

SunMicrosystems

(whichhassincebeenacquiredbyOracleCorporation)andreleasedin1995asacorecomponentofSunMicrosystems'

Javaplatform.Thelanguagederivesmuchofits

syntax

from

C

and

C++,butithasfewer

low-level

facilitiesthaneitherofthem.Theoriginaland

referenceimplementation

Java

compilers,virtualmachines,and

classlibraries

wereoriginallyreleasedbySununderproprietarylicences.AsofMay2007,incompliancewiththespecificationsofthe

JavaCommunityProcess,SunrelicensedmostofitsJavatechnologiesunderthe

GNUGeneralPublicLicense.OthershavealsodevelopedalternativeimplementationsoftheseSuntechnologies,suchasthe

GNUCompilerforJava

(bytecodecompiler),

GNUClasspath

(standardlibraries),and

IcedTea-Web(browserpluginforapplets).文件的输出路径根据不同的程序不同，但它们在同一个目录下：hdfs://localhost:9000/user/lenovo/output.以单词共现为例，其输出文件的路径：hdfs://localhost:9000/user/lenovo/output/WordConcurrence.由于输出的文件内容比较，在这里只显现少量数据来说明结果。控制台输出显示了作业的完成过程，由于使用的输入文件相同，故此以其中的一个智育的控制台输出为例进行说明，其中略去时间戳信息。INFOinput.FileInputFormat:Totalinputpathstoprocess:2默认会读取输入路径input目录下的所有文件。INFOmapred.JobClient:Runningjob:job_local662749720_0001每个作业都会分配一个JobID来标识，JobClient为客户端与作业的通信中间件。INFOmapred.LocalJobRunner:WaitingformaptasksINFOmapred.LocalJobRunner:Startingtask:attempt_local662749720_0001_m_000000_0等待map任务进行，每个map任务都有一个ID号。INFOmapred.MapTask:Processingsplit:hdfs://localhost:9000/user/lenovo/input/word1.txt:0+2861输入分片信息，此为第一个分片，默认每个输入文件为一个分片，列出输入文件路径和偏移量信息。INFOmapred.MapTask:io.sort.mb=100INFOmapred.MapTask:databuffer=79691776/99614720INFOmapred.MapTask:recordbuffer=262144/327680默认的内存缓冲区大小为100MB，数据缓冲和记录缓冲的使用情况。INFOmapred.MapTask:Processingsplit:hdfs://localhost:9000/user/lenovo/input/word2.txt:0+1641INFOmapred.MapTask:io.sort.mb=100INFOmapred.MapTask:databuffer=79691776/99614720INFOmapred.MapTask:recordbuffer=262144/327680第二个map任务的输入信息情况。INFOmapred.JobClient:map100%reduce0%INFOmapred.Merger:Merging2sortedsegmentsmap任务完成，开始reduce任务之前，进行合并。INFOmapred.LocalJobRunner:reduce>reduceINFOmapred.Task:Task'attempt_local662749720_0001_r_000000_0'done.INFOmapred.JobClient:map100%reduce100%在有combiner的情况下，先进行combine，再进行reduce操作。INFOmapred.JobClient:Jobcomplete:job_local662749720_0001作业完成。随后会打印出许多信息，如输入输出分片，记录使用，堆栈使用等，可根据此调整系统性能。如果有多个MapReduce作业，会将以上的输出作为输入，重新将上述过程执行一遍。3.2单词计数一般情况下，只需要定义一个map过程和一个reduce过程即可，在单词计数中，输入类型为<LongWritable,Text>，中间输出类型是<Text,>，输出类型为<Text,>，在默认情况下，reducer的输出也会以键排序后输出，如果需要以<,Text>格式输出，则必须在reducer之后再加上一个mapper，使得键与值调换位置，即实现组合式MapReduce计算任务。Hadoop中，数据的输入和输出都需要指定文件路径，组合式计算任务中，可以通过临时文件的方式存储中间输出结果，在计算任务结束以后再删除临时文件即可。在具体的实现中，可以利用Hadoop中已经有的类实现。输出内容：1 "Hadoop1 "Hadoop"1 "ecosystem",[7]

or1 "map"1 "reduce"1 "write1 (HDFS))1 (HDFS)

–1 (Hadoop1 (MapReduce).1 (WORA),[13]

meaning1 (browser1 (of1 19951 2007,1 2015,1 91 All1 App3.3Grep实现Grep是一个功能强大的文本检索工具，它使用正则表达式 来搜索文本（正则表达式使用单个字符串来描述、匹配一系列符合某个句法规则的字符串），本设计的主要目标是实现一具简单的分布式Grep，对输入的文件进行逐行的正则匹配，如果匹配成功，则输出对应的内容并且输出该内容在整个文件中出现的次数。命令行参数输入依次为输入文件路径，输出文件路径，字符串，group值。本次实现中，无论输入的group值为多少都处理作0。利用Java进行开发的一大好处是它有大量的类实现，Java正则表达式通过java.util.regex包下的Pattern类与Matcher类实现。Pattern类用于创建正则表达式和实现简单的匹配操作，Matcher类实现分组支持和多次匹配支持。大致的构造构架如下所示：Patternp=Ppile(“\\d+”);Mattcherm=p.matcher(“22abc33”);在本设计中，关键要解决的问题是如何将输入的字符串参数和group值传递给map任务，本设计的实现是构造两个配置参数用来存储这两个输入参数，在mapper的初始化中（mapper中的setup函数，总是在map函数之前运行，且只在初始化时运行一次）提取这两个参数，以配置信息的方式实现通信。输入："Hadoop"输出：18 Hadoop输入："/^\s*$/"输出：输出为空，因为不存在空行输入："\d"输出：12 12 32 42 62 83 23 74 04 54 93.4文档倒排索引文档倒排索引应用十分广泛，几乎所有支持全文检索的搜索引擎都会应用到。倒排索引存储某个单词或词组在一个文档或一组文档中存储位置的映射，即可以通过文本的内容来查找对应的文档。在本设计的倒排索引附带词频属性，主要功能是在实现单词的倒排索引的同时，统计出单词在各自的文档中出现的次数。在实际应用中，索引的单词对象会除去一些没有必要实现的单词，如is,of之类，在实现中可以采用分布式缓存的机制，但是很遗憾，在本设计中未能实现。在应用中，为了减少map任务的输出数据传输，这里定义了combiner函数来对mapper端的数据进行归约。在具体分析如何实现时，可以采用多种方案，以下以键值对的形式说明对应的类设计的输出结果（输入都是默认的文本输入方式，键和值分别为该行在文件中的偏移量和该行内容）。方案一mapper输出： <word,doc> //LongWritable,Textcombiner输出： <word,doc#num> //Text,Textreducer输出： <word,doc1#num1,...> //Text,Text方案二mapper输出： <word#doc,1> //Text,IntWritablecombiner输出： <word#doc,num> //Text,IntWritablereducer输出： <word,doc1#num1,...> //Text,Text在本设计中，具体实现采用的是方案二，输出内容如下："ecosystem",[7]

or <word1.txt,1>;<total,1>."hadoop" <word1.txt,1>;<total,1>."hadoop <word1.txt,1>;<total,1>."map" <word1.txt,1>;<total,1>."reduce" <word1.txt,1>;<total,1>."write <word2.txt,1>;<total,1>.(browser <word2.txt,1>;<total,1>.(hadoop <word1.txt,1>;<total,1>.(hdfs)) <word1.txt,1>;<total,1>.(hdfs)

– <word1.txt,1>;<total,1>.(mapreduce). <word1.txt,1>;<total,1>.(of <word1.txt,1>;<total,1>.(wora),[13]

meaning <word2.txt,1>;<total,1>.1995 <word2.txt,1>;<total,1>.2007, <word2.txt,1>;<total,1>.2015, <word2.txt,1>;<total,1>.9 <word2.txt,1>;<total,1>.3.5单词共现单词共现是对实际问题的抽象化描述方式，在本设计中，实现的方式是采用pairs算法，伪代码如下：map(offset,text)forallwordwistextforallworduiswindow(w)Output(pair(w,u),1)reduce(pairp,countlist)s=0forallcountcincountlists=s+coutput(pairp,s)需要定义window，可以是一个固定长度，也可以是前后相连出现或在同一句中出现又或者在同一段落中出现的单词，根据实际的使用情况可以自行更改。在map函数的输出中，输出的键格式为pair，由于采用输出的键为单词对的格式，Hadoop中并没有相关的数据格式，因此必须先定义一个TextPair类来实现。在一般情况下也可以通过将两个单词连接成一个字符串，再转化为自带的Text类型来实现，但是在将单词对作为整体同时，也必须实现自己的partitioner，可以通过继承自带的HashPartitioner来实现自定义的分区方式（如果采用默认的分区方式，则会以单词对作为分区的关键因素而不是第一个单词，这很有可能会导致错误）。在实际应用中，还应该考虑到应该将整个文档作为一个整体，以保证在作为输入时不会被拆分而传输到不同的mapper中。为此，在这里会实现自己的WholeFileInputFormat输入格式，同时也需要自定义RecordReader来读取文件内容作为record的值。在实现中，以正则表达式的方式，只匹配单词，忽略其它字符。数据输出内容如下：All the 1Amazon Web 1Apache HBase 1Apache Hadoop 4Apache Hive 1Apache Pig 1Apache Spark 1App Engine 1As of 2Azure Amazon 1C and 2C but 1Classpath standard 1Common contains 1Community Process 1Compiler for 1Corporation and 1Distributed File 2Engine and 1Facebook and 1File System 3第四章结论在分布式环境中，Hadoop能有很好的性能，但是本设计的环境采用伪分布来模拟完全分布式环境，所以并不能显示高性能，反而，由于其复杂的框架使得在伪分布式情况下会多出许多多余的操作，大大降低了程序的速度。此外，分布式计算的实现也不是万能的，虽然分布式计算是一种很好的解决方案，但是在实际中也应该针对具体问题选择合适的方法。在选择Hadoop作为分布式计算平台的时候，必须考虑实际运行情况对参数配置进行调优，适时根据集群和作业的变化进行调整，以达到最优化效果。MapReduce是一种十分简单的计算模型，也提供了很大的灵活度，所以在实际开发过程中，可能会遇到各种问题，也正是因为其强大的扩展能力，给程序的开发造成了一定程度上的困难，开发过程中也有许多的中间结果处理方式，选择合适的中间结果输出格式和类型至关重要。在任务比较复杂和情况下，可以采用组合式MapReduce任务的方式，也可以采用自定义数据格式和输入输出格式以达到目的。总之，Hadoop是一个很好的分布式存储和计算的平台，适时选择其作为开发工具可能会有意想不到优势。致谢在整个设计过程中，我得到了许多帮助，十分感谢我的导师刘国才教授，在我对我的设计方向选择上提供了十分关键的指导，也一直在督促我的学习和设计进度，让我能时刻保持前进，路总是要靠自己去走，方向便成了至关重要的因素。从一开始接触时的迷茫到逐渐开始入门到对整体有大致的了解，我学会了大量的新知识，感谢刘国才老师给我这个难得的机会。参考文献附录程序源代码单词计数：importjava.util.Random;importjava.io.IOException;importjava.util.StringTokenizer;importorg.apache.hadoop.conf.Configuration;importorg.apache.hadoop.fs.Path;importorg.apache.hadoop.io.IntWritable;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Job;importorg.apache.hadoop.mapreduce.Mapper;importorg.apache.hadoop.mapreduce.Reducer;importorg.apache.hadoop.mapreduce.lib.input.FileInputFormat;importorg.apache.hadoop.mapreduce.lib.input.TextInputFormat;importorg.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;importorg.apache.hadoop.mapreduce.lib.output.TextOutputFormat;importorg.apache.hadoop.mapreduce.lib.output.FileOutputFormat;//importorg.apache.hadoop.mapreduce.lib.reduce.IntSumReducer;importorg.apache.hadoop.util.GenericOptionsParser;importorg.apache.hadoop.mapreduce.lib.map.InverseMapper;//importorg.apache.hadoop.mapreduce.lib.map.TokenCounterMapper;publicclassWordCount1{publicstaticclassTokenizerMapperextendsMapper<LongWritable,Text,Text,IntWritable>{privatefinalstaticIntWritableone=newIntWritable(1);privateTextword=newText();publicvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{StringTokenizeritr=newStringTokenizer(value.toString());while(itr.hasMoreTokens()){word.set(itr.nextToken());context.write(word,one);}}}publicstaticclassIntSumReducerextendsReducer<Text,IntWritable,Text,IntWritable>{privateIntWritableresult=newIntWritable();publicvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{intsum=0;for(IntWritableval:values){sum+=val.get();}result.set(sum);context.write(key,result);}}/*publicstaticclassNoTwoextendsMapper<Text,Text,Text,Text>{ publicvoidmap(Textkey,Textvalue,Contextcontext)throwsIOException,InterruptedException{context.write(value,key);}}*/publicstaticvoidmain(String[]args)throwsException{ Configurationconf1=newConfiguration(); //theconfigurationissame!String[]otherArgs=newGenericOptionsParser(conf1,args).getRemainingArgs();if(otherArgs.length!=2){System.err.println("Usage:wordcount1<in><out>");System.exit(2);} PathtempDir=newPath("word-count-temp-"+Integer.toString(newRandom().nextInt(Integer.MAX_VALUE))); Jobjob1=newJob(conf1,"wordcount1");job1.setJarByClass(WordCount1.class);job1.setInputFormatClass(TextInputFormat.class); //defaultjob1.setMapperClass(TokenizerMapper.class);job1.setCombinerClass(IntSumReducer.class); //notnecessaryjob1.setReducerClass(IntSumReducer.class);job1.setOutputKeyClass(Text.class);job1.setOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job1,newPath(otherArgs[0]));FileOutputFormat.setOutputPath(job1,tempDir); job1.setOutputFormatClass(TextOutputFormat.class); //default job1.waitForCompletion(true); Jobjob=newJob(conf1,"wordcount2"); job.setJarByClass(WordCount1.class);job.setInputFormatClass(KeyValueTextInputFormat.class);job.setMapperClass(InverseMapper.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(Text.class);FileInputFormat.addInputPath(job,tempDir);FileOutputFormat.setOutputPath(job,newPath(otherArgs[1]));job.waitForCompletion(true);}}//调用自带的mapper类和reducer类实现/*publicclassWordCount{ publicstaticvoidmain(String[]args)throwsException { Configurationconf=newConfiguration(); if(args.length!=2) { System.err.println("Usage:wordcount<in><out>"); System.exit(2); } Jobjob=newJob(conf,"wordcount"); job.setJarByClass(WordCount.class); job.setMapperClass(TokenCounterMapper.class); job.setCombinerClass(IntSumReducer.class); job.setReducerClass(IntSumReducer.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(IntWritable.class); FileInputFormat.addInputPath(job,newPath(args[0])); FileOutputFormat.setOutputPath(job,newPath(args[1])); System.exit(job.waitForCompletion(true)?0:1); }}*/Grep实现：//RegexMapper.javapackagemyHadoop.text;importorg.apache.hadoop.conf.Configuration;importjava.io.IOException;importjava.util.regex.Matcher;importjava.util.regex.Pattern;importorg.apache.hadoop.io.LongWritable;importorg.apache.hadoop.io.Text;importorg.apache.hadoop.mapreduce.Mapper;/**A{@linkMapper}thatextractstextmatchingaregularexpression.*/publicclassRegexMapper<K>extendsMapper<K,Text,T