Hadoop_Jordan

Size: px

Start display at page:

Download "Hadoop_Jordan"

风忘姬
6 years ago
Views:

1 大数据 Hadoop 平台 Jordan Li 08/08/2016

2 Agenda Hadoop 平台核心架构 HDFS 主要概念架构与运行机制演示 :HDFS 常用命令 & 配置 Yarn 主要概念架构与运行机制演示 :Yarn 常用命令 & 配置 Mapreduce 编程入门 & 演示 Mapreduce 调优 & 最佳实践 Hadoop Benchmark 性能测试

3 Hadoop 简介一个可扩展的容错性强的分布式文件系统和并行计算环境让用户可以便捷的处理海量数据 Apache 软件基金会 Google inspired Yahoo! 谁在用? 中移动, Intel, IBM, Facebook, Amazon, BAT... 用来做什么? 日志分析倒排索引排序图像转换广告分析...

4 Hadoop 架构

5 HDFS 简介 Hadoop Distributed File System 为大文件设计 ( 海量小文件问题 ) Write-once-read-many 访问模型不建议大量随机读, 不支持对文件修改 High throughput.vs. low latency Commodity hardware 错误检测 / 快速恢复

6 HDFS 主要概念 Namenode SecondaryNamenode StandbyNamenode Datanode Client Block, Replication

Apache Hadoop 2.7.1 HDFS Architecture http://hadoop.apache.

7 Apache Hadoop HDFS Architecture

8 HDFS 的文件操作读取文件 : lclient 向 NameNode 发起文件读取的请求 lnamenode 返回文件存储的 DataNode 的信息 lclient 读取文件信息写入文件 lclient 向 Namenode 发起文件写入的请求 lnamenode 根据文件大小和 block 配置大小返回 Datanode 的信息 lclient 将文件划分为多个 block, 按地址依次写入 Datanode

9 HDFS 文件读取基本过程

10 HDFS 文件写入基本过程

12 演示 :HDFS 基本操作 & 配置文件 sbin/start-dfs.sh stop-dfs.sh bin/hdfs dfs -ls/-mkdir/-rm/-put/-get bin/hdfs dfsadmin -report/-safemode/-refreshnodes bin/hdfs fsck -files -blocks bin/hdfs getconf -confkey <property> sbin/start-balancer.sh core-site.xml, hdfs-site.xml, hadoop-env.sh

13 HDFS High Availability

14 YARN 产生背景 MapReduce V1 的几个缺陷 l 扩展性受限 l 单点故障 l 难以支持 MR 之外的计算多计算框架各自为战 lmr: 离线计算 lstorm: 实时计算 lspark: 内存计算

15 YARN 模块组成

16 YARN 架构

17 YARN 容错机制

18 YARN 资源调度器

19 YARN 运行流程分析

20 演示 :Yarn 基本操作 & 配置文件 sbin/start-yarn.sh stop-yarn.sh bin/yarn node bin/yarn application yarn-env.sh, yarn-site.xml

21 MapReduce 简介 lnutch 项目 -- 开源的网络搜索引擎 lgoogle 公司发表的 Mapreduce 论文 ldoug Cutting 基于 java 设计开发了 mapreduce 并行计算框架和系统, 集成在 Hadoop 项目中 l 成为 Apache 旗下顶级项目 l 极为成功, 广为接受, 易于使用的大数据并行处理技术.

22 基本思想 l 分治法 : 将一个复杂问题, 分成若干个简单的子问题进行解决, 然后对子问题的结果进行合并, 得到原有问题的解. l lmap: 将大任务拆分成小的子任务, 并且完成子任务的计算 lreduce: 将中间结果合并成最终结果

23 Mapreduce 特性 l 易于编程 l 良好的扩展性 ldata Locality l 高容错性 l 提供状态监控工具

24 Mapreduce 计算框架 lmapreduce job 将输入数据切分成小块, 每一小块会分配给一个 map 处理, 多个 map 之间并行运行 lmapreduce 框架会对 map 的输出进行排序 lreduce 端会拉取 map 的处理结果, 进行后续的处理 l 输入输出都保存在 hdfs 中, 框架本身会调度相应的 map/reduce task, 并进行监控和容错. lmap 和 reduce 的输入输出都是 <key,value> 对

25 Mapreduce 基本操作 & 配置文件 bin/hadoop jar *.jar <main class> <input path> <output path> Job history server mapred-site.xml

26 Wordcount Example

29 Wordcount Example Job job = new Job(conf, "word count"); 4. 构建一个 job, 加载 mapred-default.xml,mapred-site.xml,yarndefault.xml,yarn-site.xml, mapreduce 框架里面一个 mapreduce 叫做一个 job, 具体的 map 和 reduce 叫做 task. 随后写入 conf 中的配置 (GenericOptionsParser 解析出来的 ), Job 的名字是 word count 这个名字会显示在前端 UI 上

30 Wordcount Example job.setjarbyclass(wordcount.class); hadoop 需要通过这个 class 定位到所在的 jar 包, 传到 hdfs. 原理 : 通过 classloader 获取类路径下的资源文件, 循环遍历, 定位到 class.

31 job.setmapperclass(tokenizermapper.class); job.setcombinerclass(intsumreducer.class); job.setreducerclass(intsumreducer.class); job.setoutputkeyclass(text.class); job.setoutputvalueclass(intwritable.class); for (int i = 0; i < otherargs.length - 1; ++i) { FileInputFormat.addInputPath(job, new Path(otherArgs[i])); } FileOutputFormat.setOutputPath(job, new Path(otherArgs[otherArgs.length - 1])); System.exit(job.waitForCompletion(true)? 0 : 1);

32 未写出的默认配置 ljob.setinputformatclass(textinputformat.class); ljob.setoutputformatclass(textoutputformat.class); ljob.setpartitionerclass(hashpartitioner.class) ljob.setnumreducetasks(1); linputformat 负责数据切分 (split) 以确定 map task 的个数及对应的 split, 还负责为 mapper 提供输入数据, 将 split 解析成 key,value 对. linputformat 将分片的数据序列化到磁盘供后续读取 l 文件格式 : textfile, sequencefile, Avro, rcfile, orcfile, 等 loutputformat 参数和 InputFormat 相对, 将 key,value 写入特定格式的文件中. lpartition 对 map 处理的结果进行分片, 每一片的所有数据交由同一 reduce 处理 lpartition 个数 =reduce 个数

34 Mapper 类讲解 public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> 定义一个自己的 Map 过程, 类名是 TokenizerMapper, 它继承了 Hadoop 的 Mapper 类 <Object, Text, Text, IntWritable> 这里 : Object, 表示输入键 key 的参数类型, 值是每行的 offset Text, 表示输入值的类型, 值是行字符串 Text, 表示输出键类型, 值是被空格区分后的一个 word IntWritable, 表示输出值类型

35 Hadoop 序列化 l 序列化 : 将结构化对象转化为字节流以便于通过网络进行传输或写入持久存储的过程. l 反序列化 : 字节流转为结构化对象的过程. l<key, value> 必须是可序列化的 : 实现 Writable lkey 需要排序 ( 比较 ): 实现 Comparable

36 Java 基本类型对应的 Writable

37 map 方法讲解 private final static IntWritable one = new IntWritable(1); private Text word = new Text(); public void map(object key, Text value, Context context ) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString()); while (itr.hasmoretokens()) { word.set(itr.nexttoken()); context.write(word, one); } }

38 Reducer 类讲解 public static class IntSumReducer extends Reducer<Text,IntWritable,Text,IntWritable> { private IntWritable result = new IntWritable(); } public void reduce(text key, Iterable<IntWritable> values, Context context ) throws IOException, InterruptedException { int sum = 0; for (IntWritable val : values) { sum += val.get(); } result.set(sum); context.write(key, result); }

39 Map task 执行流程

40 Reduce task 执行流程

41 Shuffle lmap 的输出作为 reduce 输入的过程 lmap 阶段 lmap 输出到内存中一块环形缓冲区中 l 缓冲区达到指定阈值, 触发 spill 操作, 溢写到磁盘上的一个文件, 一次 spill 会生成一个文件 lspill 操作又包含了 sort( 快排 ) 和 combiner, sort 优先级为 partition,key. 对每个 partiton 中数据进行一次 combiner l 保存分区数据的元信息, 包括偏移量, 数据大小 l 合并多个文件成一个大而有序 ( 堆排序 ) 的文件 l 删除临时的 spill 文件, 合并完成

42 Shuffle lreduce 阶段 lcopy,reduce 从 map 上远程拷贝一片数据, 只要一个 map 完成就进行拷贝 ( 并行 ). l 对多份拷贝数据文件进行 merge, 同时排序, 防止磁盘上文件过多. l 归并排序所有数据, 确保给 reduce 的数据是有序的, 聚集的.

43 Run WordCount Example lbin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduceexamples jar wordcount input output lwordcount 参数其实是通过反射调用了 WordCount 类, 这里直接指定要用的 class name: org.apache.hadoop.examples.wordcount 效果是一样的.

44 Hadoop Generic Options

45 HDFS+Yarn+Mapreduce 监控 Namenode:50070 Resourcemanager:8088 Jobhistoryserver:19888

46 MapReduce On YARN 工作原理 l 用户向 Yarn 提交一个 MapReduce 应用,Yarn 会分两阶段运行 lresourcemanager 启动 MRAppMaster l(mapreduce 的 ApplicationMaster) 负责管理作业生命周期. lmrappmaster 创建应用程序, 申请资源, 并监控其运行过程, 直到运行成功 lmrappmaster3 种运行模式 : 本地, uber, 非 uber

47 MapReduce on YARN 工作原理

48 MapReduce On YARN 工作原理 1) 用户向 YARN 中提交应用程序 / 作业, 包括 MRAppMaster 程序启动 MRAppMaster 的命令用户程序等 2)ResourceManager 为作业分配第一个 Container, 并与对应的 NodeManager 通信, 要求它在这个 Containter 中启动该作业的 MRAppMaster; 3)MRAppMaster 首先向 ResourceManager 注册, 这样用户可以直接通过 ResourceManager 查询作业的运行状态 ; 然后它将为各个任务申请资源并监控任务的运行状态, 直到运行结束即重复步骤 4-7; 4)MRAppMaster 采用轮询方式通过 RPC 请求向 RM 申请和领取资源 ; 5)MRAppMaster 申请到资源后, 便与对应的 NM 通信, 要求它启动任务 ; 6)NodeManager 使用脚本启动任务 ( 环境变量,jar 包等 ); 7) 各个任务通过 RPC 协议向 MRAppMaster 汇报自己的状态和进度, 以让 MRAppMaster 随时掌握各任务的运行状态, 从而可以在任务失败时重新启动任务 ; 在作业运行过程中, 用户可随时通过 RPC 向 ApplicationMaster 查询作业当前运行状态 ; 8) 作业完成后,MRAppMaster 向 ResourceManager 注销并关闭自己 ;

49 最佳实践 Hadoop 压缩 l 压缩算法使用 : lmap 输入, map 输出, reduce 输出 lmap 输入需要保证压缩算法可分割, 解压速度快 lmap 输出保证压缩率高, 压缩解压缩速度快 lreduce 输出保证压缩解压缩速度快, 压缩率高压缩率 : gzip>lzo>snappy 速度 : gzip<lzo<snappy

50 最佳实践 Distribute Cache l 分布式缓存类, 用来分发资源到 Nodemanager 上. l 为每一个资源生成一个唯一的 id, 防止重复下载 l 使用方式 :api 方式命令行方式 l 使用场景 : 工具 jar, join 小表 ( 配置文件 ) 等

51 最佳实践 Mapreduce Counter l 一个计数器 l 便于我们发现运行的 mapreduce 问题 l 自定义 counter 用于排错, 配合 system.out l 出于监控, 报警, 统计等需求, 在程序里获取 counter 值 lcounter counter = job.getcounters().getgroup("file System Counters").findCounter("HDFS: Number of bytes read"); lcounter.getvalue();

52 最佳实践推测执行机制 l 同一个 job 的多个 task 之间运行速度不一致, 为防止这些慢 ltask 拖慢整个 job 的进度. lhadoop 为该任务启动一个备份任务, 谁先运行完, 则将谁的结果作为最终结果. l 每个 task 最多有一个备份任务实例

53 最佳实践推测执行机制 l 已经完成的 task 数目比例不小于 5% l 启动推测 task 的总量不超过 10 l 推测执行算法可自己实现, 默认实现算法 : 周期性扫描所有正在运行的 task, 假设为一个 task 启动一个备份 task, 估算完成时间, 估算该 task 正常运行的完成时间, 时间差越大, 证明越有价值为这个 task 启动备份任务.

54 最佳实践推测执行机制 lmapreduce.map.speculative true lmapreduce.reduce.speculative true l 关闭情况 : lmapreduce 处理数据本身不均衡或计算不均衡 l 特定的任务, 如写入 hbase

55 最佳实践 Mapreduce 执行优化

56 Mapreduce 运行内存 lmapreduce.map.memory.mb lmapreduce.reduce.memory.mb l 对 map 和 reduce 任务的内存限制 lmapreduce.map.java.opts lmapreduce.reduce.java.opts l 对 jvm-xmx 堆大小的限制, 但 task 不一定是 java 程序 (streaming 中 python,perl 等 ), 为 native lib 留出余地 ( 比如压缩等 ) lmemory.mb 要大于 java.opts 否则会报内存溢出

57 Mapreduce 运行内存 ljvm 的内存溢出 lmapreduce.map.java.opts lmapreduce.reduce.java.opts l 改大的同时另外两个 memory 的参数也要相应增加. l-dmapreduce.map.java.opts=-xmx2048m l-dmapreduce.map.memory.opts=2560

58 Mapreduce 运行内存 lyarn.scheduler.minimum-allocation-mb lyarn.scheduler.maximum-allocation-mb l 计算节点 Container 的物理内存范围, 一般不需要改, 改动需要重启集群生效. lyarn.nodemanager.vmem-pmem-ratio lcontainer 允许的虚拟内存值, 改动需要重启集群生效

59 Mapreduce 运行内存 lerror: Container %s is running beyond physical/virtual memory limits lyarn.nodemanager.vmem-check-enabled lyarn.nodemanager.pmem-check-enabled false( 非常不建议, 重启集群 ) l 提交 Job 时可动态配置 -Dmapreduce.map.memory.mb lvirtual memory 溢出时, yarn.nodemanager.vmem-pmem-ratio 默认 2.1, 改大, 需要重启集群.

60 最佳实践参数调优 HDFS qname Node/Data Node Yarn qapplication Master qresource Manager qnode Manager qcapacity/fair Scheduler MapReduce qmap/reduce

62 Hadoop Benchmark 性能测试 # 使用 6 个 Map 任务并行向 HDFS 里 6 个文件里分别写入 1GB 的数据 bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-clientjobclient tests.jar TestDFSIO -write -nrfiles 6 -size 1GB # 使用 6 个 Map 任务并行从 HDFS 里 6 个文件里分别读取 1GB 的数据 bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-clientjobclient tests.jar TestDFSIO -read -nrfiles 6 -size 1GB # 清除以上生成的数据 bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-clientjobclient tests.jar TestDFSIO -clean

63 Hadoop Benchmark 性能测试 # 生成 1000 万行数据到 /terainput 路径中 bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples jar teragen /terainput # 将 /terainput 中生成的 1000 万行数据排序后存入到 /teraoutput 路径中 bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples jar terasort terainput /teraoutput # 针对已排序的 /teraoutput 中的数据, 验证每一行的数值要小于下一行 bin/hadoop jar share/hadoop/mapreduce/hadoop-mapreduce-examples jar teravalidate -D mapred.reduce.tasks=8 /teraoutput /teravalidate # 查看验证的结果 bin/hdfs dfs -cat teravalidate/part-r-00000

64 Secondary Sort Example

65 Secondary Sort Example ljob.setpartitionerclass(firstpartitioner.class); ljob.setgroupingcomparatorclass(firstgroupingcomparator.class); ljob.setsortcomparatorclass(intpair.comparator.class);// 默认的

66 Secondary Sort Example l 应用 : l1. 二次排序, 第一字段优先, 第二字段其次的规则排序 l2. 希望在同一 reduce group 中有序的处理不同 map 标记的数据. l 输入 : Table1(today) Table2(history) l userid tag1,tag2 userid tag2,tag3 l 输出 : userid tag1

67 Agent Day 2 Hadoop 快速入门指南 HBase 快速入门指南 Zookeeper 快速入门指南 Kafka 快速入门指南

68 什么是 ZooKeeper 分布式协调服务是很困难的 (race conditions, deadlock) ZK 是一个为分布式应用提供一致性服务的软件, 提供的功能包括 : 配置维护分布式同步组和命名服务等本身是分布式的, 是 Google Chubby 的一个开源实现是大数据平台如 Hadoop Hbase 的重要组件容器平台使用 etcd 或 consul 做协调服务

69 系统架构奇数个节点, 必须有半数节点以上是活的每个节点数据都一样基于内存的, 持久化有 transaction logs 和 snapshot 树状数据结构 - znode Nodes and ephemeral nodes 不是做存储用的!

70 系统架构

71 系统架构

72 集群管理生产环境一般 5 个节点每个节点需配置 zoo.cfg bin/zkserver.sh start zookeeper Commands: The Four Letter Words echo mntr nc localhost 2181

73 Agent Day 2 Hadoop 快速入门指南 ZooKeeper 快速入门指南 Kafka 快速入门指南

74 Kafka 快速入门核心概念与架构集群管理编程入门

75 什么是 Kafka 消息队列 : 缓解消息生产者与消息消费者之间不同步问题的一个缓冲区 Kafka 是一个分布式可分区多副本的高吞吐量低延迟的消息系统 3 Kafka servers, 1 producer - 821,557 records/sec(78.3 MB/sec)* 适用于处理活跃的流式数据, 比如网页的访问量日志等 *

76 架构 Producer, consumer, broker

77 架构 topic, partition

78 架构 consumer group

79 编程入门快速检查集群 Java 生产消费示例

83 关注我们 QingCloud-IaaS 青云 QingCloud

84 Thank you.

MapReduce

MapReduce 陳威宇 Review Hadoop Hdfs Datanode Namenode files / blocks Data locality ( 在地運算 ) 2 Outline What is MapReduce Process flow Yarn Configuration Java programing 3 MapReduce Introduction Objective :