说明

讲师：首席架构师 李智慧



交互方式的趋势：根据大数据分析用户喜好什么，就推荐什么。比如字节跳动推荐新闻，视频；比如淘宝推荐商品给用户。



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] + 1
	else:
		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); //指定本次执行的主类是 WordCount
	job.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 累加。





分布式计算的过程：

1. input HDFS 分片，分给给Map集群；

2. Map 对数据进行分词；

3. Map 用网络copy 到 Reduce 集群；

4. Reduce 进行排序Sort 和 合并 Merge；

5. 对结果聚合进行输出 Output HDFS。



Block 是HDFS的术语， input split 是 MapReduce的术语。



换个视角看过程，跟上面分析一样。



Shuffle process： 相同的key 传输到 Reduce里面去；所有的框架都有这个重要的过程，切记！切记！切记！ Shuffle！



再换个视角看MapReduce过程。跟上面分析一样。



MapReduce 1 整体架构



1. main函数启动，作业jar包存储在HDFS，存储到DataNode进程；

2. 提交作业到JobTracker服务器， JobScheduler作业要排队；

3. 创建JobInprocess （JIP）；

4. 创建TaskInProcess（一个split 就是一个TaskInProcess）；

5. TaskTracker & DataNode服务器如果有空闲，发送心跳通信告诉 JobTracker 服务器；

6. JobTracker 分发一个task任务到 TaskTracker & DataNode服务器。

7. TaskTracker & DataNode服务器 启动一个 TaskRunner，启动map 进程，或者启动reduce进程；

8. 第一次领任务的时候，要去HDFS去下载job.jar 包；下载完成以后，本地通过反射加载map函数，或者reduce函数；

9. 读取分片的数据块，或者输出计算结果数据。



2020主流用的是 MapReduce 2， 核心架构是变化不大的。



适合 MapReduce 的计算模型

• TopK

• K-means

• Bayes

• SQL



不适合 MapReduce 的计算模型

• Fibonacci

InputFormat

1. 验证作业的输入的正确性；

2. 将输入文件切分成逻辑的 InputSplits，一个 InputSplit 将被分配给一个单独的 Mapper task；

3. 提供 RecordReader 的实现，这个 RecordReader 会从 InputSplit 中正确读出一条一条的 K - V 提供给 Mapper 使用。

FileInputFormat

1. 得到分片的最小值 minSize 和最大值 maxSize，可以通过设置 mapred.min.split.size 和 mapred.max.split.size 来设置；

2. 对于每个输入文件，计算 max(minSize, min(maxSize, blockSize));

3. 如果 minSize <= blockSize <= maxSize, 则设为 blockSize.

4. 分片信息<file, start, length, hosts>, 通过 hosts 实现 map 本地性。



分片参数意义解析：

• file: 文件路径

• start：开始位置

• length：数据长度

• hosts： 记录InputSplit的分片信息在哪台机器上



OutputFormat

1. OutputFormat 接口决定了在哪里以及怎样持久化作业结果。

2. 默认的 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 内部实现

1. 作业控制

• 作业抽象成三层：作业监控层（Job In Progress JIP），任务控制层（Task In ProgressTIP），任务执行层。

• 任务可能会被尝试多次执行，每个任务实例被称作 Task Attempt（TA）。

• Task Attempt成功，TIP 会标注该任务成功，所有TIP成功，JIP 成功。



1. 资源管理

• 根据 TaskTracker 状态信息进行任务分配。



JobTracker 容错

• JobTracker 失败，那么未完成 Job 失败；

• 通过 Job 日志，Job 可部分恢复。

TaskTracker 容错

1. 超时

• TaskTracker 10 分钟（mapred.tasktracker.expiry.interval）未汇报心跳，则将其从集群移除。



1. 灰名单，黑名单

• TaskTracker 上部署性能监控脚本

• 如果性能表现太差，被 JobTracker 暂停调度



Task 容错

1. 允许部分 Task 失败

• 允许失败的任务占比，默认 0，mapred.max.map.failers.percent, mapred.max.reduce.failers.percent.



1. 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为例）

1. 我们向 Yarn 提交应用程序，包括 MapReduce ApplicationMaster、我们的MapReduce 程序，以及 MapReduce Application 启动命令。

2. ResourceManger 进程和 NodeManger 进程通信，根据集群资源，为用户程序分配第一个容器，并将 MapReduce ApplicationMaster 分发到这个容器上面，并在容器里面启动 MapReduce ApplicationMaster。

3. MapReduce ApplicationMaster 启动后立即向 ResourceManager 进程注册，并为自己的应用程序申请容器资源。

4. MapReduce ApplicationMaster 申请到需要的容器后，立即和相应的 NodeManger 进程通信，将用户 MapReduce 程序分发到 NodeManger 进程所在的服务器，并在容器中运行，运行的就是 Map 或者 Reduce 任务。

5. 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，Complier 转换为MapReduce能够处理的数据结构。



Hive 执行流程

1. 操作费（Operator）是Hive的最小处理单元；

2. 每个操作符处理代表 HDFS 操作或者 MR 作业

3. 编译器把 Hive SQL 转换成一组操作符

Hive 编译器





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。



1. TableScanOperator 把* 转换为需要查询的条件；

2. FilterOperator 是做条件过滤；

3. 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';



Facebook

GroupByOperator: setCombineClass 在Map端输出的时候，把数据求和以后再传给Reduce，可以减少传输的数据量。



Example Query(multi-group-by)

from (select a.status, b.school, b.gender 
	from status_updates a join profiles b
		on (a.userid = b.userid and a.ds='2009-03-20')
	 ) subql
insert overwrite table gender_summary partition(ds='2009-03-20')
select subql.gender, count(1) group by subql.gender
insert 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_users
select 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操作。）