hadoop实战培训传智播客.ppt

—高级软件人才实作培训专家高级软件人才实作培训专家! !Hahoop 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解决方案：①   为了方便系统命令查看日志，不压缩，总共需要3.0T 的空间，刚好有一些2U 的服务器，每台共1T 的磁盘空间②为了避免系统盘坏掉影响服务器使用，对系统盘做了raid1③为了避免其他存放数据的盘坏掉导致数据无法恢复，对剩下的盘做了raid5。

④所有的数据都汇聚到这几台LogBackup 服务器上来了2—高级软件人才实作培训专家高级软件人才实作培训专家! !             有了LogBackup 服务器，离线统计就可以全部在这些服务器上进行了在这套架构上，用wc、grep、sort、uniq、awk、sed 等系统命令，完成了很多的统计需求，比如统计访问频率较高的client_ip，某个新上线的的页面的referer 主要是哪些网站Ø   当业务的迅猛发展，网站流量爆发增长，产品经理如果想从中获取更多的用户特征和用户信息，     就需要我们这些数据分析人员从不同的日志中找到令他们满意的答案如果     （1） 日志总行数：10 亿/天     （2） 每天日志大小：450Byte/行* 10 亿= 420G，     （3） 日志种类：5 种3—高级软件人才实作培训专家高级软件人才实作培训专家! !Hadoop能解决哪些问题l海量数据需要及时分析和处理l海量数据需要深入分析和挖掘l数据需要长期保存问题：l磁盘IO成为一种瓶颈,而非CPU资源l网络带宽是一种稀缺资源l硬件故障成为影响稳定的一大因素4—高级软件人才实作培训专家高级软件人才实作培训专家! !Hadoop在国内的情景l奇虎360：Hadoop存储软件管家中软件，使用CDN技术将用户请求引到最近的Hadoop集群并进行下载l京东、百度：存储、分析日志、数据挖掘和机器学习（主要是推荐系统）l广告类公司：存储日志，通过协调过滤算法为客户推荐广告lYahoo：垃圾邮件过滤l华为：云计算平台lFacebook：日志存储，实时分析l某公安部项目：网民聊天记录与关联人调查系统，使用Hbase实现l某学校：学生上网与社会行为分析，使用hadoopl淘宝、阿里：国内使用Hadoop最深入的公司，整个Taobao和阿里都是数据驱动的5—高级软件人才实作培训专家高级软件人才实作培训专家! !Hadoop开发人员市场需求和待遇6—高级软件人才实作培训专家高级软件人才实作培训专家! !Hadoop在国内的人才储备7—高级软件人才实作培训专家高级软件人才实作培训专家! !Hadoop介绍1）作者：Doug Cutting2)用Java编写的开源系统，能够安排在大规模的计算平台上，从而长进计算效率。

3）Nutch搜索引擎里面的一个模块4）受Google三篇论文的启发     ---MapReduce  GFS  Bigtable5）google          hadoop     mapreduce    mapreduce      GFS             HDFS      Bigtable        Hbase8—高级软件人才实作培训专家高级软件人才实作培训专家! !Hadoop生态系统介绍uHbase         1）Nosql数据库，Key-Value存储         2）最大化利用内存uHDFS        1） hadoop distribute 分布式文件系统        2）最大化利用磁盘uMapReduce       1）编程模型，主要用来做数据的分析       2）最大化利用CPU9—高级软件人才实作培训专家高级软件人才实作培训专家! !                       HDFS篇10—高级软件人才实作培训专家高级软件人才实作培训专家! !HDFS设计原则l文件以块(block)方式存储l每个块带下远比多数文件系统来的大(预设64M)l通过副本机制提高可靠度和读取吞吐量l每个区块至少分到三台DataNode上l单一 master (NameNode)来协调存储元数据(metadata)l客户端对文件没有缓存机制 (No data caching)11—高级软件人才实作培训专家高级软件人才实作培训专家! !HDFS系统结构12—高级软件人才实作培训专家高级软件人才实作培训专家! !NameNode(NN)lNameNode主要功能提供名称查询服务，它是一个jetty服务器lNameNode保存metadate信息包括•文件owership和permissions•文件包含哪些块•Block保存在哪个DataNode（由DataNode启动时上报）lNameNode的metadate信息在启动后会加载到内存lmetadata存储到磁盘文件名为”fsimage”•Block的位置信息不会保存到fsimage13—高级软件人才实作培训专家高级软件人才实作培训专家! !NameNode块存储结构metadate物理存储结构14—高级软件人才实作培训专家高级软件人才实作培训专家! !DataNode（DN）•保存Block•启动DN线程的时候会向NN汇报block信息•通过向NN发送心跳保持与其联系（3秒一次），如果NN 10分钟没有收到DN的心跳，则认为其已经lost，并copy其上的block到其它DN 15—高级软件人才实作培训专家高级软件人才实作培训专家! !Block的副本放置策略•第一个副本：放置在上传文件的DN；如果是集群外提交，则随机挑选一台磁盘不太满，CPU不太忙的节点•第二个副本：放置在于第一个副本不同的机架的节点上•第三个副本：与第二个副本相同集群的节点•更多副本：随机节点16—高级软件人才实作培训专家高级软件人才实作培训专家! !再说Blockl设置一个Block 64MB，如果上传文件小于该值，仍然会占用一个Block的命名空间（NameNode metadata），但是物理存储上不会占用64MB的空间lBlock大小和副本数由Client端上传文件到HDFS时设置，其中副本数可以变更，Block是不可以再上传后变更的17—高级软件人才实作培训专家高级软件人才实作培训专家! !数据损坏(corruption)处理l当DN读取block的时候，它会计算checksuml如果计算后的checksum，与block创建时值不一样，说明该block已经损坏。

lclient读取其它DN上的block；NN标记该块已经损坏，然后复制block达到预期设置的文件备份数lDN在其文件创建后三周验证其checksum18—高级软件人才实作培训专家高级软件人才实作培训专家! !HDFS文件权限•与Linux文件权限类似•r: read; w:write; x:execute，权限x对于文件忽略，对于文件夹表示是否允许访问其内容•如果Linux系统用户zhangsan使用hadoop命令创建一个文件，那么这个文件在HDFS中owner就是zhangsan•HDFS的权限目的：阻止好人错错事，而不是阻止坏人做坏事HDFS相信，你告诉我你是谁，我就认为你是谁19—高级软件人才实作培训专家高级软件人才实作培训专家! !HDFS文件读取20—高级软件人才实作培训专家高级软件人才实作培训专家! !HDFS文件写入21—高级软件人才实作培训专家高级软件人才实作培训专家! !HDFS文件存储     两个文件，一个文件156M，一个文件128在HDFS里面怎么存储？        --Block为64MB        --rapliction默认拷贝3份22—高级软件人才实作培训专家高级软件人才实作培训专家! !HDFS文件存储结构23—高级软件人才实作培训专家高级软件人才实作培训专家! !HDFS开发常用命令l创建一个文件夹？l上传一个文件？l删除一个文件和文件夹？l查看一个文件夹里面有哪些文件？l查看某个文件的内容？24—高级软件人才实作培训专家高级软件人才实作培训专家! !Hadoop管理员常用命令lhadoop job –list  #列出正在运行的Joblhadoop job –kill  #kill joblhadoop fsck / #检查HDFS块状态，是否损坏lhadoop fsck / -delete #检查HDFS块状态，删除损坏块lhadoop dfsadmin –report #检查HDFS状态，包括DN信息lhadoop dfsadmin –safemode enter | leavelhadoop distcp hdfs://a:8020/xxx hdfs://b:8020/// #并行copyl./bin/start-balancer.sh #平衡集群文件25—高级软件人才实作培训专家高级软件人才实作培训专家! !HDFS API详解详解static  get(Configuration conf)operator()     {         //step1        得到Configuration对象        //step2        得到对象        //step3        进行文件操作     }26—高级软件人才实作培训专家高级软件人才实作培训专家! !用Java对HDFS编程1.文件操作l1.1  上传本地文件到hadoop fsl1.2 在hadoop fs中新建文件，并写入l1.3 删除hadoop fs上的文件l1.4  读取文件l1.5 文件修改时间2.    目录操作l2.1 在hadoop fs上创建目录l2.2 删除目录l2.3 读取某个目录下的所有文件 l2.4遍历hdfs3.hdfs信息l查找某个文件在HDFS集群的位置l获取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自带的一些案例介绍如下：   1) aggregatewordcount计算输入文件中文字个数的基于聚合的MapReduce程序  2) aggregatewordhist生成输入文件中文字个数的统计图的基于聚合的MapReduce程序  3) grep计算输入文件中匹配正则表达式的文字个数的MapReduce程序4) join合并排序的平均分割的数据集的作业5)pentomino解决五格拼版问题的分块分层的MapReduce程序29—高级软件人才实作培训专家高级软件人才实作培训专家! !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在输入文件中统计文字个数的统计器30—高级软件人才实作培训专家高级软件人才实作培训专家! !实例l写MapReduce程序的步骤：1.把问题转化为MapReduce模型2.设置运行的参数3.写map类    4.写reduce类l例子：统计单词个数      My name is liu gangWhat is your nameMy 1name 2is 2What 1your 1liu 1gang 131—高级软件人才实作培训专家高级软件人才实作培训专家! !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肯定会在一起。

经过Reduce方法处理后，形成最终的key/Value. name 1,name 1->name 2;32—高级软件人才实作培训专家高级软件人才实作培训专家! !运行步骤1）打成jar包2）创建一个word.txt文件3）把word.txt文件传到HDFS上面    hadoop fs –copyFromLocal   4)执行hadoop jar   <完整的类名>5）查看执行结果   hadoop fs –text  /path 33—高级软件人才实作培训专家高级软件人才实作培训专家! !MapReduce执行流程34—高级软件人才实作培训专家高级软件人才实作培训专家! !MapReduce基本流程35—高级软件人才实作培训专家高级软件人才实作培训专家! !JobTracker(JT)和TaskTracker(TT)简介36—高级软件人才实作培训专家高级软件人才实作培训专家! !再论JobTracker(JT)和TaskTracker(TT)•JobTracker：协作作业的运行•taskTracker：运行作业划分后的任务37—高级软件人才实作培训专家高级软件人才实作培训专家! !Mapreduce原理1）一个文件2）存储文件3) 统计文件里面”Refund”个数38—高级软件人才实作培训专家高级软件人才实作培训专家! !JobTracker失败1）JobTracker失败在所有的失败中是最严重的一种。

2）hadoop没有处理jobtracker失败的机制它是一个单点故障3）在未来的新版本中可能可以运行多个JobTracker4）可以使用ZooKeeper来协作JobTracker39—高级软件人才实作培训专家高级软件人才实作培训专家! !TaskTracker失败1）一个TaskTracker由于崩溃或运行过于缓慢而失败，它会向JobTracker发送“心跳”2）如果有未完成的作业，JobTracker会重新把这些任务分配到其他的TaskTracker上面运行3）即使TaskTracker没有失败也可以被JobTracker列入黑名单40—高级软件人才实作培训专家高级软件人才实作培训专家! !Hadoop Job Scheduler1）Hadoop默认的调度器是基于队列的FIFO调度器           所有用户的作业都被提交到一个队列中，然后由JobTracker先按照作业的优先级高低，再按照作业提交时间的先后顺序选择将被执行的作业优点: 调度算法简单明了，JobTracker工作负担轻缺点: 忽略了不同作业的需求差异2）还用两个多用户调度器         --Fair Scheduler(公平调度器)         --Capacity Scheduler(计算能力调度)  41—高级软件人才实作培训专家高级软件人才实作培训专家! !Fair Scheduler(公平调度器)•多个Pool，Job需要被提交到某个Pool中•每个pool可以设置最小 task slot，称为miniShare•FS会保证Pool的公平•Pool内部支持Priority设置•支持资源抢占42—高级软件人才实作培训专家高级软件人才实作培训专家! ! mapreduce的类型与格式Hadoop 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 Reducer extends JobConfigurable, Closeable {    void reduce(K2 key, Iterator values,   OutputCollector output, Reporter reporter) throws IOException;  }//outputCollector 是为了输出key/value对，//Reporter 是用来更新计数和状态信息。

43—高级软件人才实作培训专家高级软件人才实作培训专家! !Hadoop 0.20.x之后的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, protected void map(KEY key, VALUE value, Context context) throws IOException, InterruptedException {throws IOException, InterruptedException {}Reduce的接口：protected void reduce(KEY key, Iterable values, Context contextprotected void reduce(KEY key, Iterable values, Context context ) throws IOException, InterruptedException {throws IOException, InterruptedException {}//Context是上下文对象，这里Context等同于OutputCollector和Reporter两个函数的功能。

44—高级软件人才实作培训专家高级软件人才实作培训专家! !mapreduce的数据类型与java类型对应关系Java的基本类型Writable实现booleanBooleanWritablebyteByteWritableintIntWritablefloatFloatWritablelongLongWritabledoubleDoubleWritableStringText45—高级软件人才实作培训专家高级软件人才实作培训专家! !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接口46—高级软件人才实作培训专家高级软件人才实作培训专家! !实现自定义的mapreduce类型public class LogKey implements WritableComparable{public class LogKey implements WritableComparable{ private String customerId; private String customerId; private String dateTime; private String dateTime; private String domain; private String domain;public void readFields(DataInput in) throws IOException {public void readFields(DataInput in) throws IOException { customerId = in.readUTF(); dateTime = in.readUTF(); domain = in.readUTF();}public void write(DataOutput out) throws IOException {public void write(DataOutput out) throws IOException { out.writeUTF(customerId); out.writeUTF(dateTime); out.writeUTF(domain);}} }////当我们在实际开发的时候会遇到多个当我们在实际开发的时候会遇到多个keykey的时候，的时候，mapreducemapreduce自带的类型是不能满足我们的自带的类型是不能满足我们的需求，这样我们就要自己来定制化需求，这样我们就要自己来定制化KeyKey和和ValueValue。

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(TextInputFormat.class);conf.setInputFormat(TextInputFormat.class);conf.setNumMapTasks(1);conf.setNumMapTasks(1);conf.setMapperClass(IdentityMapper.class);conf.setMapperClass(IdentityMapper.class);conf.setMapRunnerClass(MapRunner.class);conf.setMapRunnerClass(MapRunner.class);conf.setMapOutputKeyClass(LongWritable.class);conf.setMapOutputKeyClass(LongWritable.class);conf.setMapOutputValueClass(Text.class);conf.setMapOutputValueClass(Text.class);conf.setPartitionerClass(HashPartitioner.class);conf.setPartitionerClass(HashPartitioner.class);conf.setNumReduceTasks(1);conf.setNumReduceTasks(1);conf.setReducerClass(IdentityReducer.class);conf.setReducerClass(IdentityReducer.class);conf.setOutputKeyClass(LongWritable.class);conf.setOutputKeyClass(LongWritable.class);conf.setOutputValueClass(Text.class);conf.setOutputValueClass(Text.class);conf.setOutputFormat(TextOutputFormat.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);}}48—高级软件人才实作培训专家高级软件人才实作培训专家! !mapreduce驱动默认的设置InputFormat(输入)TextInputFOrmatMapperClass(map类)IdentityMapperMapRunnerClass(map启动类)MapRunnerMapOutputKeyClassLongWritableMapOutputValueClassTextPartitionerClassHashPartitionerReduceClassIdentityReduceOutputKeyClassLongWritableOutputValueClassTextOutputFormatClassTextOutputFormat49—高级软件人才实作培训专家高级软件人才实作培训专家! !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类型完全一致，且不影响最终结果的场景比如累加，最大值等50—高级软件人才实作培训专家高级软件人才实作培训专家! ! 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 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的位置：53—高级软件人才实作培训专家高级软件人才实作培训专家! !mapreduce提供的PatitionerPartition主要作用就是将map的结果发送到相应的reduce这就对partition有两个要求：1）均衡负载，尽量的将工作均匀的分配给不同的reduce2）效率，分配速度一定要快mapreduce提供的Patitioner54—高级软件人才实作培训专家高级软件人才实作培训专家! !1.Partitioner是partitioner的基类，如果需要定制partitioner也需要继承该类。

2.      HashPartitioner是mapreduce的默认partitioner计算方法是1.which reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks，得到当前的目的reducer•BinaryPatitioner继承于Partitioner< BinaryComparable ,V>，是Partitioner的偏特化子类该类提供leftOffset和rightOffset，在计算which reducer时仅对键值K的[rightOffset，leftOffset]这个区间取hash1.Which reducer=(hash & Integer.MAX_VALUE) % numReduceTasks•KeyFieldBasedPartitioner也是基于hash的个partitioner和BinaryPatitioner不同，它提供了多个区间用于计算hash当区间数为0时KeyFieldBasedPartitioner退化成HashPartitioner5.    TotalOrderPartitioner这个类可以实现输出的全排序。

不同于以上3个partitioner，这个类并不是基于hash的在下一节里详细的介绍totalorderpartitioner55—高级软件人才实作培训专家高级软件人才实作培训专家! !自定义的Partitioner1）为何使用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。