基于hadoopde大数据应用分析(4)

1、基于基于Hadoop的大数据应用分析的大数据应用分析靳靳 鹏鹏合肥工业大学，管理学院信管系合肥工业大学，管理学院信管系内容提要内容提要 1. HADOOP体系架构 2. 基于HADOOP的大数据产品分析 3.3. HADOOPHADOOP部署部署大数据对系统的需求大数据对系统的需求High performance 高并发读写的需求高并发、实时动态获取和更新数据Huge Storage 海量数据的高效率存储和访问的需求类似SNS网站，海量用户信息的高效率实时存储和查询High Scalability & High Availability 高可扩展性和高可用性的需求需要拥有快速横向扩展能

2、力、提供7*24小时不间断服务商业模式驱动应用需求驱动p云计算改变了IT,而大数据则改变了业务p云计算是大数据的IT基础，大数据须有云计算作为基础架构，才能高效运行p通过大数据的业务需求，为云计算的落地找到了实际应用大数据和云计算的关系大数据和云计算的关系大数据大数据云计算云计算大数据主要应用技术大数据主要应用技术HadoopHadoopHadoop作为新一代的架构和技术，因为有利于并行分布处理“大数据”而备受重视。Apache Hadoop 是一个用java语言实现的软件框架，在由大量计算机组成的集群中运行海量数据的分布式计算，它可以让应用程序支持上千个节点和PB级别的数据。 Hadoop是

3、项目的总称，主要是由分布式 存 储 （ H D F S ） 、 分 布 式 计 算（MapReduce）等组成 。优点：l可扩展：不论是存储的可扩展还是计算的可扩展都是Hadoop的设计根本。l经济：框架可以运行在任何普通的PC上。l可靠：分布式文件系统的备份恢复机制以及MapReduce的任务监控保证了分布式处理的可靠性。l高效：分布式文件系统的高效数据交互实现以及MapReduce结合Local Data处理的模式，为高效处理海量的信息作了基础准备。Hadoop系统工作原理Hadoop系统构架MapReduceHDFSHBasePigChuKwaHiveZooKeeperHadoopHad

4、oop体系架构体系架构Pig是一个基于Hadoop的大规模数据分析平台，Pig为复杂的海量数据并行计算提供了一个简易的操作和编程接口Chukwa是基于Hadoop的集群监控系统，由yahoo贡献hive是基于Hadoop的一个工具，提供完整的sql查询功能，可以将sql语句转换为MapReduce任务进行运行ZooKeeper：高效的，可扩展的协调系统,存储和协调关键共享状态HBase是一个开源的，基于列存储模型的分布式数据库HDFS是一个分布式文件系统。有着高容错性的特点，并且设计用来部署在低廉的硬件上，适合那些有着超大数据集的应用程序MapReduce是一种编程模型，用于大规模数据集（大于

5、1TB）的并行运算MapReducelMap：任务的分解lReduce：结果的汇总两大核心设计HDFSlNameNode：文件管理lDataNode：文件存储lClient：文件获取HadoopHadoop核心设计核心设计HDFSHDFS分布式文件系统分布式文件系统NameNode可以看作是分布式文件系统中的管理者，存储文件系统的meta-data，主要负责管理文件系统的命名空间，集群配置信息，存储块的复制。DataNode是文件存储的基本单元。它存储文件块在本地文件系统中，保存了文件块的meta-data，同时周期性的发送所有存在的文件块的报告给NameNode。 Client就是需要获取分

6、布式文件系统文件的应用程序。 HDFS是一个高度容错性的分布式文件系统，能提供高吞吐量的数据访问，非常适合大规模数据集上的应用。HDFSHDFS具体操作具体操作文件写入：1. Client向NameNode发起文件写入的请求2. NameNode根据文件大小和文件块配置情况，返回给Client它所管理部分DataNode的信息。3. Client将文件划分为多个文件块，根据DataNode的地址信息，按顺序写入到每一个DataNode块中。文件读取：1. Client向NameNode发起文件读取的请求2. NameNode返回文件存储的DataNode的信息。3. Client读取文件信息。

7、MapReduceMapReduce映射、化简编程模型映射、化简编程模型1. 根据输入数据的大小和参数的设置把数据分成splits, 每个split对于一个map线程。2. Split中的数据作为Map的输入， Map的输出一定在Map端。3. Map的输出到Reduce的输入的过程(shuffle过程)： 第一阶段：在map端完成内存-排序-写入磁盘-复制 第二阶段：在reduce端完成映射到reduce端分区-合并-排序4. Reduce的输入到Reduce的输出 最后排好序的key/value作为Reduce的输入，输出不一定是在reduce端。MapReduce是一种编程模型，用于大规

8、模数据集的并行运算。Map（映射）和Reduce（化简），采用分而治之思想，先把任务分发到集群多个节点上，并行计算，然后再把计算结果合并，从而得到最终计算结果。多节点计算，所涉及的任务调度、负载均衡、容错处理等，都由MapReduce框架完成，不需要编程人员关心这些内容。 HBASEHBASE分布式分布式数据存储数据存储HBase Hadoop Database，是一个高可靠性、高性能、面向列、可伸缩的分布式存储系统；HBase位于结构化存储层，HDFS为HBase提供了高可靠性的底层存储支持,MapReduce为HBase提供了高性能的计算能力，Zookeeper为HBase提供了稳定服务和

9、failover机制；Pig和Hive还为HBase提供了高层语言支持，使得在HBase上进行数据统计处理变的简单。内容提要内容提要 1. HADOOP体系架构 2. 基于HADOOP的大数据产品分析 3.3. HADOOPHADOOP部署部署HadoopHadoop主要开发厂商主要开发厂商大型企业和机构在寻求解决棘手的大数据问题时，往往会使用开源软件基础架构Hadoop的服务。由于Hadoop深受欢迎，许多公司都推出了各自版本的Hadoop，也有一些公司则围绕Hadoop提供解决方案。 Hadoop的发行版除了社区的Apache hadoop外，cloudera，IBM，ORACLE等都提供

10、了自己的商业版本。商业版主要是提供Hadoop专业的技术支持，这对一些大型企业尤其重要。ClouderaEMCIBMOracle内容提要内容提要 1. HADOOP体系架构 2. 基于HADOOP的大数据产品分析 3. HADOOP部署l3 Sept 2008l NEUSOFT SECRETHadoop分布式存储与计算框架目录目录l 分布式引例l Hadoop1.0架构l Hadoop1.0编程l Hadoop1.0弊端l Hadoop2.0架构l Hadoop2.0编程l Hadoop2.0部署分布式引例分布式引例概述概述分布式引例分布式引例问题问题场景场景假设配置完全相同的机器M01M0X

11、，并且每台机器都是1个双核CPU，5G硬盘。现有两个大小都是2G的文件file01和file02第一类问题：存储问题将file01和file02存入两台不同机器，但要求对外显示它们存于同一硬盘空间问题不考虑，现有一新文件file03，大小为6G，要求存入机器后对外显示依旧为一个完整文件第二类问题：计算问题在问题下，统计file01和file02这两个文件里，每个单词出现的次数第三类问题：可靠性问题假设用于解决上述问题的机器崩溃了，您如何保证数据不丢失分布式引例分布式引例常规解决方案常规解决方案存储问题思路：取M01，M02，直接将file01存M01，file02存M02file02存储问题存

12、储问题思路：file03拆成file03-a,file03-b将file03-a存M01，file03-b存M02存储问题第一类问题 存储若能将 M01与M02构建成统一空间？思考问题解决了？问题解决了？分布式解决方案分布式引例分布式引例常规解决方案常规解决方案计算问题思路：M02存的file02拷贝至M01，M01上执行计算计算问题第二类问题 计算M01执行计算解决了！并且简单实用好维护可若现在数据不是2G，而是2T机器不是2台，而是200台l 仅数据copy这一步就要花去一周时间！l 哪台服务器硬盘有400T？问题解决了吗？若每台机器都能l先本地计算(Map)l接着根据规则发不同机器(sh

13、uffle)l再合并计算(Reduce)那！分布式解决方案Map-Shuffle-Reduce分布式引例分布式引例常规解决方案常规解决方案可靠性问题思路：用最稳定最贵的机器可靠性问题第三类问题 可靠性可靠性问题硬件需提供高可靠性这点没错火山！地震！冗余！存很多备份分布式存储架构思路分布式引例分布式引例分布式解决方案分布式解决方案第一类问题 存储思路l取M01，M02，M99l构建master/slave体系分布式集群规定lM99：不做存储，统一管理M01，M02硬盘空间lM01，M02：存储具体文件存储问题由于硬盘空间大，可任意存存储问题由于硬盘连在一起，可任意存file01采用分布式-存储问

14、题全部解决Hadoop里Hdfs架构思想由于采用master/slave架构，很容易添加slave存储主节点存储主节点存储主节点分布式引例分布式引例分布式解决方案分布式解决方案第二类问题 计算data cstordatacstordatacstor 结果存至DFS结果存至DFS若能这样计算WordCount，那真的太好了存储主节点本地计算洗牌合并再计算存结果分布式存储分布式计算为此构建分布式计算见下页PPTl 先进行本机计算l 再按规则发不同机器l 进行合并计算WordCount解题思路网网 络络分布式计算架构思路分布式引例分布式引例分布式解决方案分布式解决方案第二类问题 计算思路l取M01，

15、M02，M98l构建master/slave体系分布式集群规定lM99：不做计算，统一管理M01，M02内存,CPUlM01，M02：具体计算任务计算问题本地计算洗牌合并计算由于采用master/slave架构，很容易添加slave计算主节点 结果存至DFS网网 络络计算主节点分分布布式存式存储储DFS结果存至DFS思路Hadoop-MapReduce计算思路Hadoop分布式计算架构两个结论，但两者不同：一个架构，一个计算思路cstordatadatacstordatadatal3 Sept 2008l NEUSOFT SECRET public void map(Object key, T

16、ext value, Context context ) StringTokenizer itr = new StringTokenizer(value.toString(); while (itr.hasMoreTokens() word.set(itr.nextToken(); context.write(word, 1); public void reduce(Text key, Iterable values, Context context) int sum = 0; for (IntWritable val : values) sum += val.get(); result.se

17、t(sum); context.write(key, result);public int getPartition(K key, V value,int numReduceTasks) return (key=cstor)? 0:1;Mapper简称:MPartition简称:PPReduce简称:R分布式引例分布式解决方案定义WC所需三个函数：MPR分布式引例分布式引例分布式解决方案分布式解决方案分布式存储构建 + 分布计算构建+使用MR实现WC 整体流程data cstordatacstordatacstor 结果存至DFS结果存至DFS存储主节点本地计算洗牌合并再计算存结果分布式存储分

18、布式计算网网 络络M PRM P RM P Rfile01可靠分布式存储MMM-file01可靠分布式计算第三类问题 可靠性分布式引例分布式引例分布式解决方案分布式解决方案M-file01MM03再死就把它加入黑名单冗余可靠的分布式存储可靠的分布式计算思路：冗余l3 Sept 2008l NEUSOFT SECRETHadoop1.0-概念由来组成概念由来组成组成Hadoop1.0架构架构HDFS架构架构客户端namenodeb-0b-1datanode1b-1b-2datanode2b-2b-0datanode3b-0b-1b-2datanoden元数据操作数据块操作读写fsimageb-1

19、b-0File01分为三个块：b-0,b-1,b-2客户端jobtrackertasktracker1tasktracker2tasktracker3Tasktracker n提交任务分配并监管任务JVMJVMHadoop1.0架构架构分布式计算架构分布式计算架构public class Write public static void main(String args) throws IOException Configuration conf = new Configuration(); /实例化配置文件类Path inFile = new Path(“/user/joe/myfile”)

20、; /定义欲写文件位置FileSystem hdfs = FileSystem.get(conf); /获取文件系统接口FSDataOutputStream outputStream = hdfs.create(inFile); /创建文件流 outputStream.writeUTF(“china cstor cstor china”); /写入字符流outputStream.flush(); /强制流写入磁盘outputStream.close(); /关闭输出流 public class Read public static void main(String args) throws I

21、OException Configuration conf = new Configuration();Path inFile = new Path(/user/joe/myfile);FileSystem hdfs = FileSystem.get(conf);FSDataInputStream inputStream = hdfs.open(inFile);System.out.println(myfile: +inputStream.readUTF();inputStream.close(); WriteReadHadoop1.0编程编程Hdfs编程编程读写HdfsHdfs编程已在上页给

22、出编程已在上页给出Hadoop1.0编程编程 MapReduce编程编程 概述概述InputFormatMapCombinePartitionerGroupingComparatorReduceOutputFormatRecordReaderInputSplitRecordWriterOutputCommiter抽象类或接口抽象类或接口MR下默认实现下默认实现InputFormatTextInputFormatRecordReaderLineRecordReaderInputSplitFileSplitMapIdentityMapperCombineIdentityReducerPartiti

23、onerHashPartitionerGroupingComparator神奇的地方,参见SecondarySort.javaReduceIdentityReducerOutputFormatFileOutputFormatRecordWriterLineRecordWriterOuputCommitterFileOutputCommintterHadoop1.0编程 MapReduce编程编程过程问我们为什么每次只需写mapper和reduce因为其他都默认指定了！public class WordCount public static void main(String args) thro

24、ws Exception /主方法，函数入口 Configuration conf = new Configuration(); /实例化配置文件类 Job job = new Job(conf, WordCount); /实例化Job类 job.setInputFormatClass(TextInputFormat.class); /指定使用默认输入格式类 TextInputFormat.setInputPaths(job, inputPaths); /设置待处理文件的位置 job.setJarByClass(WordCount.class); /设置主类名 job.setMapperCla

25、ss(TokenizerMapper.class); /指定使用上述自定义Map类 job.setMapOutputKeyClass(Text.class); /指定Map类输出的，K类型 job.setMapOutputValueClass(IntWritable.class); /指定Map类输出的，V类型 job.setPartitionerClass(HashPartitioner.class); /指定使用默认的HashPartitioner类 job.setReducerClass(IntSumReducer.class); /指定使用上述自定义Reduce类 job.setNum

26、ReduceTasks(Integer.parseInt(numOfReducer); /指定Reduce个数 job.setOutputKeyClass(Text.class); /指定Reduce类输出的,K类型 job.setOutputValueClass(Text.class); /指定Reduce类输出的,V类型 job.setOutputFormatClass(TextOutputFormat.class); /指定使用默认输出格式类 TextOutputFormat.setOutputPath(job, outputDir); /设置输出结果文件位置 System.exit(j

27、ob.waitForCompletion(true) ? 0 : 1); /提交任务并监控任务状态 Hadoop1.0编程编程之之MapReduce编编程程之之WordCount实实例例入口类入口类/*定义map类，一般继承自Mapper类，里面实现读取单词，写出*/ public class TokenizerMapper extends Mapper private final static IntWritable one = new IntWritable(1); private Text word = new Text(); /map方法，划分一行文本，读一单词写出一个 public

28、void map(Object key, Text value, Context context)throws IOException, InterruptedException StringTokenizer itr = new StringTokenizer(value.toString(); while (itr.hasMoreTokens() word.set(itr.nextToken(); context.write(word, one);/写出 Hadoop1.0编程之MapReduce编程之WordCount实例Mapper类/*定义类MyPartitioner，cstor到r

29、educe 0，data到reduce 1*/public class MyPartitioner extends Partitioner public int getPartition(Text key,IntWritable value,int numReduceTasks) return return (key.toString=cstor)? 0:1; Hadoop1.0编程之MapReduce编程之WordCount实例Partitioner类/*默认Partitioner，效率高，能将数据均匀分散到各reduce，但还是可能会出现数据倾斜*/public class HashPar

30、titioner extends Partitioner public int getPartition(K key, V value, int numReduceTasks) return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks; /*定义reduce类，对相同的单词，把他们中，VList值全部相加*/public static class IntSumReducer extends Reducer private IntWritable result = new IntWritable(); public void

31、 reduce(Text key, Iterable values,Context context) throws IOException, InterruptedException int sum = 0; for (IntWritable val : values) sum += val.get();/相当于，将两个1相加。 result.set(sum); /写出这个单词，和这个单词出现次数 context.write(key, result); Hadoop1.0编程之MapReduce编程之WordCount实例Reduce类Hadoop1.0编程编程MapReduce编程之编程之W

32、ordCount逻辑过程逻辑过程data cstordatacstordatacstor 结果存至DFS结果存至DFS存储主节点本地计算洗牌合并再计算存结果分布式存储分布式计算网网 络络M PRWordCount逻辑执行过程其实就是它MapReduceProgramJobClinetJobRrackerMapTaskorReuceTaskTaskRrackerChildClient JVMClient nodejobtracker nodetasktracker node2.Get new job ID1.Runjob4.Submit job5.Initializejob3.Copy jobr

33、esources7.Heartbeat6.Retrieve input splits9.Launch10.Run8.Retrieve job resources框架执行WordCount过程之大过程JobConfMapperReducerJobClientJob jar UrlJobInProgressTaskInProgress1TaskInProgress2TaskInputDatasJob JarOutputDatasTaskTaskInProgress1TaskRunnerMapperTaskMapperCombinerLocalFSReducerReducerTaskApplicat

34、ion1.Submit JobJobTracker2.Add New Job3.Create Task4.Launch Task5.Transfer Launch Task Action6.Add New Task7.New Runner8.Create Child11.Create Child9.InvokeMapper11.InvokeReducer10.Invoke CombinerProcessInvokeDataTaskTrackerHDFS框架执行WC之细过程Hadoop1.0弊端弊端分布式计算HDFS工作量多而杂l3 Sept 2008l NEUSOFT SECRETHadoop

35、2.0架构架构概念由来组成概念由来组成组成Hadoop2.0架构架构Hdfs架构架构和原Hdfs架构相同但可配置两个主节点namenode1namenode2Zookeeper集群集群(至少至少3台台)JournalNode集群集群(至少至少3台台)datanode1datanode2datanoden采用这种机制可以实现双采用这种机制可以实现双NameNode自动、瞬间、热切换自动、瞬间、热切换原原JobTracker功能功能ResourceManagerApplicationMaster原TaskTrackerNodeManagerSlave端Master端简单的说就是简单的说就是：放权放

36、权Hadoop2.0架构架构Yarn架构架构Yarn架构思路架构思路：将原：将原JobTracker三三大功能拆分大功能拆分原原JobTracker功能功能ResourceManagerApplicationMasterMaster端端Hadoop2.0架构架构Yarn架构架构Yarn架构思考问？l 既然MR逻辑由它实现l 任务也由它监管那！程序可不可以不是MR类型的业务逻辑答，可以！Yarn编程，完全突破Hadoop1.0编程局限开发不同的ApplicationMaster，可实现不同业务逻辑如无数据只计算型，M型，MRR型Hadoop2.0架构架构Yarn架构架构Yarn架构框架执行Yar

37、n任务时流程Hadoop2.0架构架构Yarn架构架构强调l 任务逻辑完全由AppMster决定，可以不是MR逻辑l AppMster是RM启动的l AppMster负责任务执行和监管Client端向RM提交任务RM首先启动负责本此任务的AppMsterAppMster负责本此任务处理逻辑 它首先与RM协商申请资源 再NM协商启动Container执行任务AppMster完成任务后向RM注销自己Hadoop2.0编程编程概述概述Hadoop2.0编程编程Yarn编程编程Yarn编程步骤编程步骤首先编写首先编写Client端端其次编写其次编写AppMster端端Hadoop2.0编程编程Yarn

38、编程实例编程实例DistributedShell概述概述DistributedShell是Yarn自带的一个应用程序编程实例，相当于Yarn编程中的“Hello World”，它的功能是并行执行用户提交的Shell命令或Shell脚本。定位获取从hadoop官方网站下载hadoop-2.2.0-src.tar.gz（hadoop源码包）并解压后，依次进入hadoop-yarn-projecthadoop-yarnhadoop-yarn-applications，下面就是Yarn自带的两个Yarn编程实例。DistributedShell中主要包含两个类Client和ApplicationMas

39、ter。其中Client主要向RM提交任务。ApplicationMaster则需向RM申请资源，并与NM协商启动Container完成任务。讲解执行$hadoop jar /usr/lib/hadoop-yarn/hadoop-yarn-applications-distributedshell.jar org.apache.hadoop.yarn.applications.distributedshell.Client -jar /usr/lib/hadoop-yarn/hadoop-yarn-applications-distributedshell.jar -shell_command

40、 /bin/date -num_containers 100Client端代码：见下页pptApplicationMaster端代码：见下页ppt源码YarnClient yarnClient = YarnClient.createYarnClient(); /新建Yarn客户端yarnClient.start(); /启动Yarn客户端YarnClientApplication app = yarnClient.createApplication(); /获取提交程序句柄ApplicationSubmissionContext appContext = app.getApplicationS

41、ubmissionContext(); /获取上下文句柄ApplicationId appId = appContext.getApplicationId(); /获取RM分配的appIdappContext.setResource(capability); /设置任务其他信息举例appContext.setQueue(amQueue); /设置作业所属对了appContext.setPriority(priority); /设置作业优先级/实例化ApplicationMaster对应的ContainerContainerLaunchContext amContainer = Records.

42、newRecord(ContainerLaunchContext.class);amContainer.setCommands(commands); /参数commands为用户预执行的shell命令appContext.setAMContainerSpec(amContainer); /指定ApplicationMaster的ContaineryarnClient.submitApplication(appContext); /提交作业Hadoop2.0编程编程Yarn编程编程DistributedShell之之Client类分析类分析AMRMClientAsync amRMClient =

43、 AMRMClientAsync.createAMRMClientAsync(1000, allocListener); /新建RM代理amRMClient.init(conf); /设置配置文件amRMClient.start(); /获取RM代理amRMClient.registerApplicationMaster(appMasterHostname, appMasterRpcPort,appMasterTrackingUrl); /向RM注册NMClientAsync nmClientAsync = new NMClientAsyncImpl(containerListener); /

44、新建NM代理nmClientAsync.init(conf); /初始化NM端连接nmClientAsync.start(); /获取NM代理for (int i = 0; i numTotalContainers; +i) /向RM申请资源ContainerRequest containerAsk = setupContainerAskForRM();amRMClient.addContainerRequest(containerAsk);numRequestedContainers.set(numTotalContainers); /设置任务所需全部资源量（以Containers为单位）C

45、ontainerLaunchContext ctx = Records.newRecord(ContainerLaunchContext.class); /设置Container上下文ctx.setCommands(commands); /设置NM上所启动的Container需完成的任务nmClientAsync.startContainerAsync(container, ctx); /要求NM启动containernmClientAsync.stop(); /containerListener汇报此NM完成任务后，关闭此NMamRMClient.unregisterApplicationM

46、aster(appStatus, appMessage, null); /向RM注销amRMClient.stop(); /任务完成，停止RM端代理Hadoop2.0编程Yarn编程(DistributedShell之ApplicationMaster类分析)Hadoop2.0编程编程Yarn编程编程MapReduce编程编程Yarn下的MapReduce和MRv1里的MapReduce相同，是Google-MapReduce思想的实现，不同的是，Yarn下编程不只有MR这一种模式，如果我们将MR看成一种编程模板，那么MRv1下，则只有这一种模板，可是Yarn下可以有各种各样的编程模板，如无数据只耗CPU计算形式，如MR形式，尽管编程形式多了，但MR依旧是分布式编程首选模板。定位同DistributedShell相同，编写MapReduce程序基本步骤依旧是： 编写Client：默认实现类MRClientService 编写ApplicationMaster：默认实现类MRAppMasterHadoop开发人员为MapReduce编程模型开发了Client和ApplicationMaster的默认实现了MRClientServe和MRAppMaster，MRv1版本的MapReduce代码在Yarn里是兼容的，用户只需重新编译一下