hadoop实战培训

nullHahoop 培训—高级软件人才实作培训专家!Hahoop 培训讲师：刘刚海量数据处理平台架构北京传智播客教育 www.itcast.cn*海量数据处理平台架构一般网站把用户的访问行为记录以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 解决 方案 气瓶 现场处置方案 .pdf气瓶 现场处置方案 .doc见习基地管理方案.doc关于群访事件的化解方案建筑工地扬尘治理专项方案下载 ： 为了方便系统命令查看日志，不压缩，总共需要3.0T 的空间，刚好有一些2U 的服务器，每台共1T 的磁盘空间。 为了避免系统盘坏掉影响服务器使用，对系统盘做了raid1。 为了避免其他存放数据的盘坏掉导致数据无法恢复，对剩下的盘做了raid5。 所有的数据都汇聚到这几台LogBackup 服务器上来了。null北京传智播客教育 www.itcast.cn* 有了LogBackup 服务器，离线统计就可以全部在这些服务器上进行了。在这套架构上，用wc、grep、sort、uniq、awk、sed 等系统命令，完成了很多的统计需求，比如统计访问频率较高的client_ip，某个新上线的的页面的referer 主要是哪些网站。 当业务的迅猛发展，网站流量爆发增长，产品经理如果想从中获取更多的用户特征和用户信息， 就需要我们这些数据分析人员从不同的日志中找到令他们满意的 答案 八年级地理上册填图题岩土工程勘察试题省略号的作用及举例应急救援安全知识车间5s试题及答案 。如果 （1） 日志总行数：10 亿/天 （2） 每天日志大小：450Byte/行* 10 亿= 420G， （3） 日志种类：5 种Hadoop能解决哪些问题北京传智播客教育 www.itcast.cnHadoop能解决哪些问题海量数据需要及时分析和处理。 海量数据需要深入分析和挖掘。 数据需要长期保存 问题： 磁盘IO成为一种瓶颈,而非CPU资源。 网络带宽是一种稀缺资源 硬件故障成为影响稳定的一大因素 Hadoop在国内的情景北京传智播客教育 www.itcast.cnHadoop在国内的情景奇虎360：Hadoop存储软件管家中软件，使用CDN技术将用户请求引到最近的Hadoop集群并进行下载 京东、百度：存储、分析日志、数据挖掘和机器学习（主要是推荐系统） 广告类公司：存储日志，通过协调过滤算法为客户推荐广告 Yahoo：垃圾邮件过滤 华为：云计算平台 Facebook：日志存储，实时分析 某公安部项目：网民QQ聊天记录与关联人调查系统，使用Hbase实现 某学校：学生上网与社会行为分析，使用hadoop 淘宝、阿里：国内使用Hadoop最深入的公司，整个Taobao和阿里都是数据驱动的 Hadoop开发人员市场需求和待遇北京传智播客教育 www.itcast.cnHadoop开发人员市场需求和待遇Hadoop在国内的人才储备北京传智播客教育 www.itcast.cnHadoop在国内的人才储备Hadoop介绍北京传智播客教育 www.itcast.cnHadoop介绍1）作者：Doug Cutting 2)用Java编写的开源系统，能够安排在大规模的计算平台上，从而长进计算效率。 3）Nutch搜索引擎里面的一个模块。 4）受Google三篇 论文 政研论文下载论文大学下载论文大学下载关于长拳的论文浙大论文封面下载 的启发 ---MapReduce GFS Bigtable 5）google hadoop mapreduce mapreduce GFS HDFS Bigtable Hbase Hadoop生态系统介绍北京传智播客教育 www.itcast.cnHadoop生态系统介绍Hbase 1）Nosql数据库，Key-Value存储 2）最大化利用内存 HDFS 1） hadoop distribute file system分布式文件系统 2）最大化利用磁盘 MapReduce 1）编程模型，主要用来做数据的分析 2）最大化利用CPUnull北京传智播客教育 www.itcast.cn HDFS篇HDFS 设计 领导形象设计圆作业设计ao工艺污水处理厂设计附属工程施工组织设计清扫机器人结构设计 原则北京传智播客教育 www.itcast.cnHDFS设计原则文件以块(block)方式存储 每个块带下远比多数文件系统来的大(预设64M) 通过副本机制提高可靠度和读取吞吐量 每个区块至少分到三台DataNode上 单一 master (NameNode)来协调存储元数据(metadata) 客户端对文件没有缓存机制 (No data caching) HDFS系统结构北京传智播客教育 www.itcast.cnHDFS系统结构NameNode(NN)北京传智播客教育 www.itcast.cnNameNode(NN)NameNode主要功能提供名称查询服务，它是一个jetty服务器 NameNode保存metadate信息包括 文件owership和permissions 文件包含哪些块 Block保存在哪个DataNode（由DataNode启动时上报） NameNode的metadate信息在启动后会加载到内存 metadata存储到磁盘文件名为”fsimage” Block的位置信息不会保存到fsimage NameNodeNameNode块存储结构metadate物理存储结构DataNode（DN）DataNode（DN）保存Block 启动DN线程的时候会向NN汇报block信息 通过向NN发送心跳保持与其联系（3秒一次），如果NN 10分钟没有收到DN的心跳，则认为其已经lost，并copy其上的block到其它DN Block的副本放置策略Block的副本放置策略第一个副本：放置在上传文件的DN；如果是集群外提交，则随机挑选一台磁盘不太满，CPU不太忙的节点 第二个副本：放置在于第一个副本不同的机架的节点上 第三个副本：与第二个副本相同集群的节点 更多副本：随机节点再说Block北京传智播客教育 www.itcast.cn再说Block设置一个Block 64MB，如果上传文件小于该值，仍然会占用一个Block的命名空间（NameNode metadata），但是物理存储上不会占用64MB的空间 Block大小和副本数由Client端上传文件到HDFS时设置，其中副本数可以变更，Block是不可以再上传后变更的 数据损坏(corruption)处理北京传智播客教育 www.itcast.cn数据损坏(corruption)处理当DN读取block的时候，它会计算checksum 如果计算后的checksum，与block创建时值不一样，说明该block已经损坏。 client读取其它DN上的block；NN标记该块已经损坏，然后复制block达到预期设置的文件备份数 DN在其文件创建后三周验证其checksum HDFS文件权限*HDFS文件权限与Linux文件权限类似 r: read; w:write; x:execute，权限x对于文件忽略，对于文件夹表示是否允许访问其内容 如果Linux系统用户zhangsan使用hadoop命令创建一个文件，那么这个文件在HDFS中owner就是zhangsan HDFS的权限目的：阻止好人错错事，而不是阻止坏人做坏事。HDFS相信，你告诉我你是谁，我就认为你是谁 HDFS文件读取HDFS文件读取HDFS文件写入HDFS文件写入HDFS文件存储北京传智播客教育 www.itcast.cnHDFS文件存储 两个文件，一个文件156M，一个文件128在HDFS里面怎么存储？ --Block为64MB --rapliction默认拷贝3份HDFS文件存储结构北京传智播客教育 www.itcast.cnHDFS文件存储结构HDFS开发常用命令北京传智播客教育 www.itcast.cnHDFS开发常用命令创建一个文件夹？ Hadoop fs –mkdir /user/hadoop/20120722 上传一个文件？ Hadoop fs –put/copyFromLocal word .txt /user/hadoop/20120722 删除一个文件和文件夹？ Hadoop fs –rm /user/hadoop/20120722/word.txt 查看一个文件夹里面有哪些文件？ Hadoop fs –ls /user/hadoop/20120722/ 查看某个文件的内容？ Hadoop fs -text/cat/tail /user/hadoop/20120722/word.txt Hadoop管理员常用命令北京传智播客教育 www.itcast.cnHadoop管理员常用命令hadoop job –list #列出正在运行的Job hadoop job –kill #kill job hadoop fsck / #检查HDFS块状态，是否损坏 hadoop fsck / -delete #检查HDFS块状态，删除损坏块 hadoop dfsadmin –report #检查HDFS状态，包括DN信息 hadoop dfsadmin –safemode enter | leave hadoop distcp hdfs://a:8020/xxx hdfs://b:8020/// #并行copy ./bin/start-balancer.sh #平衡集群文件 HDFS API详解北京传智播客教育 www.itcast.cnHDFS API详解static FileSystem get(Configuration conf) operator() { //step1 得到Configuration对象 //step2 得到FileSystem对象 //step3 进行文件操作 } 用Java对HDFS编程北京传智播客教育 www.itcast.cn用Java对HDFS编程文件操作 1.1  上传本地文件到hadoop fs 1.2 在hadoop fs中新建文件，并写入 1.3 删除hadoop fs上的文件 1.4  读取文件 1.5 文件修改时间 2. 目录操作 2.1 在hadoop fs上创建目录 2.2 删除目录 2.3 读取某个目录下的所有文件  2.4遍历hdfs hdfs信息 查找某个文件在HDFS集群的位置 获取HDFS集群上所有节点名称信息 null北京传智播客教育 www.itcast.cn mapreduce篇 开发hadoop依赖的jar和自带的example北京传智播客教育 www.itcast.cn 开发hadoop依赖的jar和自带的example到http://hadoop.apache.org/ 下载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程序。Hadoop自带的examples.jar介绍北京传智播客教育 www.itcast.cnHadoop自带的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 在输入文件中统计文字个数的统计器。实例实例 写MapReduce程序的步骤： 1.把问题转化为MapReduce模型 2.设置运行的参数 3.写map类 4.写reduce类 例子：统计单词个数 My name is liu gang What is your nameMy 1 name 2 is 2 What 1 your 1 liu 1 gang 1 MapReduce模型北京传智播客教育 www.itcast.cnMapReduce模型 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 1 2.Reduce端 相同的key肯定会在一起。经过Reduce方法处理后， 形成最终的key/Value. name 1,name 1->name 2;运行步骤北京传智播客教育 www.itcast.cn运行步骤1）打成jar包。 2）创建一个word.txt文件 3）把word.txt文件传到HDFS上面 hadoop fs –copyFromLocal 4)执行hadoop jar <完整的类名> 5）查看执行结果 hadoop fs –text /path MapReduce执行流程MapReduce执行流程MapReduce基本流程*MapReduce基本流程JobTracker(JT)和TaskTracker(TT)简介JobTracker(JT)和TaskTracker(TT)简介再论JobTracker(JT)和TaskTracker(TT)再论JobTracker(JT)和TaskTracker(TT)JobTracker：协作作业的运行 taskTracker：运行作业划分后的任务Mapreduce原理Mapreduce原理1）一个文件file.txt 2）存储file.txt文件 3) 统计file.txt文件里面”Refund”个数 JobTracker失败JobTracker失败1）JobTracker失败在所有的失败中是最严重的一种。 2）hadoop没有处理jobtracker失败的机制。--它是一个单点故障。 3）在未来的新版本中可能可以运行多个JobTracker。 4）可以使用ZooKeeper来协作JobTracker。 TaskTracker失败TaskTracker失败1）一个TaskTracker由于崩溃或运行过于缓慢而失败，它会向JobTracker发送“心跳”。 2）如果有未完成的作业，JobTracker会重新把这些任务分配到其他的TaskTracker上面运行。 3）即使TaskTracker没有失败也可以被JobTracker列入黑名单。 Hadoop Job SchedulerHadoop Job Scheduler1）Hadoop默认的调度器是基于队列的FIFO调度器。 所有用户的作业都被提交到一个队列中，然后由JobTracker先按照作业的优先级高低，再按照作业提交时间的先后顺序选择将被执行的作业。 优点: 调度算法简单明了，JobTracker工作负担轻。 缺点: 忽略了不同作业的需求差异。 2）还用两个多用户调度器 --Fair Scheduler(公平调度器) --Capacity Scheduler(计算能力调度) Fair Scheduler(公平调度器)Fair Scheduler(公平调度器)多个Pool，Job需要被提交到某个Pool中 每个pool可以设置最小 task slot，称为miniShare FS会保证Pool的公平 Pool内部支持Priority设置 支持资源抢占 mapreduce的类型与格式北京传智播客教育 www.itcast.cn mapreduce的类型与格式Hadoop 0.20.x之前的API Hadoop的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 是用来更新计数和状态信息。 北京传智播客教育 www.itcast.cn Hadoop 0.20.x之后的API Hadoop的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, Context context ) throws IOException, InterruptedException { } //Context是上下文对象，这里Context等同于OutputCollector和Reporter两个函数的功能。 mapreduce的数据类型与java类型对应关系北京传智播客教育 www.itcast.cn mapreduce的数据类型与java类型对应关系Writable接口北京传智播客教育 www.itcast.cnWritable接口1.对Java中的int型进行封装那么就是hadoop中的IntWritable类 在写程序时可以把IntWritable可以看着是int 类型，它实现 了WritableComparable接口。 WritableComparable又是Writable、java.lang.comparable接口的子接口。 2.Writable类对所有的Java基本类型进行封装： 如：boolean -> BooleanWritable;Byte -> ByteWritable 3. 我们可以自定义Writable接口，来编写更复杂的结构的类。 核心：hadoop有自己一套的I/O机制。I/O类都必须实现Writable接口。 实现自定义的mapreduce类型北京传智播客教育 www.itcast.cn* 实现自定义的mapreduce类型 public class LogKey implements WritableComparable{ private String customerId; private String dateTime; private 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的时候，mapreduce自带的类型是不能满足我们的需求，这样我们就要自己来定制化Key和Value。 最小的MapReduce驱动北京传智播客教育 www.itcast.cn* 最小的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.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.setReducerClass(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 MinimalMapReduceWithDefaults(), args); System.exit(exitCode); }} mapreduce驱动默认的设置北京传智播客教育 www.itcast.cn mapreduce驱动默认的设置 Combiners和Partitioner编程北京传智播客教育 www.itcast.cn 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类型完全一致，且不影响最终结果的场景。比如累加，最大值等。 Combiners分析北京传智播客教育 www.itcast.cn 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要小心北京传智播客教育 www.itcast.cn 使用 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.5 17.5和15？ 所以 ：使用combiner一定要注意。 Partitioner分析北京传智播客教育 www.itcast.cn Partitioner分析Partitioner 在mapreduce的位置： mapreduce提供的Patitioner北京传智播客教育 www.itcast.cn mapreduce提供的PatitionerPartition主要作用就是将map的结果发送到相应的reduce。这就对partition有两个要求： 1）均衡负载，尽量的将工作均匀的分配给不同的reduce。 2）效率，分配速度一定要快。 mapreduce提供的Patitioner 北京传智播客教育 www.itcast.cn Partitioner是partitioner的基类，如果需要定制partitioner也需要继承该类。 2. HashPartitioner是mapreduce的默认partitioner。计算方法是 which reducer=(key.hashCode() & Integer.MAX_VALUE) % numReduceTasks，得到当前的目的reducer。 BinaryPatitioner继承于Partitioner< BinaryComparable ,V>，是Partitioner的偏特化子类。该类提供leftOffset和rightOffset，在计算which reducer时仅对键值K的[rightOffset，leftOffset]这个区间取hash。 Which reducer=(hash & Integer.MAX_VALUE) % numReduceTasks KeyFieldBasedPartitioner也是基于hash的个partitioner。和BinaryPatitioner不同，它提供了多个区间用于计算hash。当区间数为0时KeyFieldBasedPartitioner退化成HashPartitioner。 5. TotalOrderPartitioner这个类可以实现输出的全排序。不同于以上3个partitioner，这个类并不是基于hash的。在下一节里详细的介绍totalorderpartitioner。 自定义的Partitioner北京传智播客教育 www.itcast.cn 自定义的Partitioner1）为何使用Partitioner，主要是想reduce的结果能够根据key再次分类输出到不同的文件夹中。 2）结果能够直观，同时做到对数据结果的简单的统计分析。 需求： 1、输入的数据文件内容如下(1条数据内容少，1条数据内容超长，3条数据内容正常)： Kaka 1 28 hua 0 26 chao 1 tao 1 22 mao 0 29 22 2、目的是为了分别输出结果，正确的结果输出到一个文本，太短的数据输出到一个文本，太长的输出到一个文本，共三个文本输出。 Patitioner接口： public int getPartition(Text key, Text value, int numPartitions)； numPartitions为Reduce的个数。 注：在本地作业运行器上运行时，只支持0个或一个Reduce。 null北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cnnull北京传智播客教育 www.itcast.cn