【《基于Hadoop的数据分析系统设计》15000字（论文）】

第一章引言1.1课题背景以及意义随着信息网络的发展和普及,人们已经进入了一个信息化的社会。许多新生产业借助互联网的风口蓬勃发展。如今，人们传统的生活方式已经被新生的产业所改变，像网络购物，网上点餐等，人们日常生活的种种需求都可以基于网络应用得到满足。和现实生活中一样，人们在网络中也会留下足迹，而这些足迹是以数据为载体进行记录的。这些数据的特点是多元，异构，且大量的。如何从这些数据中发现价值，并进行应用，是当下人们所研究的热点，谁能从数据中发现价值，谁就能在激烈的商业竞争中占据有利位置，也有不少企业借此，实现了企业飞速的发展。但是大家都会面临一个难题：已有的传统架构已经无法解决庞大数据的存储和计算。所以开启数据价值挖掘的前提是必须具有对海量数据进行存储和计算的能力。但是对海量数据进行存储和计算并非容事，主要有以下几个原因：1.存储压力：数据的量级是比较庞大的大，在一些较大的互联网企业中，每天产生的数据的量应该在TB，甚至PB级别。2.计算压力：对这样巨大的数据进行计算，需要消耗巨大的资源，目前没有任何一台服务器可以做到这一点。随着技术的不断迭代和发展，目前人们基于分布式技术已经可以做到对庞大的数据集进行存储和计算。随着信息化产业的飞速发展，信息的爆发增长已经是必然的趋势，后续对数据的处理和计算必将都会向分布式处理进行倾斜。因此掌握对数据进行分布式处理是十分有必要的。所以本论文基于分布式框架Hadoop，通过hadoop，搭建数据处理计算平台，利用Hadoop的分布式文件存储系统HDFS进行数据的存储，利用Hadoop的分布式计算框架MapReduce进行数据的存储和计算。在以下的文章中，我将以企业实际业务场景下数据为例，介绍数据分析平台的架构和实现，以及如何基于数据计算平台进行数计算处理，最终进行数据的可视化展示。1.2Hadoop发展1.2.1Hadoop的诞生和发展Hadoop最初的模型是Apache的Nutch项目,其作者是Doug。最初主要想要解决的问题是如何在一个网络上进行搜索和抓取数据。基于Nutch，人们可以比较简单的实现一个网络引擎，提供网页抓取、网页的索引、以及查询等功能。最开始，Doug通过四台廉价的服务器初步实现了数据的存储。但是对于存储海量的网络数据还是远远不够，很快，四台服务器就到达了存储瓶颈。此时受益于google的两篇论文，介绍了google内部对于分布式文件存储GFS和分布式计算MapReduce的实现思路。Doug使用java程序对google的GFS和MapReduce进行了实现，用于大规模分布式计算。很快就成为Apache的顶级项目。2006年,Doug把Hadoop从Nutch项目中分离出来,组建了Hadoop项目,此时Hadoop正式面世,在第一个版本，hadoop主要由三个组件组成：HDFS分布式文件系统，Mapreduce计算框架，以及公用拓展包common组成。但是此时的hadoop集群的稳定性有待改善，当集群的工作节点出现故障的时候，整个集群处于不可用的状态，此外，整个MapReduce框架不仅负责分布式任务的计算还负责整个集群计算资源的分配，导致整个系统的耦合性较大。2012年，hadoop推出2.X版本，这也是hadoop被业界广泛接受并应用的一年。在hadoop2.x版本，着重保障了整个hadopp集群的高可用，采用主从的模式，保证当有一个集群的工作节点出现故障的时候，从节点会去保证整个集群的稳定。此外，还单独抽象出一个资源管理分配组件yarn，将分布式计算和资源分配独立出来，减少了系统欸但耦合性。2017年11月，hadoop推出了3.X版本，主要对2.x进行了一些性能优化，将集群的java版本从7升级到了8，此外，在计算方面，对内部mapReduce算法继续了优化，对于频繁进行磁盘io的作业，性能可以优化30%。基于hadoop，人们可以直接将数据导入在分布式文件系统进行存储，而且MapReduce框架可以将人们自己编写的代码和本身的分布式计算框架组件进行整合,形成一个新的完整的分布式运算程序,并将其发布运行在一个Hadoop集群上，开发人员可以不需要有什么并发处理或者分布式系统的经验，就可对大量数据进行分布式计算，不需要关心分布式框架的底层实现原理，大大提高了开发效率。这也是hadoop在业界被广泛使用的原因。这也是本论文使用hadoop来实现计算处理平台的原因。1.2.2Hadoop的特点1、在不需要十分低延时的场景下,有很高的吞吐量,十分的适合针对于海量的数据的运算有一定的可靠性保障:HDFS对于数据是自动去维护多个副本的,所以在任务运行失败了以后，能自动的再去重新的部署和再次运行发布的分布式计算任务。2、耗费的成本比较低,在可以控制范围内:可以通过普通的机器来搭建服务器集群,去存储、计算这些数据。这些廉价的服务器集群数量可以很多,而且可以运行在开源的操作系统上3、有很高的处理效率:通过对数据进行分布式的存储,可以在集群所在的节点上并行地处理它们,处理速度很快。1.3论文结构第一章介绍了课题的背景和意义，对研究内容和研究目标进行了阐述。重点介绍了Hadoop的研究现状和以及优势。第二章阐述基于Hadoop的基本理论，详细介绍了Hadoop的分布式文件系统HDFS和它的核心分布式计算思想：MapReduce并行编程模型，以及基于hadoop的作业提交流程。第三章给出了基于Hadoop的数据计算平台的设计思路介绍数据集的收集处理流程。第四章介绍了计算平台的部署实例和工作流程，并摘录了部分核心代码，以及最终计算结果的可视化展示。第五章总结了本课题的研究成果，指出了需要进一步改进的方面。第二章相关技术研究2.1Hadoop2.1.1Hadoop简介Hadoop是软件基金会(Apache)的旗下的一个开源的分布式的计算平台，主要解决的海量数据的存储以及对海量数据的分析计算。本文以hadoop2.7版本为基础，进行计算平台的搭建2.1.2Hadoop架构图一为Hadoop1.X架构的组成图-1Hadoop1.X架构集群工作节点:整个机器集群中的最重要的节点,主要用来管理整个集群当中的各种数据。集群工作节点的辅助节点:主要是协助集群工作节点来进行数据管理。集群的服务节点:集群当中的工作节点,主要是用于存储整个集群当中的数据块信息。作业管理节点:是一个接受客户端提交的计算任务的服务节点,负责计算资源的调度，将任务分配给集群的工作节点运行。集群的工作节点:集群的实际工作节点，负责执行作业管理节点分配的任务。图二为Hadoop2.X架构组成：图-2Hadoop2.X架构集群工作节点:整个机器集群中的最重要的节点,主要用来管理整个集群当中的各种数据。集群工作节点的辅助节点:主要是协助集群工作节点来进行数据管理。集群服务节点:也是整个集群当中的工作节点,主要是用于存储整个集群当中的数据块的信息。资源管理者:是一个接受客户端提交的计算任务的服务,并将任务分配给集群的工作节点。作业执行节点:主要来执行资源管理者分配的计算任务。在Hadoop1.x版本,Hadoop中的MR逻辑不仅要处理业务上的计算运算,还要对作业的资源调度进行分配,导致整个任务的耦合性比较大,所以在第二个较新的版本2.x,增加了资源管理组件,资源管理组件只负责资源调度,计算框架MR只负责运算。2.2Hadoop中的分布式文件系统的实现2.2.1HDFS概述数据的增长速度是飞快的,随着产生的数据量越来越多,一个单一的操作系统已经无法在存储这些海量的数据,因此需要将数据都分配到更多的操作系统来进行存储,但是此时就需要一个文件系统去管理多个操作系统上的文件,分布式文件管理系统由此而生，HDFS本质上就是一种分布式文件管理系统,适合存储海量的数据而且不支持文件的修改,适合用来作为数据分析，面向一次写入,多次读出的场景。2.2.2HDFS架构图-3HDFS架构HDFS主要由集群工作节点,集群服务节点,集群工作节点的辅助节点,客户端组成。集群工作节点:是整个集群的管理者,管理分布式的文件系统的名称空间,负责去处理与客户端之间的请求,配置置副本信息,管理各个数据快之间的映射关系。集群服务节点:是工作节点,存储实际的数据块,负责数据块的读/写操作。客户端:向集群工作节点获取文件位置信息,与集群服务节点进行交互,来进行文件的上传和进行数据的读取。集群工作节点的辅助节点:协助集群工作节点工作,分担集群工作节点的数据压力,在一些比较紧急情况下可以用于辅助的恢复集群工作节点的数据。2.2.3HDFS的存储实现图-4HDFS中数据的写入流程1.首先，客户端会发起文件的上传请求,通过RPC接口与集群工作节点建立连接通道,集群工作节点会先去检查一下希望被写入的文件名称是否是已经存在,以及其父目录是否已经存在,并检查用户是否有相应的上传权限,若检查通过,会为该文件创建一个新的记录,否则，文目录会创建失败,然后会给客户端返回是否可以上传文件。2.当客户端收到可以上传文件的请求时，客户端会再次向集群工作节点发送请求,来获取第一个数据块应该被传输到哪些集群的服务节点所在的服务器上。3.收到客户端的请求后，集群的工作节点首先会根据文件中的配置文件来去设定数据块存储的副本数量,然后根据机架的感知原理对文件进行文件存储节点的分配,然后集群的工作节点会给客户端返回一组正常工作的集群的服务节点的地址,这里用三台集群的服务节点为例:X、Y、Z。当初在对Hadoop结构进行设计时候,考虑到对数据块文件的完整性进行安全与高效地保障,决定默认将数据块在HDFS上按照保存三个副本策略进行存储,存储的主要策略是:选择和客户端相同的集群的服务节点进行第一个备份的存放，如果若客户端请求不在集群的内部,集群中任意一个服务节点都可以来存放第一个副本,然后会选择与第一个副本所在的机架不同的一个随机集群上的服务节点上进行放置二个副本,最后会在第二个副本相同机架上的放置第三个副本，如果当集群文件块的副本数大于三,那么随后的副本在集群中的存放策略就是随机存放,尽量地去避免较多的副本被存放在了同一个机架上。4.当收到集群的工作节点返回的服务节点列表后，客户端会对其中的一个服务节点X发送上传数据请求,本质上是一个远程接口的调用,会建立一个网络运输管道,X收到请求后,先进行数据块的存储，然后通过接口调用Y，Y节点会进行同样的操作，在收到请求后,会去调用节点Z,最后,将完整的网络运输管道搭建完成,依次返回确认信息给客户端。5.当网络运输管道建立完成后，客户端会先将数据块放在本地的内存进行缓存，然后会向X上传第一个数据快,以数据包为单位,X节点在收到数据包后,会再次将这个数据包传递给Y,Y同样,在收到数据包后,会再次传递给Z，其中X在每一次传递完数据包后,都会将这个数据包放到自己内部的一个用于容灾的应答队列中去等待应答，这样设计原因主要是应对数据传输中，集群出现故障的情况，会在下文进行说明。6.在数据包的传输过程中，当某个提供服务的集群数据服务节点出现了故障,HDFS会进行一系列的操作来保障数据的准确性和完整性。首先客户端会把这些数据块进行分割,分割成为一个个的数据包,默认大小是64K,这些数据包会被再次发送到网络运输的管道中,再进行一次传输,而在网络运输管道反方向上,客户端的等待队列会等待每个数据包的正确应答信号,也就是ack消息,最终由网络运输的管道中第一个数据服务节点,将网络运输管道的命令正确应答发送给客户端。具体流程如下:首先会将先前的网络运输管道进行关闭,然后那些已经在确认队列上的数据包都会被移动到队列的最开始也就是首部,再次被发送,这样做的目的是保障在网络运输管道上,那些损坏的集群的服务节点的下游数据服务节点不会被损坏的集群数据服务节点所影响,导致数据包的丢失。其次，对于没有出现故障的集群的数据服务节点，HDFS会在正常的数据块上去做一个特殊的标识,这样当出现故障的的集群的数据服务节点在重新启动以后，集群的管理节点可以根据标识来判断数据服务节点上哪些数据块是刚才故障节点残留下的，从而把他删除掉。然后故障的数据服务节点会从运输管道中移除,而对于那些不受影响的数据块会继续被写入到其他还在正常运行的数据服务节点中,此时集群工作节点会根据元信息知道哪些数据块是处于即备份数量不够的状态下,然后它会进行协调处理，在一个服务节点生成一个新的副本来保证集群所要求的备份数量，后续的数据块写入就不受影响。最后当一个数据块被成功的接受以后,客户端再次发送请求向集群的工作节点,去将下一个数据块发送到集群的服务节点上。HDFS数据读取实现：与数据写入HDFS流程相似。首先,客户端会通过分布式的文件系统中的一个DistributedFileSystem的函数向集群的工作节点去请求下载文件,集群的工作节点会先去查询数据对应的数据块之间的元数据,这样就可以找到数据块所在的集群的工作节点的地址。然后客户端会随机地去挑选一个数据服务节点,保持就近的原则,选择最近的服务器,来请求读取数据。当收到客户端当请求后，集群的服务节点开始将数据块传输给客户端。最后客户端以数据包为单位接收,先在本地缓存,然后写入目标文件。2.3MapReduce计算框架底层原理2.3.1MapReduce概述MapReduce是一个分布式的运算的程序编程框架,下文称为MR,它的核心功能是将人们自己编写的代码和本身的分布式计算框架组件进行整合,形成一个新的完整的分布式运算程序,并将其发布运行在一个Hadoop集群上。2.3.2MapReduce核心思想图-5Mapreducec处理流程MR运算程序会被分为两个阶段:Map实例阶段和Reduce实例阶段。在Map阶段:多个Map阶段的实例会并发的运行,之间不会进行干扰。在Reduce阶段:多个Reduce阶段的实例会被并发的运行,之间不会进行干扰,但是他们都会去依赖于上一个阶段的所有Map实例阶段的输出数据。因为MR的编程模型规定只能允许包含一个Map实例运行阶段和一个Reduce实例运行阶段,如果我们的作业逻辑比较复杂,那我们只能将作业拆解为多个MR程序,依次串行地运行。2.3.3MapReduce实现原理Map阶段：(1)读取处理阶段:在读取处理阶段,首先会通过我们自己定义的数据读取方法RecordReader（）,从指定的文件中读取数据,并按照定义的解析方式进行解析,解析成为一个个健值对。(2)Map实例处理阶段:在这个阶段,主要的工作是将前一步读取阶段解析出来的健值对,交给我们自己编写定义的一个函数map()函数中去进行处理，按照我们自己的逻辑去处理,生成一系列新的健值对。(3)收集处理阶段:在我们编写的map函数中,执行完处理逻辑后,会将新的健值对,通过调用叫做OutputCollector.collect()的方法写到一个预留的内存缓冲区中。在这个函数内部,它会再次调用一个Partitioner（）方法将新的健值对进行分区处理。(4)Spill阶段：即“溢写”，当环形缓冲区满后，MapReduce会将数据写到本地磁盘上，生成一个临时文件。需要注意的是，将数据写入本地磁盘之前，先要对数据进行一次本地排序，并在必要时对数据进行合并、压缩等操作(5)合并处理阶段:当所有输入的数据处理完成后,Map阶段的实例会对先前所有文件再进行一次合并,来确保在map阶段运行所有实例最终只会生成一个文件。Reduce阶段：(1)复制（copy）处理阶段:Reduce阶段的实例会从先前的各个Map阶段实例所产生的文件进行远程拷贝,在拷贝一个文件的时候,如果这个文件的大小超过规定的阈值,则写到磁盘上,否则直接放到内存中。(2)Reduce合并处理阶段:因为进行远程拷贝数据,Reduce运行实例会启动两个后台线程。主要的目的是对内存和磁盘上的文件进行一次合并,防止远程拷贝的文件占用过多的磁盘存储空间以及使用的内存超出限制。(3)Sort阶段:这个时候会按照MapReduce定义的逻辑,调用我们自己编写的reduce()函数,这里Hadoop采用了一个基于排序的策略，来对map阶段运行实例产出的数据按照健值对中的key进行聚合,可以这样做的目的是为了将健值对中key相同的数据聚和在一起。因为在各个Map阶段运行实例已经对自己的处理结果数据进行了局部排序,因此,Reduce阶段实例只需对所有数据再进行一次归并排序即可。(4)聚合（merge）阶段:在这个阶段,会将我们最终的计算结果写到HDFS指定的路径下。2.4Yarn资源调度系统2.4.1Yarn概述Yarn是在Hadoop新版本中新增的一个组件,我们可以理解为它是一个针对于资源的一个调度平台,主要是负责为我们提交的MR运算程序来提供计算资源,相当于一个分布式的资源调度管理的系统平台,我们提交的MR运算程序是基于这个资源调度平台进行运作的。2.4.2Yarn的系统架构图-6Yarn的系统架构Yarn主要由资源管理节点、工作节点、应用管理节点和容器等组件构成。资源管理节点:主要是面向客户端的提交请求,其次是监控工作节点的状态,启动/监控应用管理节点,还有就是进行资源的分配和调度。工作节点:是每个计算作业实际运行的节点,我们可以理解为资源管理节点相当于是公司的老板，而工作节点是公司的经理，工作节点管理单个节点上的所有资源,响应处理资源管理节点的命令,处理应用管理节点的请求。应用管理节点:主要是负责数据的切分工作,为内部程序申请资源,进行一些任务的监控。容器:可以理解是Yarn对资源的一种抽象,其中封装了多个维度的资源,如内存,CPU,磁盘,网络等。2.4.3基于Yarn的作业运行实现首先客户端会调用一个job.waitForCompletion方法,向整个计算集群提交一个MR作业。这时候,客户端向资源管理节点申请一个作业id。资源管理节点会给客户端返回该这个作业依赖的资源需要提交的路径和作业id。收到资源管理节点的返回信息后,客户端会提交需要运行的代码、数据的切片相关的信息和配置文件到资源管理节点指定的提交路径。当客户端提交完资源后,会向资源管理节点申请运行应用管理节点。当资源管理节点收到客户端的请求后,会将这个作业添加到自己的资源容量调度器中,等待某一个空闲的工作节点来领取这个任务。当有空闲的工作节点认领任务后,该工作节点会创建容器,并创建应用管理节点。这时会获取先前客户端提交的程序以及配置信息,将这些数据下载到本地。这个时候,应用管理节点会根据作业的配置,向资源管理节点申请运行多个Map阶段实例任务资源。资源管理节点会将运行Map阶段实例任务分配给另外两个工作节点,另两个工作节点同样会领取任务并创建容器。应用管理节点向其他两个接收到任务的工作节点发送程序启动脚本,这是，三个工作工作节点会分别启动Map阶段运行实例对数据分区排序处理。应用管理节点在等到所有Map阶段运行实例运行完毕后,向资源管理节点再次申请容器,运行Reduce阶段实例。Reduce阶段实例从先前Map阶段运行实例产出的路径下获取相应分区的数据。等待程序运行完成后,应用管理节点会向资源管理节点申请注销自己。这样一个完整的MR程序就执行完成了。2.5HIVE首先Hive任务是一个用于构建数据仓库的分析工具,它目前已经支持且提供比较多的SQL语法,这样在我们对存储在Hadoop分布式的文件系统中的数据进行查询的时候,可以对一些有结构的数据数据进行映射,这样我们就可以直接通过hive提供的sql语句进行数据的查询分析，减少了代码的开发，提高了开发效率。所以总的来讲,Hive十分适用数据仓库相关的应用,因为我们都是对相关的静态的数据数据进行分析,所以在不要求时效性,不需要在很短的时间内给出结果的场景下,且数据本身不会频繁变化,那么可以使用hive来维护海量数据,对数据进行处理计算。

第三章数据分析平台设计3.1系统需求分析目前业务发展十分迅速，每天产生的数据的体量是十分大的，我们需要将这些数据收集起来，进行统计计算，得到对业务有价值的数据，为业务发展提供支持。3.2数据分析平台的架构设计结合业务需求，本文构建基于Hadoop搭建数分析平台，主要包括数据接入模块，数据归一化模块，实际同步模块，可视化模块。图-7数据计算平台的架构设计3.2.1数据接入目前数据按照收集流程可以分为两个部分：用户的行为数据收集：图-8用户数据收集设计这部分数据是用户在浏览应用时所产生的，每个应用在后台会对用户的行为埋点数据进行上报，这部分数据会被写入到我们指定的消息队列中，然后我们会编写代码，将消息队列中的日志数据写入到指定的文件路径下。但是考虑到用户的行为是随时发生的，此时移动文件，会导致后续的数据被写入新的文件，导致产生大量的小文件，会在后续导入分布式文件系统带来额外的系统开销，所以我们这边会在每天的凌晨12点执行命令，将前天的用户行为数据导入到分布式文件系统中去。业务数据收集：图-9业务数据收集设计这部分数据主要来源是mysql数据库，因为这部分数据需要保证事务的一致性，而且这部分数据的格式是已经定义的，所以我们只需要将数据从mysql中导入到HDFS系统中去。同样，这部分数据是在每天的凌晨12点执行命令，将前天的为数据导入到分布式文件系统中去。3.2.2数据结构归一化因为我们目前基于数据收集模块，是将所有的数据都导入到HDFS分布式文件系统中去的。这些数据是是多样性的且格式是不一致的，我们没有办法对这些数据直接进行计算，所以我们需要对数据进行建模：就是对我们每日存储在数据仓库中的数据按照我们自己定义的业务主题对这些数据进行划分和处理，不遵循传统的第三范式，是面向业务的。往往为了能方便的得到需要的指标数据，会进行一定程度数据的冗余。这样，就可以复用用一套代码对这些数据进行计算。首先在我们进行数据建模之前，先明确几个概念：主题。主题是我们自己根据业务场景，业务特点进行综合、归类和分析利用的一个抽象概念,每一个主题面向的是一个特定业务领域。维度。是事物或现象的某种特征,如性别、地区、时间等，这些都属于都是维度。时间是我们分析中最常用的一个维度，可以通过时间进行对比，可以知道事物的发展轨迹是否符合我们的预期目前广泛使用的数据维度建模分为三种模型:星型模型、雪花模型、星座模型。星型模型和雪花模型的区别在于维度的层级,标准的星型模型维度只有一层,而雪花模型可能会设计多层。星座模型与前两者的区别在于事实表的数量,星座模型是基于多个事实表。我在这里我们按照业务主体，按照星形模型，使用进行数据维度建模，来规范数据的格式。计算平台按照用户行为数据和业务数据，以星座模型来进行数数据建模。对数据按照主题进行划分用户行为数据主题。按照用户浏览行为，可以分为以下几点用户登录行为用户的下单行为用户的支付行为用户的收藏行为用户的评论行为用户的点击行为业务数据主题订单数据订单详情数据商品数据支付流水数据退单数据优惠卷顺序活动数据我们对这些数据按照分层策略来进行计算，将整个计算流程分为四个部分：ODS层：在这一层，是直接复用hdfs中的数据，bujinxi保持数据原貌不做任何修改，起到备份数据的作用。DWD层：在这一层，主要对ods层的数据进行清洗处理，核心数据进行判空过滤。DWS：基于dwd层数据，对数据进行聚合统计。ADS：基于DWS层数据，进行指标数据统计计算。这样做的主要考虑如下：首先是性能的提升，可以帮助我们快速查询需要的数据,减少系统的开销。其次可以降低存储成本，极大减少不必要的数据冗余,能更好的实现计算结果复用。最后是效率和质量，好的数据模型可以提高数据的使用效率,改善数据统计口径不一致性,提高数据计算效率。3.2.3数据可视化展示这一部分只要是针对ADS层计算的指标结果数据进行前端可视化看板的展示，采用开源的可视化工具SuperSet来进行处理。它是目前开源的数据分析和可视化工具中比较好用的，图表类型多，功能简单，灵活支持支持多种数据源。我们先将指标结果写入mysql中，然后SuperSet再访问数据库将指标进行展示。3.3平台功能模块3.3.1数据收集模块1.用户的行为收集：将用户的行为数据从消息队列中写入到指定HDFS文件图-10业务数据收集步骤这里我们使用hadoop支持的flume工具来完成这个工作，flume是一个日志采集、聚合和传输的系统，通过flume，我们只需要填写配置文件，就可以运行flume来实现日志数据的采集。编写flume配置文件：，Vimcollect.sha1.sources=r1a1.channels=c1a1.sinks=k1##source1a1.sources.r1.type=org.apache.flume.source.kafka.KafkaSource--定义数据源a1.sources.r1.batchSize=5000a1.sources.r1.batchDurationMillis=2000a1.sources.r1.kafka.bootstrap.servers=hadoop102:9092,hadoop103:9092,hadoop104:9092--配置hadoop集群的信息a1.sources.r1.kafka.topics=topic_erceptors=erceptors.i1.type=erceptor.TimeStampInterceptor$Builder##channel1a1.channels.c1.type=filea1.channels.c1.checkpointDir=/opt/module/flume/checkpoint/behavior1a1.channels.c1.dataDirs=/opt/module/flume/data/behavior1/配置写入HDFS的文件路径a1.channels.c1.maxFileSize=2146435071a1.channels.c1.capacity=1000000a1.channels.c1.keep-alive=6##sink1a1.sinks.k1.type=hdfsa1.sinks.k1.hdfs.path=/origin_data/gmall/log/topic_log/%Y-%m-%da1.sinks.k1.hdfs.filePrefix=log-a1.sinks.k1.hdfs.round=falsea1.sinks.k1.hdfs.rollInterval=10a1.sinks.k1.hdfs.rollSize=134217728a1.sinks.k1.hdfs.rollCount=0##控制输出文件是原生文件。a1.sinks.k1.hdfs.fileType=CompressedStreama1.sinks.k1.hdfs.codeC=lzop##拼装a1.sources.r1.channels=c1a1.sinks.k1.channel=c1编写flume的启动脚本：vimcollect.sh#!/bin/bashcase$1in"start"){foriinhadoop104doecho"启动$i消费flume"ssh$i"nohup/opt/module/flume/bin/flume-ngagent--conf-file/opt/module/flume/conf/kafka-flume-hdfs.conf--namea1-Dflume.root.logger=INFO,LOGFILE>/opt/module/flume/log2.txt2>&1&"done};;"stop"){foriinhadoop104doecho"停止$i消费flume"ssh$i"ps-ef|grepkafka-flume-hdfs|grep-vgrep|awk'{print\$2}'|xargs-n1kill"done};;Esac业务数据的导入;因为业务数据都是存储在mysql中，我们选择使用hadoop体系的sqoop工具来实现数据从mysql到hdfs的导入。编写导入脚本：vimmysql_to_hdfs.shsqoop=/opt/module/sqoop/bin/sqoop//sqoop的安装路径if[-n"$2"];thendo_date=$2elsedo_date=`date-d'-1day'+%F`fiimport_data(){$sqoopimport\--connectjdbc:mysql://hadoop102:3306/$APP\//需要查询的mysql数据库--usernameroot\--password000000\--target-dir/origin_data/$APP/db/$1/$do_date\//需要导入的HDFS路径--delete-target-dir\--query"$2and\$CONDITIONS"\--num-mappers1\--fields-terminated-by'\t'\--compress\--compression-codeclzop\--null-string'\\N'\--null-non-string'\\N'import_order_info(){//需要导入的mysql数据表import_dataorder_info"selectid,final_total_amount,order_status,user_id,out_trade_no,create_time,operate_time,province_id,benefit_reduce_amount,original_total_amount,feight_feefromorder_infowhere(date_format(create_time,'%Y-%m-%d')='$do_date'ordate_format(operate_time,'%Y-%m-%d')='$do_date')"}由于数据表较多，限于篇幅，以导入一张mysql数据表为例说明。到此，所有的数据均已经导入到hdfs中。3.3.2分布式计算模块我们使用hive来实现分布式计算，为对于平常的计算逻辑，如果我们使用MapReduce进行发开，逻辑是十分繁杂的，而hive对此做了一个封装，通过提供HSQL语言来简化代码逻辑,其实底层还是MapReduce程序。所以总的来讲,Hive十分适用数据仓库相关的应用,因为我们都是对相关的静态的数据数据进行分析。以次数用户出现次数为例说明：使用HSQL编写Selectuser_id,count（user_id）fromtable_namegroupbyuser_id；当我们在hive提交这段代码后，hive会对这个SQL程序进行解析，通过抽象语法树，将这段sql语言转换为mapReduce程序，提交到hadoop集群进行运行。而我们使用底层的MapReduce来实现，则需要多出许多的工作：首先编写我们的map函数：protectedvoidmap(LongWritablekey,Textvalue,Contextcontext)throwsIOException,InterruptedException{Stringline=value.toString();//定义输入的格式String[]words=line.split("");for(Stringword:words){context.write(newText(word),newIntWritable(1));}}编写我们的reduce函数：protectedvoidreduce(Textkey,Iterable<IntWritable>values,Contextcontext)throwsIOException,InterruptedException{Iterator<IntWritable>iterator=values.iterator();intsum=0;while(iterator.hasNext()){sum+=iterator.next().get();}context.write(key,newIntWritable(sum));}提交到hadoop集群：Configurationconf=newConfiguration();//在代码中设置JVM系统参数，用于给job对象来获取访问HDFS的用户身份System.setProperty("HADOOP_USER_NAME","gugu");//1、设置job运行时要访问的默认文件系统conf.set("fs.defaultFS","hdfs://master:9000");//2、设置job提交到哪去运行[默认local、yarn指定集群]conf.set("","yarn");//conf.set("yarn.resourcemanager.hostname","master");//3、如果要从windows系统上运行这个job提交客户端程序，则需要加这个跨平台提交的参数conf.set("mapreduce.app-submission.cross-platform","true");Jobjob=Job.getInstance(conf);//2、封装参数：本次job所要调用的Mapper实现类、Reducer实现类job.setMapperClass(WordCountMapper.class);job.setReducerClass(WordCountReduce.class);//3、封装参数：本次job的Mapper实现类、Reducer实现类产生的结果数据的key、value类型job.setMapOutputKeyClass(Text.class);job.setMapOutputValueClass(IntWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(IntWritable.class);job.setJar("D:\\ApplicationFiles\\IDEA\\hadoopTest\\myhadoop\\target\\hadoopTest.jar");//job.setJarByClass(JobSubmit.class);Pathoutput=newPath("/wc/output");FileSystemfs=FileSystem.get(newURI("hdfs://master:9000"),conf,"gugu");if(fs.exists(output)){fs.delete(output,true);}//4、封装参数：本次job要处理的输入数据集所在路径、最终结果的输出路径FileInputFormat.setInputPaths(job,newPath("/wc/input"));FileOutputFormat.setOutputPath(job,output);//注意：输出路径必须不存在//5、封装参数：想要启动的reducetask的数量job.setNumReduceTasks(2);//6、提交job给yarnbooleanres=job.waitForCompletion(true);可见，使用hive会我们的计算逻辑进行很大程度上的简化，来提高开发效率。3.3.3结构归一化模块图-11数据归一化和计算基于我们的数据收集流程，目前我们已经将我们的数据导入到了HDFS中，现在需要将hdfs的数据导入到hive中去，在这个环节，我们主要是按照我们定义好的分层处理模型，利用hive来进行数据格式的归一化，定义每个层级数据的结构，这里我们会将数据转换成hive表的结构来进行存储，然后基于hiveSQL来进行数据的处理和计算Ods数据导入：首先使用户的行为数据，这一部分数据是以json字符串的格式进行存储的，所以我们定义我们的用户行为表的格式如下：droptableifexistsods_event_log;CREATEEXTERNALTABLEods_log(`line`string)PARTITIONEDBY(`dt`string)STOREDASINPUTFORMAT'com.hadoop.mapred.DeprecatedLzoTextInputFormat'OUTPUTFORMAT'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'LOCATION'/warehouse/gmall/ods/ods_event_log';使用HDFS命令，将数据同步带hive中去：loaddatainpath'/origin_data/gmall/log/topic_event/2020-05-10'intotablegmall.ods_logpartition(dt='2020-05-10');使用hive支持的string类型来存储用户行为信息中的业务数据导入：因为业务数据都是存储在mysql，已经是表结构格式了，所以我们直接通过脚本导入即可，以部分表进行举例。loaddatainpath'/origin/gmall/db/sku_info/'2021-05-10''OVERWRITEintotablegmall.ods_sku_infopartition(dt=''2021-05-10'');loaddatainpath'/origin/gmall/db/order_info/‘2021-05-10'OVERWRITEintotablegmall.ods_order_infopartition(dt='2021-05-10');DWD首先是按照定义好的主题，进行数据维度的打散和聚合。分别建立数据表用户行为主题就会建商品点击表，商品详情页表，商品列表页表，广告表，消息通知表，用户后台活跃表，评论表，收藏表，点赞表，错误日志表。每个表的字段按照主题的维度进行定义。以用户评论表为例说明：定义用户评论表：包括用户的信息以及评论的内容CREATEEXTERNALTABLEdwd_comment_log(`mid_id`string,`user_id`string,`version_code`string,`version_name`string,`lang`string,`source`string,`os`string,`area`string,`model`string,`brand`string,`sdk_version`string,`gmail`string,`height_width`string,`app_time`string,`network`string,`lng`string,`lat`string,`comment_id`int,`userid`int,`p_comment_id`int,`content`string,`addtime`string,`other_id`int,`praise_count`int,`reply_count`int,`server_time`string)PARTITIONEDBY(dtstring)storedasparquetlocation'/warehouse/gmall/dwd/dwd_comment_log/'TBLPROPERTIES('pression'='lzo');但是在我们ods层的用户行为表中是包含了所有的用户行为的，所以我们需要从ods层的用户行为表中筛选出有评论行为的用户，这样，我们后续再次需要统计用户的评论行为，直接基于dwd层的用户评论表就可以得到结果，不用重复从ods层的用户行为表进行统计。Dwd层数据导入代码如下：selectmid_id,user_id,version_code,version_name,lang,source,os,area,model,brand,sdk_version,gmail,height_width,app_time,network,lng,lat,get_json_object(event_json,'$.ment_id')comment_id,get_json_object(event_json,'$.kv.userid')userid,get_json_object(event_json,'$.kv.p_comment_id')p_comment_id,get_json_object(event_json,'$.kv.content')content,get_json_object(event_json,'$.kv.addtime')addtime,get_json_object(event_json,'$.kv.other_id')other_id,get_json_object(event_json,'$.kv.praise_count')praise_count,get_json_object(event_json,'$.kv.reply_count')reply_count,server_timefromdwd_base_event_logwheredt='2020-05-10'andevent_name='comment';其中get_json_object是我们自己实现的函数，主要作用是从json字符串中解析出我们需要的字段。具体逻辑如下：//该方法中，我们将指定输出参数的名称和参数类型：publicStructObjectInspectorinitialize(StructObjectInspectorargOIs)throwsUDFArgumentException{ArrayList<String>fieldNames=newArrayList<String>();ArrayList<ObjectInspector>fieldOIs=newArrayList<ObjectInspector>();fieldNames.add("event_name");fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);fieldNames.add("event_json");fieldOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);returnObjectInspectorFactory.getStandardStructObjectInspector(fieldNames,fieldOIs);}//输入1条记录，输出若干条结果@Overridepublicvoidprocess(Object[]objects)throwsHiveException{//获取传入的etStringinput=objects[0].toString();//如果传进来的数据为空，直接返回过滤掉该数据if(StringUtils.isBlank(input)){return;}else{try{//获取一共有几个事件（ad/facoriters）JSONArrayja=newJSONArray(input);if(ja==null)return;//循环遍历每一个事件for(inti=0;i<ja.length();i++){String[]result=newString[2];try{//取出每个的事件名称（ad/facoriters）result[0]=ja.getJSONObject(i).getString("en");//取出每一个事件整体result[1]=ja.getString(i);}catch(JSONExceptione){continue;}//将结果返回forward(result);}}catch(JSONExceptione){e.printStackTrace();}}}//当没有记录处理的时候该方法会被调用，用来清理代码或者产生额外的输出@Overridepublicvoidclose()throwsHiveException{}然后我们需要将自己编写的函数进行打包，发送到HDFS文件系统中去。到此，dwd层的用户评论表数据就导入完成。其他层级的数据导入步骤类似，限于篇幅，不再进行展开阐述。3.3.4数据同步模块在这个模块，主要是将我们最终计算的指标结果从hive导入到外部存储Mysql。同样是通过sqoop脚本来实现：vimsqoop_export.sh#!/bin/bashhive_db_name=gmallmysql_db_name=gmall_reportexport_data(){/opt/module/sqoop/bin/sqoopexport\--connect"jdbc:mysql://hadoop102:3306/${mysql_db_name}?useUnicode=true&characterEncoding=utf-8"\//和需要导入的mysql数据库建立链接--usernameroot\--password000000\--table$1\--num-mappers1\--export-dir/warehouse/$hive_db_name/ads/$1\--input-fields-terminated-by"\t"\--update-modeallowinsert\--update-key$2\--input-null-string'\\N'\--input-null-non-string'\\N'}case$1in"ads_uv_count")export_data"ads_uv_count""dt"//导入hive的那张数据表;;"ads_user_action_convert_day")export_data"ads_user_action_convert_day""dt";;"ads_user_topic")export_data"ads_user_topic""dt";;"all")export_data"ads_uv_count""dt"export_data"ads_user_action_convert_day""dt"export_data"ads_user_topic""dt";;esac3.3.5数据可视化模块这个环节主要是从mysql中读取我们的指标结果，然后按照我们配置的数据展示格式进行最终结果的可视化展示。

第四章基于Hadoop的数据分析平台的部署4.1数据分析平台架构表4-1数据分析平台服务部署服务名称子服务Hadoop01Hadoop02Hadoop03HDFSNameNode部署DataNode部署部署部署SecondNameNode部署YARNNodeManager部署部署部署ResourceManager部署HIVEHiveService部署4.1.1硬件环境准备服务器三台：三台虚拟机，每台虚拟机2G内存，8G存储。操作系统：contos6.8图-12虚拟机配置参数4.1.2软件环境准备hadoop-2.7.2：Hadoop环境，配置hadoop集群环境。hive-2.3.6：hive，apache--bin.tar.gz构建hive环境，利用hive进行数据结构归一化Sqoop：sqoop-1.4.7.bin__hadoop-2.6.0.tar主要用于mysql和HDFS之间的数据同步Mysql：主要用于存储mysql数据库Java执行环境：jdk-8u144-linux-x64.tar.gz，hadoop的运行环境是基于java编程语言的，需要服务器具有java环境。IDEA：java开发工具，主要用于自定义函数的编写，界面如下图-13集成开发环境4.1.3运行环境配置java环境配置：修改虚拟机配置文件vim/etc/profile.d/my_env.sh添加如下内容：#JAVA_HOMEexportJAVA_HOME=/opt/module/jdk1.8.0_212exportPATH=$PATH:$JAVA_HOME/bin查看是否配置成功：java-version，出现以下说明配置成功。图-12java环境配置修改hadoop核心配置文件：vimcore-site.xml，添加下面配置图-14Hadopp环境配置分布式文件系统配置：指定文件存储副本个数，以及辅助节点地址图-15HDFS配置MapReduce配置：指定使用yarn来进行资源分配管理图-16MapReduce配置Yarn配置文件配置：图-17Yarn配置至此。Hadoop的集群环境搭建完毕，启动hadoop集群启动分布式文件系统：start-dfs.sh图-18启动HDFS启动Yarn：start-yarn.sh图-19启动YarnHive环境的搭建：配置hive和hadoop集群的映射图-20hadoop集群集成hive测试启动：bin/hive图-21启动hiveSqoop环境搭建:与hadoop集群进行集成。图-22配置sqoop在这里，我们需要将NameNode和SecondNameNode部署在两台机器上，这和他们的工作机制有关。工作机制如下：集群的工作节点在启动时,首先会滚动生成一个操作文件,然后会再生成一个空的操作文件,这时会将操作文件和上一次的快照信息加载到内存中,此时集群的工作节点的内存就相当于是有集群中最新的数据块映射的元数据信息以及创建的目录和上传的文件的序列化信息。在这期间如果客户端对集群的工作节点再次发送对元数据的增删改的请求,那么这些新增的增删改请求的操作首先会被记录到空的操作文件中,但是查询元数据的操作不会被记录在空的操作文件中,因为查询操作不会更改数据的元数据信息,如果此时集群的工作节点突然损坏,在重启后会从操作文件中读取那些增删改的操作。最后,集群的工作节点会在内存中去执行对元数据的增删改的操作。但是我们都知道,随着集群的运行,在操作文件中记录的增删改的操作会越来越多,操作文件占用的内存会越来越大,这就可能导致集群的工作节点在启动并把操作文件加载到内存的效率就会很慢,考虑到这一点,我们就需要定期对操作文件和快照进行合并,将操作文件和快照文件加载到内存中,按照操作文件中记录的操作一步步执行,来生成一个新的快照文件。但是这会加重集群的工作节点的负担，所以需要集群工作节点的辅助节点来帮助集群的工作节点来进行操作文件和快照文件的合并工作。首先,集群工作节点的辅助节点会定时向集群的工作节点询问是否需要进行合并，一般需要满足以下条件中的一个,一个是到达了配置文件中规定的合并时间，另外一个是操作文件写入的数据到达了设置的大小。当需要进行文件合并的时候，集群的工作节点会通过网络,告知集群工作节点的辅助节点进行文件的合并。集群工作节点的辅助节点在执行合并操作的时候,首先会让集群工作节点去滚动生成一个操作文件,还会生成一个空的操作文件,生成新的操作文件的目的是给操作文件打一个标记,以后所有新的操作都写入到空的操作文件中去。先前的操作文件和快照文件会被拷贝到集群工作节点的辅助节点本地,然后将拷贝的操作文件和快照文件加载到内存中进行合并,生成一个新的快照文件,然后将新的快照文件拷贝给集群的工作节点,重命名为新的快照,然后会去替换掉原来的快照文件。所以之后集群的工作节点再启动时就只需要加载之前未合并的操作文件和快照文件。因为合并过的操作文件中的元数据信息已经被记录在最新的快照文件中了。所以如果我们将这两台节点部署在同一个节点，会使这个节点的负载压力过大，会影响集群的稳定性，反而违背了hadoop设计的初衷。至此，整个数据计算平台搭建完成。图-23Hadoop集群UI界面4.2基于计算平台进行指标计算：1.启动数据导入HDFS脚本:hdfs_to_ods_log.sh2020-03-15#!/bin/bash#定义变量方便修改APP=gmallhive=/opt/module/hive/b