大数据培训.pptx

大数据培训.pptx

《大数据培训.pptx》由会员分享，可在线阅读，更多相关《大数据培训.pptx（143页珍藏版）》请在冰点文库上搜索。

大数据培训,大数据概念Hadoop介绍HDFSMAPREDUCE编程HIVEFLUMESTORMZOOKEEPER,培训内容,大数据定义大小超过常规的数据库工具获取、存储、管理和分析能力的数据集事务所涉及的资料量规模巨大到无法通过目前主流软件工具，在合理时间内达到撷取、管理、处理、并整理成为帮助企业经营决策更积极目的的资讯。

四大特征：

大量（Volume）存储大；计算量大；多样（Variety）来源多；格式多；快速（Velocity）增长速度快处理速度要求快价值（Value）浪里淘沙却又弥足珍贵,大数据概念,大数据趋势,BigData,People,Devices,Sensors,移动互联网MobileInternet,物联网InternetofThings,新量级、新处理模式、新企业智能,大数据趋势,对企业数据处理的挑战,对企业数据处理的挑战,每天几百GB、几TB的资料，且持续成长中,储存Storing,在收数据的同时做必要的前置处理（pre-processing），并区分数据处理的优先等级（prioritizing）,计算Processing,如何有效的避免因硬件毁坏所导致的资料损毁,管理Managing,如何从中挖掘出所关注事件的pattern或behavior,分析Analyzing,5,典型互联网大数据应用技术体系,典型互联网大数据应用技术体系,数据存储、计算,规则、业务处理,数据采集,数据服务,高效、智能的数据采集技术,高效数据仓库技术（ETL）,实时计算技术,大数据存储技术,大数据计算技术,机器学习,关键技术,数据分析&可视化技术,数据共享技术,大数据生态圈,Hadoop介绍,Hadoop概况,Hadoop概况,Apache开源项目源于Lucene项目的一部分,2006.1成为子项目,现为Apache顶级项目之一Google的三篇论文MapReduce,GFS,BigTableYahoo!

是最主要的源代码贡献者,其他贡献者:

Powerset,Facebook等已知为接近150家的大型组织实际使用:

Yahoo!

Amazon,EBay,AOL,Google,IBM,Facebook,Twitter,Baidu,Alibaba,Tencent,（http:

/wiki.apache.org/hadoop/PoweredBy）Hadoop核心功能高可靠性,高效率的分布式文件系统一个海量数据处理的编程框架Hadoop特点可扩展性:

Petabytes（1015Bytes）级别的数据量,数千个节点经济性:

利用商品级（commodity）硬件完成海量数据存储和计-高效率：

通过分发数据，可以在数据所在的节点上并行处理，使得处理效率变快可靠性:

在大规模集群上提供应用级别的可靠性,HadoopVS.RDB,Hadoop特性,Hadoop特性,传统并行计算架构,并行计算+分布式存储,运算,储存,传统储存架构,计算与存储一体，计算向数据靠拢，高效专用存储模式为程序员屏蔽通性、并发、同步与一致性等问题任务之间无依赖（share-nothing），具有高系统延展性（scale-out）,HDFS,Hadoop1.XHDFS架构图,NameSpaceState,BlockMap,SecondaryNameNode,Client,DataNode,DataNode,DataNode,DataNode,NameNode,Heartbeat&Blockreport,本地磁盘,元数据,HDFS设计原则,文件以块（block）方式存储每个块带下远比多数文件系统来的大通过副本机制提高可靠度和读取吞吐量每个区块缺省分到三台DataNode上master（NameNode）来协调存储元数据（metadata）客户端对文件没有缓存机制（Nodatacaching）,NameNode（NN）,NameNode主要功能提供名称查询服务NameNode保存metadate信息包括文件owership和permissions文件包含哪些块Block保存在哪个DataNode（由DataNode启动时上报）NameNode的metadate信息在启动后会加载到内存metadata存储到磁盘文件名为”fsimage”Block的位置信息不会保存到fsimage,NameNode,块存储结构,metadata物理存储结构,DataNode（DN）,保存Block启动DN线程的时候会向NN汇报block信息通过向NN发送心跳保持与其联系（3秒一次），如果NN10分钟没有收到DN的心跳，则认为其已经lost，并copy其上的block到其它DN,Block的副本放置策略,第一个副本：

放置在上传文件的DN；如果是集群外提交，则随机挑选一台磁盘不太满，CPU不太忙的节点第二个副本：

放置在于第一个副本不同的机架的节点上第三个副本：

第二个在同一个机架，随机放在不同的node中。

更多副本：

随机节点,再说Block,设置一个Block64MB，如果上传文件小于该值，仍然会占用一个Block的命名空间（NameNodemetadata），但是物理存储上不会占用64MB的空间Block大小和副本数由Client端上传文件到HDFS时设置，其中副本数可以变更，Block是不可以再上传后变更的,数据损坏（corruption）处理,当DN读取block的时候，它会计算checksum如果计算后的checksum，与block创建时值不一样，说明该block已经损坏。

client读取其它DN上的block；NN标记该块已经损坏，然后复制block达到预期设置的文件备份数DN在其文件创建后三周验证其checksum,HDFS文件权限,21,与Linux文件权限类似r:

read;w:

write;x:

execute，权限x对于文件忽略，对于文件夹表示是否允许访问其内容如果Linux系统用户test使用hadoop命令创建一个文件，那么这个文件在HDFS中owner就是testHDFS的权限目的：

阻止好人错错事，而不是阻止坏人做坏事。

HDFS相信，你告诉我你是谁，我就认为你是谁,ACL权限控制,setfacl-mu:

cheng:

wrtest.txtyarmrmadmin-refreshServiceAclhadoopdfsadmin-refreshServiceAcl,HDFS文件写入,HDFS文件读取,MapReduce,Map/Reduce,Map/Reduce,什么是Map/Reduce一种高效,海量的分布式计算编程模型海量:

相比于MPI,Map处理之间的独立性使得整个系统的可靠性大为提高.高效:

用调度计算代替调度数据!

分布式操作和容错机制由系统实现,应用级编程非常简单.计算流程非常类似于简单的Unixpipe:

Pipe:

catinput|grep|sort|uniq-coutputM/R:

Input|map|shuffle&sort|reduce|output多样的编程接口:

Javanativemap/reduce可以操作M/R各细节Streaming利用标准输入输出模拟以上pipelinePig只关注数据逻辑，无须考虑M/R实现,MapReduce编程接口,1）迭代（iteration）。

遍历输入数据，并将之解析成key/value对。

2）将输入key/value对映射（map）成另外一些key/value对。

3）依据key对中间数据进行分组（grouping）。

4）以组为单位对数据进行归约（reduce）。

5）迭代。

将最终产生的key/value对保存到输出文件中。

MapReduce编程模型,用户定义一个map函数来处理一个key/value对以生成一批中间的key/value对，再定义一个reduce函数将所有这些中间的有着相同key的values合并起来。

Map/reduce计算流程,用Java进行Map/Reduce编程,MapReduce,单一Reduce处理,多个Reduce处理,无Reduce处理,MapReduce,MapReduce,物理上,MapReduce,MapReduce处理流程,MapReduce,MapReduce,MapReduce,MapReduce,MapReduce,MapReduce,MapReduce,MapReduce,MapReduce,MapReduce,MapReduce,JobTracker一直在等待JobClient提交作业TaskTracker每隔3秒向JobTracker发送心跳heartbeat询问有没有任务可做，如果有，让其派发任务给它执行这是一道pull过程:

slave主动向master拉生意,MapReduce,特点：

Fault-tolerant容错，很重要！

M/RFailuresTaskfailsTryagain?

Tryagainsomewhereelse?

Reportfailure只有当map处理全部结束后，reduce过程才能够开始,Hadoop开发注意点,1.Hadoop将运行一次Map/Reduce作业叫做运行一个Job2.Hadoop需要计算的源数据都存储在HDFS中。

3.Map阶段计算结果存储在本地文件系统中。

4.Hadoop最终计算的结果也存储在HDFS中。

5.Map/Reduce框架的运作完全基于对，即数据的输入是一批对，生成的结果也是一批对。

6.默认情况下，Key与value之间用t分隔。

WordCount：

MapReduce版HelloWorld，,WordCount代码解析-MappublicstaticclassMapextendsMapReduceBaseimplementsMapperprivatefinalstaticIntWritableone=newIntWritable

（1）;privateTextword=newText（）;/map函数publicvoidmap（LongWritablekey,Textvalue,OutputCollectoroutput,Reporterreporter）throwsIOExceptionStringline=value.toString（）;StringTokenizertokenizer=newStringTokenizer（line）;while（tokenizer.hasMoreTokens（）word.set（tokenizer.nextToken（）;output.collect（word,one）;,WordCount代码解析-ReducepublicstaticclassReduceextendsMapReduceBaseimplementsReducer/reduce函数publicvoidreduce（Textkey,Iteratorvalues,OutputCollectoroutput,Reporterreporter）throwsIOExceptionintsum=0;while（values.hasNext（）sum+=values.next（）.get（）;output.collect（key,newIntWritable（sum）;,WordCount代码解析Main主函数publicstaticvoidmain（Stringargs）throwsExceptionJobConfconf=newJobConf（WordCount.class）;conf.setJobName（wordcount）;conf.setOutputKeyClass（Text.class）;conf.setOutputValueClass（IntWritable.class）;conf.setMapperClass（Map.class）;conf.setCombinerClass（Reduce.class）;conf.setReducerClass（Reduce.class）;conf.setInputFormat（TextInputFormat.class）;conf.setOutputFormat（TextOutputFormat.class）;FileInputFormat.setInputPaths（conf,newPath（args0）;/hdfs文件路径FileOutputFormat.setOutputPath（conf,newPath（args1）;/结果输出路径JobClient.runJob（conf）;,海量数据存储,计算结果,Combiner和Partitioner,基于Map和Reduce的并行计算模型,InjectPoints,Inputjob.setInputFormatClass（）SplitRecordReaderMapjob.setMapperClass（）Combine*job.setCombinerClass（）Shufflingjob.setPartitionerClass（）Sortjob.setSortComparatorClass（）Groupingjob.setGroupingComparatorClass（）Reducejob.setReducerClass（）Outputjob.setOutputFormatClass（）,Combiners和Partitioner编程,Combiners的作用：

每一个map可能会产生大量的输出，combiner的作用就是在map端对输出先做一次合并，以减少传输到reducer的数据量，1）combiner最基本是实现本地key的聚合，对map输出的key排序，value进行迭代。

如下所示：

map:

（K1,V1）list（K2,V2）combine:

（K2,list（V2）list（K2,V2）reduce:

（K2,list（V2）list（K3,V3）2）combiner还具有类似本地的reduce功能.例如hadoop自带的wordcount的例子和找出value的最大值的程序，combiner和reduce完全一致。

如下所示：

map:

（K1,V1）list（K2,V2）combine:

（K2,list（V2）list（K3,V3）reduce:

（K3,list（V3）list（K4,V4）3）如果不用combiner，那么，所有的结果都是reduce完成，效率会相对低下。

使用combiner，先完成的map会在本地聚合，提升速度。

4）对于hadoop自带的wordcount的例子，value就是一个叠加的数字，所以map一结束就可以进行reduce的value叠加，而不必要等到所有的map结束再去进行reduce的value叠加。

注意：

combiner使用的合适，可以在满足业务的情况下提升job的速度，如果不合适，则将导致输出的结果不正确。

Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。

所以从我的想法来看，Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。

比如累加，最大值等。

Combiner,Combiners分析,假设有两个map。

第一个map的输出为：

（1950,0）（1950,20）（1950,10）第二个map输出为：

（1950,25）（1950,15）（1950,30）Reduce函数被调用是，输入如下：

（1950，0,20,10,25,15,30）因为30是最大的值，所以输出如下：

（1950,30）如果我们使用combiner：

那么reduce调用的时候传入的数据如下：

（1950，20,30）-（1950,30）用表达式表示为：

Max（0,20,10,25,15,30）=max（max（0,20,10）,max（25,15,30）=max（20,30）=30,刚才我们是计算最大值可以使用Combiners能提高效率。

如果我们要是求平均值呢？

Avg（0,20,10,25,15,30）=15如果使用Combiner会得到什么样的结果呢？

第一个map输出为：

avg（0,20,10）=10第二个map输出为：

Avg（25,15,30）=23输入到reduce出来的结果为：

Avg（10,23）=17.517.5和15？

所以：

使用combiner一定要注意。

Partitioner,Partition主要作用就是将map的结果发送到相应的reduce。

这就对partition有两个要求：

1）均衡负载，尽量的将工作均匀的分配给不同的reduce。

2）效率，分配速度一定要快。

mapreduce提供的Patitioner,Partitioner是partitioner的基类，如果需要定制partitioner也需要继承该类。

2.HashPartitioner是mapreduce的默认partitioner。

计算方法是whichreducer=（key.hashCode（）&Integer.MAX_VALUE）%numReduceTasks，得到当前的目的reducer。

BinaryPatitioner继承于Partitioner，是Partitioner的偏特化子类。

该类提供leftOffset和rightOffset，在计算whichreducer时仅对键值K的rightOffset，leftOffset这个区间取hash。

Whichreducer=（hash&Integer.MAX_VALUE）%numReduceTasksKeyFieldBasedPartitioner也是基于hash的个partitioner。

和BinaryPatitioner不同，它提供了多个区间用于计算hash。

当区间数为0时KeyFieldBasedPartitioner退化成HashPartitioner。

5.TotalOrderPartitioner这个类可以实现输出的全排序。

不同于以上3个partitioner，这个类并不是基于hash的。

在下一节里详细的介绍totalorderpartitioner。

mapreduce的类型与格式,Hadoop0.20.x之前的APIHadoop的MapReduce中，map和reduce函数遵循如下常规格式：

map:

（K1,V1）list（K2,V2）reduce:

（K2,list（V2）list（K3,V3）Mapper的接口：

publicinterfaceMapperextendsJobConfigurable,Closeablevoidmap（K1key,V1value,OutputCollectoroutput,Reporterreporter）throwsIOException;Reduce的接口：

publicinterfaceReducerextendsJobConfigurable,Closeablevoidreduce（K2key,Iteratorvalues,OutputCollectoroutput,Reporterreporter）throwsIOException;/outputCollector是为了输出key/value对，/Reporter是用来更新计数和状态信息。

Hadoop0.20.x之后的APIHadoop的MapReduce中，map和reduce函数遵循如下常规格式：

map:

（K1,V1）list（K2,V2）reduce:

（K2,list（V2）list（K3,V3）Mapper的接口：

protectedvoidmap（KEYkey,VALUEvalue,Contextcontext）throwsIOException,InterruptedExceptionReduce的接口：

protectedvoidreduce（KEYkey,Iterablevalues,Contextcontext）throwsIOException,InterruptedException/Context是上下文对象，这里Context等同于OutputCollector和Reporter两个函数的功能。

mapreduce的数据类型与java类型对应关系,Writable接口,1.对Java中的int型进行封装那么就是hadoop中的IntWritable类在写程序时可以把IntWritable可以看着是int类型，它实现了WritableComparable接口。

WritableComparable又是Writable、parable接口的子接口。

2.Writable类对所有的Java基本类型进行封装：

如：

boolean-BooleanWritable;Byte-ByteWritable3.我们可以自定义Writable接口，来编写更复杂的结构的类。

核心：

hadoop有自己一套的I/O机制。

I/O类都必须实现Writable接口。

mapreduce驱动默认的设置,DistributedCache,代码,HIVE,Hive,由Facebook开发架构于Hadoop之上,设计用来管理结构化数据的中间件以MapReduce为执行环境数据储存于HDFS上Metadata储存于RDMBS中Hive的设计原则采用类SQL语法扩充性Types,Functions,Formats,Scripts性能与平水扩展能力兼具,HiveSQLlikeHadoopDatabase,Driver（compiler,optimizer,executor）,metastore,DataNode,DataNode,DataNode,DataNode,HadoopCluster,M/R,M/R,M/R,M/R,WebUI,CLI,JDBCODBC,CreateM/RJob,Hive中所有的数据都存储在HDFS中，Hive中包含以下数据模型：

Table，ExternalTable，Partition，Bucket。

Hive中的Table和数据库中的Table在概念上是类似的，每一个Table在Hive中都有一个相应的目录存储数据。

例如，一个表xiaojun，它在HDFS中的路径为：

/warehouse/xiaojun，其中，wh是在hive-site.xml中由$hive.metastore.warehouse.dir指定的数据仓库的目录，所有的Table数据（不包括ExternalTable）都保存在这个目录中。

Partition对应于数据库中的Partition列的密集索引，但是Hive中Partition的组织方式和数据库中的很不相同。

在Hive中，表中的一个Partition对应于表下的一个目录，所有的Partition的数据都存储在对应的目录中。

例如：

xiaojun表中包含dt和city两个Partition，则对应于dt=20100801,ctry=US的HDFS子目录为：

/warehouse/xiaojun/dt=20100801/ctry=US；对应于dt=20100801,ctry=CA的HDFS子目录为；/warehouse/xiaojun/dt=20100801/ctry=CA,Buckets对指定列计算hash，根据hash值切分数据，目的是为了并行，每一个Bucket对应一个文件。

将user列分散至32个bucket，首先对user列的值计算hash