《hadoop实战培训-传智播客.ppt》由会员分享，可在线阅读，更多相关《hadoop实战培训-传智播客.ppt（56页珍藏版）》请在三一办公上搜索。
1、,高级软件人才实作培训专家!,Hahoop 培训,讲师：刘刚,北京传智播客教育,2,海量数据处理平台架构,一般网站把用户的访问行为记录以apach 日志的形式记录下来了，这些日志中包含了下面一些 关键字段：client_ip,user_id,access_time,url,referer,status,page_size,agent因为需要统一对数据进行离线计算，所以常常把它们全部移到同一个地方。简单算了一下：（1）网站请求数：1kw/天（2）每天日志大小：450Byte/行*1kw=4.2G，（3）日志存储周期：2 年 一天产生4.5G 的日志，2 年需要4.2G*2*365=3.0T解决方
2、案：为了方便系统命令查看日志，不压缩，总共需要3.0T 的空间，刚好有一些2U 的服务器，每台共1T 的磁盘空间。为了避免系统盘坏掉影响服务器使用，对系统盘做了raid1。为了避免其他存放数据的盘坏掉导致数据无法恢复，对剩下的盘做了raid5。所有的数据都汇聚到这几台LogBackup 服务器上来了。,北京传智播客教育,3,有了LogBackup 服务器，离线统计就可以全部在这些服务器上进行了。在这套架构上，用wc、grep、sort、uniq、awk、sed 等系统命令，完成了很多的统计需求，比如统计访问频率较高的client_ip，某个新上线的的页面的referer 主要是哪些网站。当业务
3、的迅猛发展，网站流量爆发增长，产品经理如果想从中获取更多的用户特征和用户信息，就需要我们这些数据分析人员从不同的日志中找到令他们满意的答案。如果（1）日志总行数：10 亿/天（2）每天日志大小：450Byte/行*10 亿=420G，（3）日志种类：5 种,北京传智播客教育,Hadoop能解决哪些问题,海量数据需要及时分析和处理。海量数据需要深入分析和挖掘。数据需要长期保存问题：磁盘IO成为一种瓶颈,而非CPU资源。网络带宽是一种稀缺资源硬件故障成为影响稳定的一大因素,北京传智播客教育,Hadoop在国内的情景,奇虎360：Hadoop存储软件管家中软件，使用CDN技术将用户请求引到最近的Ha
4、doop集群并进行下载京东、百度：存储、分析日志、数据挖掘和机器学习（主要是推荐系统）广告类公司：存储日志，通过协调过滤算法为客户推荐广告Yahoo：垃圾邮件过滤华为：云计算平台Facebook：日志存储，实时分析某公安部项目：网民QQ聊天记录与关联人调查系统，使用Hbase实现某学校：学生上网与社会行为分析，使用hadoop淘宝、阿里：国内使用Hadoop最深入的公司，整个Taobao和阿里都是数据驱动的,北京传智播客教育,Hadoop开发人员市场需求和待遇,北京传智播客教育,Hadoop在国内的人才储备,北京传智播客教育,Hadoop介绍,1）作者：Doug Cutting2)用Java编
5、写的开源系统，能够安排在大规模的计算平台上，从而长进计算效率。3）Nutch搜索引擎里面的一个模块。4）受Google三篇论文的启发-MapReduce GFS Bigtable5）google hadoop mapreduce mapreduce GFS HDFS Bigtable Hbase,北京传智播客教育,Hadoop生态系统介绍,Hbase 1）Nosql数据库，Key-Value存储 2）最大化利用内存HDFS 1）hadoop distribute file system分布式文件系统 2）最大化利用磁盘MapReduce 1）编程模型，主要用来做数据的分析 2）最大化利用CPU
6、,北京传智播客教育,HDFS篇,北京传智播客教育,HDFS设计原则,文件以块(block)方式存储每个块带下远比多数文件系统来的大(预设64M)通过副本机制提高可靠度和读取吞吐量每个区块至少分到三台DataNode上单一 master(NameNode)来协调存储元数据(metadata)客户端对文件没有缓存机制(No data caching),北京传智播客教育,HDFS系统结构,北京传智播客教育,NameNode(NN),NameNode主要功能提供名称查询服务，它是一个jetty服务器NameNode保存metadate信息包括文件owership和permissions文件包含哪些块B
7、lock保存在哪个DataNode（由DataNode启动时上报）NameNode的metadate信息在启动后会加载到内存metadata存储到磁盘文件名为”fsimage”Block的位置信息不会保存到fsimage,NameNode,块存储结构,metadate物理存储结构,DataNode（DN）,保存Block启动DN线程的时候会向NN汇报block信息通过向NN发送心跳保持与其联系（3秒一次），如果NN 10分钟没有收到DN的心跳，则认为其已经lost，并copy其上的block到其它DN,Block的副本放置策略,第一个副本：放置在上传文件的DN；如果是集群外提交，则随机挑选一台
8、磁盘不太满，CPU不太忙的节点第二个副本：放置在于第一个副本不同的机架的节点上第三个副本：与第二个副本相同集群的节点更多副本：随机节点,北京传智播客教育,再说Block,设置一个Block 64MB，如果上传文件小于该值，仍然会占用一个Block的命名空间（NameNode metadata），但是物理存储上不会占用64MB的空间Block大小和副本数由Client端上传文件到HDFS时设置，其中副本数可以变更，Block是不可以再上传后变更的,北京传智播客教育,数据损坏(corruption)处理,当DN读取block的时候，它会计算checksum如果计算后的checksum，与block
9、创建时值不一样，说明该block已经损坏。client读取其它DN上的block；NN标记该块已经损坏，然后复制block达到预期设置的文件备份数DN在其文件创建后三周验证其checksum,19,HDFS文件权限,与Linux文件权限类似r:read;w:write;x:execute，权限x对于文件忽略，对于文件夹表示是否允许访问其内容如果Linux系统用户zhangsan使用hadoop命令创建一个文件，那么这个文件在HDFS中owner就是zhangsanHDFS的权限目的：阻止好人错错事，而不是阻止坏人做坏事。HDFS相信，你告诉我你是谁，我就认为你是谁,HDFS文件读取,HDFS文
10、件写入,北京传智播客教育,HDFS文件存储,两个文件，一个文件156M，一个文件128在HDFS里面怎么存储？-Block为64MB-rapliction默认拷贝3份,北京传智播客教育,HDFS文件存储结构,北京传智播客教育,HDFS开发常用命令,创建一个文件夹？上传一个文件？删除一个文件和文件夹？查看一个文件夹里面有哪些文件？查看某个文件的内容？,北京传智播客教育,Hadoop管理员常用命令,hadoop job list#列出正在运行的Jobhadoop job kill#kill jobhadoop fsck/#检查HDFS块状态，是否损坏hadoop fsck/-delete#检查HD
11、FS块状态，删除损坏块hadoop dfsadmin report#检查HDFS状态，包括DN信息hadoop dfsadmin safemode enter|leavehadoop distcp hdfs:/a:8020/xxx hdfs:/b:8020/#并行copy./bin/start-balancer.sh#平衡集群文件,北京传智播客教育,HDFS API详解,static FileSystem get(Configuration conf)operator()/step1 得到Configuration对象/step2 得到FileSystem对象/step3 进行文件操作,北京传
12、智播客教育,用Java对HDFS编程,文件操作1.1上传本地文件到hadoop fs1.2 在hadoop fs中新建文件，并写入1.3 删除hadoop fs上的文件1.4读取文件1.5 文件修改时间2.目录操作2.1 在hadoop fs上创建目录2.2 删除目录2.3 读取某个目录下的所有文件2.4遍历hdfshdfs信息查找某个文件在HDFS集群的位置获取HDFS集群上所有节点名称信息,北京传智播客教育,mapreduce篇,北京传智播客教育,开发hadoop依赖的jar和自带的example,到 下载，目前最新版为。下载完后解压文件,y有hadoop-0.20.2-core.jar,
13、hadoop-0.20.2-examples.jar,hadoop-0.20.2-core.jar:hadoop的核心类库Hadoop所依赖的jar：hadoop所依赖的jar在lib目录下面。Hadoop 自带的一些案例分析：是自带的一些案例。介绍如下：1)aggregatewordcount计算输入文件中文字个数的基于聚合的MapReduce程序。2)aggregatewordhist生成输入文件中文字个数的统计图的基于聚合的MapReduce程序。3)grep计算输入文件中匹配正则表达式的文字个数的MapReduce程序。4)join合并排序的平均分割的数据集的作业。5)pentomin
14、o解决五格拼版问题的分块分层的MapReduce程序。,北京传智播客教育,Hadoop自带的examples.jar介绍,6)pi使用蒙地卡罗法计算PI的MapReduce程序。7)Randomtextwriter在一个节点上写10G随机文本的MapReduce程序。8)randomwriter在每个节点上写10G随机数据的MapReduce程序。9)sleep在每个Map和Reduce作业中休憩的程序。10)sort排序随机写入器生成的数据的MapReduce程序。11)sudoku一个九宫格游戏的解决方案。12)wordcount在输入文件中统计文字个数的统计器。,实例,写MapReduc
15、e程序的步骤：1.把问题转化为MapReduce模型2.设置运行的参数3.写map类 4.写reduce类例子：统计单词个数,My name is liu gangWhat is your name,My 1name 2is 2What 1your 1liu 1gang 1,北京传智播客教育,MapReduce模型,1.Map端一行行读文件，程序转化为中间Key/Value.My name is liu gang-My 1,name 1,is 1,liu 1,gang 1 What is your name-What 1,is 1,your 1 name 12.Reduce端相同的key肯定会
16、在一起。经过Reduce方法处理后，形成最终的key/Value.name 1,name 1-name 2;,北京传智播客教育,运行步骤,1）打成jar包。2）创建一个word.txt文件3）把word.txt文件传到HDFS上面 hadoop fs copyFromLocal 4)执行hadoop jar 5）查看执行结果 hadoop fs text/path,MapReduce执行流程,35,MapReduce基本流程,JobTracker(JT)和TaskTracker(TT)简介,再论JobTracker(JT)和TaskTracker(TT),JobTracker：协作作业的运行t
17、askTracker：运行作业划分后的任务,Mapreduce原理,1）一个文件file.txt2）存储file.txt文件3)统计file.txt文件里面”Refund”个数,JobTracker失败,1）JobTracker失败在所有的失败中是最严重的一种。2）hadoop没有处理jobtracker失败的机制。-它是一个单点故障。3）在未来的新版本中可能可以运行多个JobTracker。4）可以使用ZooKeeper来协作JobTracker。,TaskTracker失败,1）一个TaskTracker由于崩溃或运行过于缓慢而失败，它会向JobTracker发送“心跳”。2）如果有未完成
18、的作业，JobTracker会重新把这些任务分配到其他的TaskTracker上面运行。3）即使TaskTracker没有失败也可以被JobTracker列入黑名单。,Hadoop Job Scheduler,1）Hadoop默认的调度器是基于队列的FIFO调度器。所有用户的作业都被提交到一个队列中，然后由JobTracker先按照作业的优先级高低，再按照作业提交时间的先后顺序选择将被执行的作业。优点:调度算法简单明了，JobTracker工作负担轻。缺点:忽略了不同作业的需求差异。2）还用两个多用户调度器-Fair Scheduler(公平调度器)-Capacity Scheduler(计算
19、能力调度),Fair Scheduler(公平调度器),多个Pool，Job需要被提交到某个Pool中每个pool可以设置最小 task slot，称为miniShareFS会保证Pool的公平Pool内部支持Priority设置支持资源抢占,北京传智播客教育,mapreduce的类型与格式,Hadoop之前的APIHadoop的MapReduce中，map和reduce函数遵循如下常规格式：map:(K1,V1)list(K2,V2)reduce:(K2,list(V2)list(K3,V3)Mapper的接口：public interface Mapper extends JobConfi
20、gurable,Closeable void map(K1 key,V1 value,OutputCollector output,Reporter reporter)throws IOException;Reduce的接口：public interface Reducer extends JobConfigurable,Closeable void reduce(K2 key,Iterator values,OutputCollector output,Reporter reporter)throws IOException;/outputCollector 是为了输出key/value对，
21、/Reporter 是用来更新计数和状态信息。,北京传智播客教育,Hadoop之后的APIHadoop的MapReduce中，map和reduce函数遵循如下常规格式：map:(K1,V1)list(K2,V2)reduce:(K2,list(V2)list(K3,V3)Mapper的接口：protected void map(KEY key,VALUE value,Context context)throws IOException,InterruptedException Reduce的接口：protected void reduce(KEY key,Iterable values,Con
22、text context)throws IOException,InterruptedException/Context是上下文对象，这里Context等同于OutputCollector和Reporter两个函数的功能。,北京传智播客教育,mapreduce的数据类型与java类型对应关系,北京传智播客教育,Writable接口,1.对Java中的int型进行封装那么就是hadoop中的IntWritable类在写程序时可以把IntWritable可以看着是int 类型，它实现 了WritableComparable接口。WritableComparable又是Writable、接口的子接口
23、。2.Writable类对所有的Java基本类型进行封装：如：boolean-BooleanWritable;Byte-ByteWritable3.我们可以自定义Writable接口，来编写更复杂的结构的类。核心：hadoop有自己一套的I/O机制。I/O类都必须实现Writable接口。,北京传智播客教育,47,实现自定义的mapreduce类型,public class LogKey implements WritableComparable private String customerId;private String dateTime;private String domain;pu
24、blic void readFields(DataInput in)throws IOException customerId=in.readUTF();dateTime=in.readUTF();domain=in.readUTF();public void write(DataOutput out)throws IOException out.writeUTF(customerId);out.writeUTF(dateTime);out.writeUTF(domain);/当我们在实际开发的时候会遇到多个key的时候，mapreduce自带的类型是不能满足我们的需求，这样我们就要自己来定制
25、化Key和Value。,北京传智播客教育,48,最小的MapReduce驱动,public class MinimalMapReduceWithDefaults extends Configured implements Tool public int run(String args)throws IOException JobConf conf=JobBuilder.parseInputAndOutput(this,getConf(),args);if(conf=null)return-1;conf.setInputFormat(TextInputFormat.class);conf.set
26、NumMapTasks(1);conf.setMapperClass(IdentityMapper.class);conf.setMapRunnerClass(MapRunner.class);conf.setMapOutputKeyClass(LongWritable.class);conf.setMapOutputValueClass(Text.class);conf.setPartitionerClass(HashPartitioner.class);conf.setNumReduceTasks(1);conf.setReducerClass(IdentityReducer.class)
27、;conf.setOutputKeyClass(LongWritable.class);conf.setOutputValueClass(Text.class);conf.setOutputFormat(TextOutputFormat.class);JobClient.runJob(conf);return 0;public static void main(String args)throws Exception int exitCode=ToolRunner.run(new MinimalMapReduceWithDefaults(),args);System.exit(exitCode
28、);,北京传智播客教育,mapreduce驱动默认的设置,北京传智播客教育,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功能
29、.例如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叠加，而不必要等到所有的ma
30、p结束再去进行reduce的value叠加。注意：combiner使用的合适，可以在满足业务的情况下提升job的速度，如果不合适，则将导致输出的结果不正确。Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。所以从我的想法来看，Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。,北京传智播客教育,Combiners分析,假设有两个map。第一个map的输出为：（1950,0）(1950,20)(1950,10)第二个map输出为：（1950,25）(1950,1
31、5)(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要小心,刚才我们是计算最大值可以使用Combiners能提高效率。如果我们要是求平均值呢？Avg（0,20,10,25,15,30）=
32、15如果使用Combiner会得到什么样的结果呢？第一个map输出为：avg(0,20,10)=10第二个map输出为：Avg（25,15,30）=23输入到reduce出来的结果为：Avg(10,23)=17.517.5和15？所以：使用combiner一定要注意。,北京传智播客教育,Partitioner分析,Partitioner 在mapreduce的位置：,北京传智播客教育,mapreduce提供的Patitioner,Partition主要作用就是将map的结果发送到相应的reduce。这就对partition有两个要求：1）均衡负载，尽量的将工作均匀的分配给不同的reduce。2
33、）效率，分配速度一定要快。mapreduce提供的Patitioner,北京传智播客教育,Partitioner是partitioner的基类，如果需要定制partitioner也需要继承该类。2.HashPartitioner是mapreduce的默认partitioner。计算方法是which reducer=(key.hashCode()&Integer.MAX_VALUE)%numReduceTasks，得到当前的目的reducer。BinaryPatitioner继承于Partitioner，是Partitioner的偏特化子类。该类提供leftOffset和rightOffset，
34、在计算which reducer时仅对键值K的rightOffset，leftOffset这个区间取hash。Which reducer=(hash&Integer.MAX_VALUE)%numReduceTasksKeyFieldBasedPartitioner也是基于hash的个partitioner。和BinaryPatitioner不同，它提供了多个区间用于计算hash。当区间数为0时KeyFieldBasedPartitioner退化成HashPartitioner。5.TotalOrderPartitioner这个类可以实现输出的全排序。不同于以上3个partitioner，这个类
35、并不是基于hash的。在下一节里详细的介绍totalorderpartitioner。,北京传智播客教育,自定义的Partitioner,1）为何使用Partitioner，主要是想reduce的结果能够根据key再次分类输出到不同的文件夹中。2）结果能够直观，同时做到对数据结果的简单的统计分析。需求：1、输入的数据文件内容如下(1条数据内容少，1条数据内容超长，3条数据内容正常)：Kaka 1 28 hua 0 26 chao 1 tao 1 22 mao 0 29 222、目的是为了分别输出结果，正确的结果输出到一个文本，太短的数据输出到一个文本，太长的输出到一个文本，共三个文本输出。Patitioner接口：public int getPartition(Text key,Text value,int numPartitions)；numPartitions为Reduce的个数。注：在本地作业运行器上运行时，只支持0个或一个Reduce。,
