极客大学架构师训练营 大数据架构 MapReduce Yarn Hive SQL 第24课 听课总结
说明
讲师:首席架构师 李智慧
交互方式的趋势:根据大数据分析用户喜好什么,就推荐什么。比如字节跳动推荐新闻,视频;比如淘宝推荐商品给用户。
MapReduce:大规模数据处理
- 处理海量数据(> 1TB)
- 成百上千 CPU 实现并行处理
简单地实现以上目的:移动计算比移动数据更划算
分而治之(Divide and Conquer)
MapReduce 特性
- 自动实现分布式并行计算
- 容错
- 提供状态监控工具
- 模型抽象简洁,程序员易用
MapReduce 的组成程序
MapReduce 由称为 Map 和 Reduce 的两部分用户程序组成,然后利用框架在计算机集群上面根据需求运行多个程序实例来处理各个子任务,然后再对结果进行归并。
WordCount 举例
文本前期处理
str_list = str.replace('\n', '').lower().split(' ')
count_dict = {}
如果字典里有该单词则加1,否则添加入字典
for str in str_list:if str in count_dict.keys():count_dict[str] = count_dict[str] + 1else:count_dict[str] = 1
MapReduce 的 WordCount
public class WordCount {public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {private final static IntWritable ont = 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);}}}
}
Class TokenizerMapper
的核心内容:
- 框架调用方法
Map
, - 输入参数:
key
(一行的第一个字符的偏移量),value
(文件输入的一行),context
(结果输出对象) - 输出结果:
word
(单词),one
就是序列化的1
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);}
}
class IntSumReducer 的核心内容:
- 框架调用方法: reduce。
- 入参:key就是Map输出的word, values就是1组成的list。
- 结果:key还是Map输出的word,result就是单词的词频。
十万亿行的文本文件,用几分钟到十几分钟就算出了词频。(看集群的大小)
public static void main(String[] args) throws Exception {Configuration conf = new Configuration(); //得到集群配置参数Job job = Job.getInstance(conf, "WordCount"); //设置到本次的job实例中job.setJarByClass(WordCount.class); //指定本次执行的主类是 WordCountjob.setMapperClass(TokenizerMapper.class); //指定Map类job.setCombinerClass(IntSumReduce.class); //指定Combiner类job.setReducerClass(IntSumReducer.class); //指定Reducer类job.setOutputKeyClass(Text.class);jobjob.setOutputValueClass(IntWritable.class);FileInputFormat.addInputPath(job, new Path(args[0])); //指定输入数据的路径FileOutputFormat.setOutputPath(job, new Path(args[1])); //指定输出路径System.exit(job.waitForCompletion(true) ? 0 : 1); //指定job执行模式,等待任务执行完成后,提交任务的客户端才会退出!}
setCombineClass 在Map端输出的时候,把数据求和以后再传给Reduce,可以减少传输的数据量。
Map 的入参数,第一行第一个参数的偏移量是0, 第二个参数为 Hello World;第一行第一个参数的偏移量是12, 第二个参数为 Bye World.
Map的输出:word 为 Hello,数值为1; word 为 World,数值为1…
Reduce 的输入参数为:word list;
Reduce 的输出为:相同的word 累加。
分布式计算的过程:
- input HDFS 分片,分给给Map集群;
- Map 对数据进行分词;
- Map 用网络copy 到 Reduce 集群;
- Reduce 进行排序Sort 和 合并 Merge;
- 对结果聚合进行输出 Output HDFS。
Block 是HDFS的术语, input split 是 MapReduce的术语。
换个视角看过程,跟上面分析一样。
Shuffle process: 相同的key 传输到 Reduce里面去;所有的框架都有这个重要的过程,切记!切记!切记! Shuffle!
再换个视角看MapReduce过程。跟上面分析一样。
MapReduce 1 整体架构
- main函数启动,作业jar包存储在HDFS,存储到DataNode进程;
- 提交作业到JobTracker服务器, JobScheduler作业要排队;
- 创建JobInprocess (JIP);
- 创建TaskInProcess(一个split 就是一个TaskInProcess);
- TaskTracker & DataNode服务器如果有空闲,发送心跳通信告诉 JobTracker 服务器;
- JobTracker 分发一个task任务到 TaskTracker & DataNode服务器。
- TaskTracker & DataNode服务器 启动一个 TaskRunner,启动map 进程,或者启动reduce进程;
- 第一次领任务的时候,要去HDFS去下载job.jar 包;下载完成以后,本地通过反射加载map函数,或者reduce函数;
- 读取分片的数据块,或者输出计算结果数据。
2020主流用的是 MapReduce 2, 核心架构是变化不大的。
适合 MapReduce 的计算模型
- TopK
- K-means
- Bayes
- SQL
不适合 MapReduce 的计算模型
- Fibonacci
InputFormat
- 验证作业的输入的正确性;
- 将输入文件切分成逻辑的 InputSplits,一个 InputSplit 将被分配给一个单独的 Mapper task;
- 提供 RecordReader 的实现,这个 RecordReader 会从 InputSplit 中正确读出一条一条的 K - V 提供给 Mapper 使用。
FileInputFormat
- 得到分片的最小值
minSize
和最大值maxSize
,可以通过设置mapred.min.split.size
和mapred.max.split.size
来设置; - 对于每个输入文件,计算
max(minSize, min(maxSize, blockSize))
; - 如果
minSize <= blockSize <= maxSize
, 则设为blockSize
. - 分片信息
<file, start, length, hosts>
, 通过 hosts 实现 map 本地性。
分片参数意义解析:
- file: 文件路径
- start:开始位置
- length:数据长度
- hosts: 记录InputSplit的分片信息在哪台机器上
OutputFormat
- OutputFormat 接口决定了在哪里以及怎样持久化作业结果。
- 默认的 OutputFormat 就是 TextOutputFormat, 它是一种以行分割,包含制表符界定的键值对的文本文件格式。
Partitioner
什么是 Partitioner
- MapReduce 通过 Partitioner 对 Key 进行分区,进而把数据按我们自己的需求来分发。
什么情况下使用 Partitioner
- 如果你需要 key 按照自己意愿分发,那么你需要这样的组件。
- 框架默认的 HashPartitioner。
- 例如:数据内包含省份,而输入要求每个省份输出一个文件。
public int getPartition(K key, V value, int numReduceTasks) {return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
}
主要调度方法
单队列调度
- 特点:FIFO
- 优点:简单
- 缺点:资源利用率低
容量调度(Capacity Scheduler,Hadoop-0.19.0)
- 特点:
☞ 多队列,每个队列分配一定系统容量(Guaranteed Capacity)
☞ 空闲资源可以被动态分配给负载中的队列
☞ 支持作业优先级 - 作业选择:
☞ 选择队列:资源回收请求队列优先;最多自由空间队列优先。
☞ 选择作业:按提交时间、优先级排队;检查用户配额;检查内存。 - 优点:
☞ 支持多作业并行执行,提高资源利用率
☞ 动态调整资源分配,提高作业执行效率 - 缺点:
☞ 队列设置和队列选择无法自动进行,用户需要了解大量系统信息
公平调度(Fair Scheduler,Hadoop-0.19.0)
- 目标:
☞ 改善小作业的影响时间。
☞ 确保生产性作业的服务水平。 - 特点:
☞ 将作业分钟 – 形成作业池(based on a configurable attribute, such as user name, unix group…)。
☞ 给每个作业池分配最小共享资源(Minimum map slots, Minimum reduce slots)。
☞ 将多余的资源平均分配给每个作业。 - 作业选择:
☞ 优先调度资源小于最小共享资源的作用。
☞ 选择分配资源与所需资源插件最大的作业。 - 优点:
☞ 支持作业分类调度,使不同类型的作业获得不同的资源分配,提高服务质量。
☞ 动态调整并行作业数量,充分利用资源。 - 缺点:
☞ 不考虑节点的实际负载状态,导致节点负载实际不均衡。
JobTracker 内部实现
- 作业控制
- 作业抽象成三层:作业监控层(Job In Progress JIP),任务控制层(Task In ProgressTIP),任务执行层。
- 任务可能会被尝试多次执行,每个任务实例被称作 Task Attempt(TA)。
- Task Attempt成功,TIP 会标注该任务成功,所有TIP成功,JIP 成功。
- 资源管理
- 根据 TaskTracker 状态信息进行任务分配。
JobTracker 容错
- JobTracker 失败,那么未完成 Job 失败;
- 通过 Job 日志,Job 可部分恢复。
TaskTracker 容错
- 超时
- TaskTracker 10 分钟(
mapred.tasktracker.expiry.interval
)未汇报心跳,则将其从集群移除。
- 灰名单,黑名单
- TaskTracker 上部署性能监控脚本
- 如果性能表现太差,被 JobTracker 暂停调度
Task 容错
- 允许部分 Task 失败
- 允许失败的任务占比,默认 0,
mapred.max.map.failers.percent
,mapred.max.reduce.failers.percent
.
- Task 由 TIP 监控,失败任务多次尝试,慢任务启动备份任务
- 每次都是一个 TA(Task Attempt),最大允许尝试次数:
mapred.map.max.attempts
、mapred.reduce.max.attempts
Record 容错
跳过导致 Task 失败的坏记录:
- K,V 超大,导致 OOM(Out of Memory, 内存溢出),超出截断
mapred.linerecordreader.maxlength
. - 异常数据引发程序 bug, task 重试几次后,自动进入 skip node, 跳过导致失败的记录,
mapred.skip.attempts.to.start.skipping
.
Yarn (Yet Another Resource Negotiator)
MapReduce 2.0 框架就是基于 Yarn的。
不再是一个传统的 MapReduce 框架,甚至于 MapReduce 无关。
一个通用的运行时框架,用户可以编写自己的计算框架,在该运行环境中运行。
Yarn 统一资源管理。
Yarn 解决的问题
MapReduce 的架构,在 MapReduce 应用程序的启动过程中,最重要的就是要把 MapReduce 程序分发到大数据集群的服务器上,在 Hadoop 1 中,这个过程主要是通过 TaskTracker 和 JobTracker 通信来完成。
这种架构方案的主要缺点是,服务器集群资源调度管理和 MapReduce 执行过程耦合在一起,如果想在当前集群中运行其它计算任务,比如 Spark 或者 Storm,就无法统一使用集群中的资源了。
在 Hadoop 早期的时候,大数据技术就只有 Hadoop 一家,这个缺点并不明显。但随着大数据技术的发展,各种新的计算框架不断出现,我们不可能为每一种计算框架部署一个服务器集群,而且计算能部署新集群,数据还是在原来集群的 HDFS 上。所以我们需要把 MapReduce 的资源管理和计算框架分开,这也是 Hadoop 2 最主要的变化,就是将 Yarn 从 MapReduce 中分离出来,成为一个独立的资源调度框架。
MRv2 最基本的设计思想就是将 JobTracker 的两个主要功能,即资源管理和作业管理分成两个独立的进程。
- 在该解决方案中包含两个组件:全局的 ResourceManager (RM)和 与每个应用相关的 ApplicationMaster (AM)。
- 这里的 “应用” 指一个单独的 MapReduce 作业或者 DAG 作业。
Yarn 架构
Resource Manager:资源管理器。
Node Manager:节点管理器。
资源管理器,从 节点管理器获取容器资源,容器调用ApplicationMaster去处理应用程序。资源管理器可以启动多个不同类型的容器。
Job.jar 分发给ApplicationMaster,ApplicationMaster 启动TaskProcess, 反射加载Job.jar,调用Map方法,或者Reduce方法。
ApplicationMaster 作业完成,释放容器资源。
JobTracker和TaskTracker 是动态启动的。MapReduce 1.0是静态启动的。
Yarn 包括两个部分:资源管理器(Resource Manager) 和 节点管理器(Node Manager)
这也是 Yarn 的两种主要进程: ResourceManager 进程负责整个集群的资源调度管理,通常部署在独立的服务器上,NodeManger 进程负责具体服务器上的资源和任务管理,在集群的每一台计算服务器上都会启动,基本上跟 HDFS 的 DataNode 进程一起出现。
资源管理器又包括两个主要组件:调度器和应用程序管理器
调度器其实就是一个资源分配算法,根据应用程序(Client)提交的资源申请和当前服务器集群的资源情况进行资源分配。Yarn 内置了几种资源调度算法,包括 Fair Scheduler、Capacity Scheduler 等。你也可以开发自己的资源调度算法供 Yarn 调用。
Yarn 进行资源分配的单位是容器(Container),每个容器包含了一定量的内存、CPU 等计算资源,默认配置下,每个容器包含一个 CPU 核心。容器由 NodeManger 进程启动和管理,NodeManger 进程会监控本节点上容器的运行状况并想 ResourceManger 进程汇报。
应用程序管理器负责应用程序的提交、监控应用程序运行状态等。应用程序启动后需要在集群中运行一个 ApplicationMaster,ApplicationMaster 也需要运行在容器里面。每个应用程序启动后都会先启动自己的 ApplicationMaster,由 ApplicationMaster 根据应用程序的资源需求进一步向 ResourceManager 进程申请容器资源,得到容器以后就会分发自己的应用程序代码到容器上启动,进而开始分布式计算。
Yarn 的工作流程(MapReduce为例)
- 我们向 Yarn 提交应用程序,包括 MapReduce ApplicationMaster、我们的MapReduce 程序,以及 MapReduce Application 启动命令。
- ResourceManger 进程和 NodeManger 进程通信,根据集群资源,为用户程序分配第一个容器,并将 MapReduce ApplicationMaster 分发到这个容器上面,并在容器里面启动 MapReduce ApplicationMaster。
- MapReduce ApplicationMaster 启动后立即向 ResourceManager 进程注册,并为自己的应用程序申请容器资源。
- MapReduce ApplicationMaster 申请到需要的容器后,立即和相应的 NodeManger 进程通信,将用户 MapReduce 程序分发到 NodeManger 进程所在的服务器,并在容器中运行,运行的就是 Map 或者 Reduce 任务。
- Map 或者 Reduce 任务在运行期和 MapReduce ApplicationMaster 通信,汇报自己的运行状态,如果运行结束,MapReduce ApplicationMaster 向 ResourceManager 进程注销并释放所有的容器资源。
资源管理器 HA
Hive
如何用MR 实现下面语句?
分析的场景:为什么25岁的用户喜好pageid = 2 的页面。那么就可以用例推该页面,那么PV就会更多,广告的转化率就更高了。
就是这么简单
hive > create table pokes (foo int, bar string);
hive > show tables;
hive > alter table pokes add columns (new_col int);
hive > drop table pokes;
hive > load data local inpath './examples/files/kv1.txt' overwrite into table pokes;
hive > select a.foo from invites a where a.ds='2008-08-08';
Hive 架构
表的结构、数据信息记录到Metastore里面。
核心是 Driver要把SQL,Compiler 转换为MapReduce能够处理的数据结构。
Hive 执行流程
- 操作费(Operator)是Hive的最小处理单元;
- 每个操作符处理代表 HDFS 操作或者 MR 作业
- 编译器把 Hive SQL 转换成一组操作符
Hive 编译器
![在这里插入图片描述](https://img-blog.csdnimg.cn/20200829212426391.png#pic_center =6=500x)
Example Query(Filter)
Filter status updates containing 'Michael Jackson';select * from status_updates where status like 'Michael Jackson';
提前预置好各种 Operator,根据语法的执行计划去调用各种 Operator。
这个Job里面只有Map,没有Reduce。因为只有一个where条件。里面有三个Operators。
- TableScanOperator 把* 转换为需要查询的条件;
- FilterOperator 是做条件过滤;
- FileOutputOperator 把结果输出。根据Bool值返回结果:false 没有输出;true 返回字符串结果。
Example Query(Aggregation)
Figure out total number of status_updates in a given day
select count(1) from status_updates where ds = '2009-08-01';
GroupByOperator: setCombineClass 在Map端输出的时候,把数据求和以后再传给Reduce,可以减少传输的数据量。
Example Query(multi-group-by)
from (select a.status, b.school, b.gender from status_updates a join profiles bon (a.userid = b.userid and a.ds='2009-03-20')) subqlinsert overwrite table gender_summary partition(ds='2009-03-20')select subql.gender, count(1) group by subql.genderinsert overwrite table school_summary partition(ds='2009-03-20')select subql.school, coung(1) group by subql.school
一次数据访问,多次数据统计。在大数据场景下,读数据是个重工作。所以有一个From,两个insert。
Hive MetaStore
Multi User Mode
Remote User Mode
Hive QL – Join
insert overwrite table pv_usersselect pv.pageid, u.age from page_view pv join user u on (pv.userid = u.userid);
Hive QL – Join in MapReduce
第一张表
key是偏移量,不重要。
value 就是一行记录。key 是 userId,value是<来源于哪张表id,pageId>
第二张表
key是偏移量,不重要。
value 就是一行记录。key 是 userId,<user表的id,age>
Shuffle Sort:相同的key放在一起。
Reduce:每张表相同的key要进行join操作,最后把结果输出。
Join Optimizations
Map Joins
- User specified small tables stored in hash tables on the mapper backed by jdbc
- No reducer needed
insert into table pv_users
select /*+ mapjoin(pv) */ pv.pagedid, u.age from page_view pv join user u
on (pv.userid = u.userid);
Hive QL – Map Join
技巧:如果有一张表比较小,在Map的时候,就把小表加载进来。那么只用Map就可以计算出结果。(不需要Reduce操作。)
极客大学架构师训练营 大数据架构 MapReduce Yarn Hive SQL 第24课 听课总结相关推荐
- Java架构师,大数据架构师,高并发设计模式,机器学习知识点分享
第一章:java精品课程目录大全 1.亿级流量电商详情页系统的大型高并发与高可用缓存架构实战 1课程介绍以及高并发高可用复杂系统中的缓存架构有哪些东西?32分钟 2基于大型电商网站中的商品详情页系统贯 ...
- java架构师和大数据架构师有哪些不同
Java是我们熟悉的编程语言,大数据是当今科学技术的明星技术.Java和Java大数据架构的内容是否相同??两者有什么不同呢?今天小编就从Java和大数据架构的以下方面谈谈两者的区别. Java架构方 ...
- 极客大学产品经理训练营:业务架构与概念模型 第10课总结
讲师:邱岳 1. 什么是状态图 系统中的概念或对象随着一些事件的发生,改变了状态,通常这个状态比较多.比较复杂时,我们需要用状态图来表示他们的边界,以及触发状态转换的活动. 2. 动手画 – 员工状态 ...
- 极客大学产品经理训练营:PRD 产品原型文档框架 第12课总结
讲师:邱岳 1. PRD 产品原型文档的各种元素 PRD: Product Requirement Document 产品原型文档 基本元素(目录.标题.作者.版本.历史等). 场景描述. 需求背景. ...
- 20200814极客时间从零学习大数据
极客时间从零学习大数据: Yarn将MapReduce执行引擎和资源调度分离开来, yarn是大数据平台上最主流的资源调度系统 MapReduce,Spark 这类计算框架处理的业务场景都背称作批处理 ...
- .NET 云原生架构师训练营(系统架构)--学习笔记
▲ 点击上方"DotNet NB"关注公众号 回复"1"获取开发者路线图 学习分享 丨作者 / 郑 子 铭 这是DotNet NB 公众号的第176篇原创文章 ...
- 极客大学产品经理训练营 极客时间购买课程-大作业
1. 标题作者修改历史 标题:[极客时间]购买课程 作者 历史 时间 易筋 创建 2021-01-09 易筋 添加购买流程图6 2021-03-02 易筋 添加购买时序图7 2021-03-16 2. ...
- 【送书福利-第四期】从程序员到架构师:大数据量、缓存、高并发、微服务、多团队协同等核心场景实战书籍
大家好,我是洲洲,欢迎关注,一个爱听周杰伦的程序员.关注公众号[程序员洲洲]即可获得10G学习资料.面试笔记.大厂独家学习体系路线等-还可以加入技术交流群欢迎大家在CSDN后台私信我! 送书福利-第四 ...
- java 用户日志_2020年阿里巴巴最全Java、架构师、大数据、算法PPT技术栈图册
我只截图不说话,PPT大全,氛围研发篇.算法篇.大数据.Java后端架构!除了大家熟悉的交易.支付场景外,支撑起阿里双十一交易1682亿元的"超级工程"其实包括以下但不限于客服.搜 ...
- 极客大学产品经理训练营:数据分析与商业分析,商业分析到业务分析 第18课总结
讲师:邱岳 1. 产品经理眼中的利润.成本.收入 利润 = 收入 - 成本 奶茶利润率极高,达到60%左右.但是奶茶盈利比较难. 所有买水的产品利润率都极高,比如可口可乐,咖啡,奶茶等. 案例:有个面 ...
最新文章
- 【矩阵乘法】Quad Tiling(poj 3420)
- 26、临时表的创建和重复数据的处理
- java实现递归层次遍历_Java实现二叉树的前序、中序、后序、层序遍历(递归方法)...
- 《PWA实战:面向下一代的Progressive Web APP》读书笔记
- mysql 建表报错_mysql innodb 引擎 ,建表时报错:ERROR 1118 (42000)
- 如何利用计算机模拟分子生物学,虚拟分子生物学学习实验室构建
- Oracle 10g 中 X$KCVFH 说明
- JAVA classpath jar问题[zz]
- IDEA Debug 无法进入断点的解决方法
- 第2期--从零开始仿制新浪微博---android
- 如何使用水经注下载地图
- win10跳过计算机密码,Win10怎么取消开机密码?Win10如何跳过开机密码?
- 无线通信信号的功率表示方法中dBm,dB与w的换算关系
- 测量MATLAB安装哪些产品,matlab需要安装哪些工具箱
- maven镜像源及代理配置
- Ransomware的斗士——云备份系统
- iphone上app store切换国家和地区
- YOLOX论文讲解和无人机检测项目实战
- Bailian4005 拼点游戏【贪心】
- Android APP签名和签名等信息查看
热门文章
- python小熊代码_Android开源项目-小熊日记讲解【附源码】
- 逸管家坚守诚信是中小企业走上持续发展道路的准则
- ubuntu 18.04安装搜狗输入法
- VMware虚拟机、centos以及securecrt的安装
- 异步读写之利用完成历程
- 中拉光伏产业合作蓬勃发展前景远大
- 【读书笔记《Bootstrap 实战》】6.单页营销网站
- CABasicAnimation添加动画离开屏幕就动画停止的问题
- 在ASP.NET MVC项目中使用React
- ASP.NET MVC+EF框架+EasyUI实现权限管理系列(8)-DbSession线程内唯一