hadoop实战培训传智播客

1、高级软件人才实作培训专家!Hahoop 培训1海量数据处理平台架构一般网站把用户的访问行为记录以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 服务器上来了。2 有了LogBackup 服务器，离线统计就可以全部在这些服务器上进行了。在这套架构上，用wc、grep、sort、uniq、awk、sed 等系统命令，完成了很多的统计需求，比如统计访问频率较高的client_ip，某个新上线的的页面的referer 主要是哪些网站。 当业务的迅猛发展，网站流量爆

3、发增长，产品经理如果想从中获取更多的用户特征和用户信息， 就需要我们这些数据分析人员从不同的日志中找到令他们满意的答案。如果 （1） 日志总行数：10 亿/天 （2） 每天日志大小：450Byte/行* 10 亿= 420G， （3） 日志种类：5 种3Hadoop能解决哪些问题海量数据需要及时分析和处理。海量数据需要深入分析和挖掘。数据需要长期保存问题：磁盘IO成为一种瓶颈,而非CPU资源。网络带宽是一种稀缺资源硬件故障成为影响稳定的一大因素4Hadoop在国内的情景奇虎360：Hadoop存储软件管家中软件，使用CDN技术将用户请求引到最近的Hadoop集群并进行下载京东、百度：存储、分析

4、日志、数据挖掘和机器学习（主要是推荐系统）广告类公司：存储日志，通过协调过滤算法为客户推荐广告Yahoo：垃圾邮件过滤华为：云计算平台Facebook：日志存储，实时分析某公安部项目：网民QQ聊天记录与关联人调查系统，使用Hbase实现某学校：学生上网与社会行为分析，使用hadoop淘宝、阿里：国内使用Hadoop最深入的公司，整个Taobao和阿里都是数据驱动的5Hadoop开发人员市场需求和待遇6Hadoop在国内的人才储备7Hadoop介绍1）作者：Doug Cutting2)用Java编写的开源系统，能够安排在大规模的计算平台上，从而长进计算效率。3）Nutch搜索引擎里面的一个模块。

5、4）受Google三篇论文的启发 MapReduce GFS Bigtable5）google hadoop mapreduce mapreduce GFS HDFS Bigtable Hbase8Hadoop生态系统介绍Hbase 1）Nosql数据库，Key-Value存储 2）最大化利用内存HDFS 1） hadoop distribute 分布式文件系统 2）最大化利用磁盘MapReduce 1）编程模型，主要用来做数据的分析 2）最大化利用CPU9 HDFS篇10HDFS设计原则文件以块(block)方式存储每个块带下远比多数文件系统来的大(预设64M)通过副本机制提高可靠度和读取吞

6、吐量每个区块至少分到三台DataNode上单一 master (NameNode)来协调存储元数据(metadata)客户端对文件没有缓存机制 (No data caching)11HDFS系统结构12NameNode(NN)NameNode主要功能提供名称查询服务，它是一个jetty服务器NameNode保存metadate信息包括文件owership和permissions文件包含哪些块Block保存在哪个DataNode（由DataNode启动时上报）NameNode的metadate信息在启动后会加载到内存metadata存储到磁盘文件名为”fsimage”Block的位置信息不会保存

7、到fsimage13NameNode块存储结构metadate物理存储结构14DataNode（DN）保存Block启动DN线程的时候会向NN汇报block信息通过向NN发送心跳保持与其联系（3秒一次），如果NN 10分钟没有收到DN的心跳，则认为其已经lost，并copy其上的block到其它DN 15Block的副本放置策略第一个副本：放置在上传文件的DN；如果是集群外提交，则随机挑选一台磁盘不太满，CPU不太忙的节点第二个副本：放置在于第一个副本不同的机架的节点上第三个副本：与第二个副本相同集群的节点更多副本：随机节点16再说Block设置一个Block 64MB，如果上传文件小于该值，

8、仍然会占用一个Block的命名空间（NameNode metadata），但是物理存储上不会占用64MB的空间Block大小和副本数由Client端上传文件到HDFS时设置，其中副本数可以变更，Block是不可以再上传后变更的17数据损坏(corruption)处理当DN读取block的时候，它会计算checksum如果计算后的checksum，与block创建时值不一样，说明该block已经损坏。client读取其它DN上的block；NN标记该块已经损坏，然后复制block达到预期设置的文件备份数DN在其文件创建后三周验证其checksum18HDFS文件权限与Linux文件权限类似r:

9、read; w:write; x:execute，权限x对于文件忽略，对于文件夹表示是否允许访问其内容如果Linux系统用户zhangsan使用hadoop命令创建一个文件，那么这个文件在HDFS中owner就是zhangsanHDFS的权限目的：阻止好人错错事，而不是阻止坏人做坏事。HDFS相信，你告诉我你是谁，我就认为你是谁19HDFS文件读取20HDFS文件写入21HDFS文件存储 两个文件，一个文件156M，一个文件128在HDFS里面怎么存储？ -Block为64MB -rapliction默认拷贝3份22HDFS文件存储结构23HDFS开发常用命令创建一个文件夹？上传一个文件？删除

10、一个文件和文件夹？查看一个文件夹里面有哪些文件？查看某个文件的内容？24Hadoop管理员常用命令hadoop job list #列出正在运行的Jobhadoop job kill #kill jobhadoop fsck / #检查HDFS块状态，是否损坏hadoop fsck / -delete #检查HDFS块状态，删除损坏块hadoop dfsadmin report #检查HDFS状态，包括DN信息hadoop dfsadmin safemode enter | leavehadoop distcp hdfs:/a:8020/xxx hdfs:/b:8020/ #并行copy./b

11、in/start-balancer.sh #平衡集群文件25HDFS API详解static get(Configuration conf)operator() /step1 得到Configuration对象 /step2 得到对象 /step3 进行文件操作 26用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信息查找某个文件在

12、HDFS集群的位置获取HDFS集群上所有节点名称信息27 mapreduce篇28开发hadoop依赖的jar和自带的example到 下载hadoop-0.20.2.tar.gz，目前最新版为1.0.3。下载完后解压文件,y有hadoop-0.20.2-core.jar, hadoop-0.20.2-examples.jar,hadoop-0.20.2-core.jar :hadoop的核心类库Hadoop所依赖的jar： hadoop所依赖的jar在lib目录下面。Hadoop 自带的一些案例分析： hadoop-0.20.2-examples.jar是hadoop-0.20.2自带的一些

13、案例。介绍如下： 1) aggregatewordcount计算输入文件中文字个数的基于聚合的MapReduce程序。 2) aggregatewordhist生成输入文件中文字个数的统计图的基于聚合的MapReduce程序。 3) grep计算输入文件中匹配正则表达式的文字个数的MapReduce程序。4) join合并排序的平均分割的数据集的作业。5)pentomino解决五格拼版问题的分块分层的MapReduce程序。29Hadoop自带的examples.jar介绍6)pi使用蒙地卡罗法计算PI的MapReduce程序。7)Randomtextwriter在一个节点上写10G随机文本的

14、MapReduce程序。8)randomwriter在每个节点上写10G随机数据的MapReduce程序。9) sleep在每个Map和Reduce作业中休憩的程序。10)sort排序随机写入器生成的数据的MapReduce程序。11)sudoku一个九宫格游戏的解决方案。12)wordcount在输入文件中统计文字个数的统计器。30实例写MapReduce程序的步骤：1.把问题转化为MapReduce模型2.设置运行的参数3.写map类 4.写reduce类例子：统计单词个数 My name is liu gangWhat is your nameMy 1name 2is 2What 1yo

15、ur 1liu 1gang 131MapReduce模型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肯定会在一起。经过Reduce方法处理后，形成最终的key/Value. name 1,name 1-name 2;32运行步骤1）打成jar包。2）创建一个word.txt文件3）把word.txt文件传到HDFS上面 hadoop fs copyF

16、romLocal 4)执行hadoop jar 5）查看执行结果 hadoop fs text /path 33MapReduce执行流程34MapReduce基本流程35JobTracker(JT)和TaskTracker(TT)简介36再论JobTracker(JT)和TaskTracker(TT)JobTracker：协作作业的运行taskTracker：运行作业划分后的任务37Mapreduce原理1）一个文件2）存储文件3) 统计文件里面”Refund”个数38JobTracker失败1）JobTracker失败在所有的失败中是最严重的一种。2）hadoop没有处理jobtracke

17、r失败的机制。-它是一个单点故障。3）在未来的新版本中可能可以运行多个JobTracker。4）可以使用ZooKeeper来协作JobTracker。39TaskTracker失败1）一个TaskTracker由于崩溃或运行过于缓慢而失败，它会向JobTracker发送“心跳”。2）如果有未完成的作业，JobTracker会重新把这些任务分配到其他的TaskTracker上面运行。3）即使TaskTracker没有失败也可以被JobTracker列入黑名单。40Hadoop Job Scheduler1）Hadoop默认的调度器是基于队列的FIFO调度器。 所有用户的作业都被提交到一个队列中，

18、然后由JobTracker先按照作业的优先级高低，再按照作业提交时间的先后顺序选择将被执行的作业。优点: 调度算法简单明了，JobTracker工作负担轻。缺点: 忽略了不同作业的需求差异。2）还用两个多用户调度器 -Fair Scheduler(公平调度器) -Capacity Scheduler(计算能力调度) 41Fair Scheduler(公平调度器)多个Pool，Job需要被提交到某个Pool中每个pool可以设置最小 task slot，称为miniShareFS会保证Pool的公平Pool内部支持Priority设置支持资源抢占42 mapreduce的类型与格式Hadoop

19、0.20.x之前的APIHadoop的MapReduce中，map和reduce函数遵循如下常规格式： map: (K1, V1) list(K2, V2) reduce: (K2, list(V2) list(K3, V3)Mapper的接口： public interface Mapper extends JobConfigurable, Closeable void map(K1 key, V1 value, OutputCollector output, Reporter reporter) throws IOException; Reduce的接口： public interface

20、 Reducer extends JobConfigurable, Closeable void reduce(K2 key, Iterator values, OutputCollector output, Reporter reporter) throws IOException; /outputCollector 是为了输出key/value对，/Reporter 是用来更新计数和状态信息。43Hadoop 0.20.x之后的APIHadoop的MapReduce中，map和reduce函数遵循如下常规格式： map: (K1, V1) list(K2, V2) reduce: (K2,

21、 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, Context context ) throws IOException, InterruptedException /Context是上下文对象，这里Context等同于OutputCollector和Reporter两

22、个函数的功能。44mapreduce的数据类型与java类型对应关系Java的基本类型Writable实现booleanBooleanWritablebyteByteWritableintIntWritablefloatFloatWritablelongLongWritabledoubleDoubleWritableStringText45Writable接口1.对Java中的int型进行封装那么就是hadoop中的IntWritable类在写程序时可以把IntWritable可以看着是int 类型，它实现 了WritableComparable接口。 WritableComparable又是

23、Writable、parable接口的子接口。2.Writable类对所有的Java基本类型进行封装：如：boolean - BooleanWritable;Byte - ByteWritable3. 我们可以自定义Writable接口，来编写更复杂的结构的类。核心：hadoop有自己一套的I/O机制。I/O类都必须实现Writable接口。46实现自定义的mapreduce类型public class LogKey implements WritableComparable private String customerId; private String dateTime; private

24、 String domain;public 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的时候，mapredu

25、ce自带的类型是不能满足我们的需求，这样我们就要自己来定制化Key和Value。47最小的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(Text

26、InputFormat.class);conf.setNumMapTasks(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.setReducerC

27、lass(IdentityReducer.class);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 MinimalMapReduceWithDefault

28、s(), args);System.exit(exitCode);48mapreduce驱动默认的设置InputFormat(输入)TextInputFOrmatMapperClass(map类)IdentityMapperMapRunnerClass(map启动类)MapRunnerMapOutputKeyClassLongWritableMapOutputValueClassTextPartitionerClassHashPartitionerReduceClassIdentityReduceOutputKeyClassLongWritableOutputValueClassTextOut

29、putFormatClassTextOutputFormat49Combiners和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还具有类似本地的

30、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一结束就可以进行redu

31、ce的value叠加，而不必要等到所有的map结束再去进行reduce的value叠加。注意：combiner使用的合适，可以在满足业务的情况下提升job的速度，如果不合适，则将导致输出的结果不正确。Combiner的输出是Reducer的输入，Combiner绝不能改变最终的计算结果。所以从我的想法来看，Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致，且不影响最终结果的场景。比如累加，最大值等。50 Combiners分析假设有两个map。第一个map的输出为：（1950,0） (1950,20) (1950,10)第二个map输出为：

32、（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 51使用 Combiners要小心刚才我们是计算最大值可以使用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一定要注意。52 Partitioner分析Partitioner 在mapreduce的位置：53mapreduce提供的PatitionerPartition主要作用就是将