首页 > 其他分享 >Hadoop-MapReduce的 原理 | 块和片 | Shuffle 过程 | Combiner

Hadoop-MapReduce的 原理 | 块和片 | Shuffle 过程 | Combiner

时间:2024-09-20 11:48:21浏览次数:12  
标签:map Combiner Shuffle 分区 reduce Hadoop MapTask 排序 数据

MapReduce的原理

简单版本:

AppMaster: 整个Job任务的核心协调工具
MapTask: 主要用于Map任务的执行
ReduceTask: 主要用于Reduce任务的执行

一个任务提交Job --> AppMaster(项目经理)--> 根据切片的数量统计出需要多少个MapTask任务 --> 向ResourceManager(Yarn平台的老大)索要资源 --> 执行Map任务,先读取一个分片的数据,传递给map方法。--> map 方法不断的溢写 --> reduce 方法 --> 将统计的结果存放在磁盘上。

分开讲解版:

MapTask执行阶段

1. maptask调用FileInputFormat的getRecordReader读取分片数据
2. 每行数据读取一次,返回一个(K,V)对,K是offset(偏移量),V是一行数据
3. 将k-v对交给MapTask处理
4. 每对k-v调用一次map(K,V,context)方法,然后context.write(k,v)
5. 写出的数据交给收集器OutputCollector.collector()处理
6. 将数据写入环形缓冲区,并记录写入的起始偏移量,终止偏移量,环形缓冲区默认大小100M
7. 默认写到80%的时候要溢写到磁盘,溢写磁盘的过程中数据继续写入剩余20%
8. 溢写磁盘之前要先进行分区然后分区内进行排序
9. 默认的分区规则是hashpatitioner,即key的  hash%reduceNum
      所有的mapreduce,其实都用到了分区,如果不写,使用的是默认的分区。
      job.setNumReduceTask(3);
10. 默认的排序规则是key的字典顺序,使用的是快速排序
11. 溢写会形成多个文件,在maptask读取完一个分片数据后,先将环形缓冲区数据刷写到磁盘
12. 将数据多个溢写文件进行合并,分区内排序(外部排序===》归并排序)

关于9 的再次解释:

ReduceTask的执行流程:

1. 数据按照分区规则发送到reducetask
2. reducetask将来自多个maptask的数据进行合并,排序(外部排序===》归并排序)
3. 按照key相同分组
4. 一组数据调用一次reduce(k,iterable<v>values,context)
5. 处理后的数据交由reducetask
6. reducetask调用FileOutputFormat组件
7. FileOutputFormat组件中的write方法将数据写出。

总结

ReduceTask任务的数量是由谁决定的?
job.setNumReduceTasks(5);
是指定的,设置的几个就执行几个。
这个值不能瞎设置,要参考分区数量,加入有三个分区,ReduceTask任务就需要指定为3个。

关于片和块

假如我现在500M这样的数据,如何存储?
500M = 128M + 128M + 128M + 116M  分为四个块进行存储。
计算的时候,是按照片儿计算的,而不是块儿。
块是物理概念,一个块就是128M ,妥妥的,毋庸置疑。
片是逻辑概念,一个片大约等于一个块。

假如我现在需要计算一个300M的文件,这个时候启动多少个MapTask任务?答案是有多少个片儿,就启动多少个任务。
一个片儿约等于 一个块,但是最大可以 128M*1.1倍= 140.8
300M 
128M  启动一个Map任务进行读取
172M  172M  和   128M * 1.1 =140.8M 进行比较,如果大于 ,继续进行切割
128M  启动一个任务Map任务
剩余44M   剩余的44M 和  128M*1.1倍比较,小于这个值,剩余的44M 就单独起一个Map任务
300m的数据,分给了3个MapTask任务进行处理。

如果是260M的数据,由多少个Map任务处理?
128M   第一个任务
132M   跟 128M * 1.1 进行比较,发现小于这个值,直接一个Map任务搞定,不在启动第三个任务了。

比如班里的同学一起搬砖,每人规定搬3块,假定砖还剩4块,到某个同学了,他就直接搬完即可,没必要让另一个同学因为一块砖,而专门跑一趟。

1、什么是片,什么是块?
块是物理概念,片是逻辑概念。一般片 = 块的,但是到最后一次的时候,有可能片> 块,但是绝对不能超过块的1.1倍。
2、mapreduce 启动多少个MapTask任务?
跟片有关系,有多少个片,就启动多少个map任务。跟块儿无关。

Shuffle 过程

MapReduce的Shuffle过程指的是MapTask的后半程,以及ReduceTask的前半程,共同组成的。
从MapTask中的map方法结束,到ReduceTask中的reduce方法开始,这个中间的部分就是Shuffle。是MapReduce的核心,心脏。

map端:

1、map中的context.write方法,对外写出的时候,其实是写入到了一个环形缓冲区内(内存形式的),这个环形缓冲区大小是100M,可以通过参数设置。如果里面的数据大于80M,就开始溢写(从内存中将数据写入到磁盘上)。溢写的文件存放地址可以设置。
2、在溢写过程中,环形缓冲区不会停止工作,是会利用剩余的20%继续存入环形缓冲区的。除非是环形缓冲区的内存满了,map任务就被阻塞了。
在溢写出来的文件中,是排过序的,排序规则:快速排序算法。在排序之前,会根据分区的算法,对数据进行分区。是在内存中,先分区,在每一个分区中再排序,接着溢写到磁盘上的。
3、溢写出来的小文件需要合并为一个大文件,因为每一个MapTask只能有一份数据。就将相同的分区文件合并,并且排序(此处是归并排序)。每次合并的时候是10个小文件合并为一个大文件,进行多次合并,最终每一个分区的文件只能有一份。
假如100个小文件,需要合并几次呢?
100  每10分合并一次,第一轮:100个文件合并为了10个文件,这10个文件又合并为一个大文件,总共合并了11次。

4、将内存中的数据,溢写到磁盘上,还可以指定是否需要压缩,以及压缩的算法是什么。

reduce端:

1、reduce端根据不同的分区,拉取每个服务器上的相同的分区的数据。
reduce任务有少量复制线程,因此能够并行取得map输出。默认值是5个线程,但这个默认值可以修改设置mapreduce.reduce.shuffle. parallelcopies 属性即可。
2、如果map上的数据非常的小,该数据会拉取到reduce端的内存中,如果数据量比较大,直接拉取到reduce端的硬盘上。

环形缓冲区【嘚瑟部分】:

环形缓冲区,其实是一个数组,将数组分为两部分,分割的这个点就称之为轴心。存储KV键值对,是从左到右,类似于顺时针,因为每一个KV键值对都有对应的元数据。元数据是从轴心开始,从右向左执行。

当两者数据占用空间达到80%的时候,需要清理数组,清理完之后,轴心发生了变化。

KV键值对的元数据,

(每四个是一组,共计4组)

前面四个第一组::表示Value的起始位置,第二组:Key值的起始位置,第三组:分区信息,第四组:val的长度。这些内容称之为KV键值对的meta数据(元数据)。

Combiner

这个Combiner是一个优化的代码,对于我们最终的结果没有任何的影响。

map端产生的数据,会被拉去到reduce端进行合并,有可能map端产生的数据非常的大,不便于在网络间传输,那么有没有办法可以缩小map端的数据呢?

之前: java 1 java 1 java 1 传递给reduce

现在: java 3 传递给reduce

Combiner其实就是运行在mapTask中的reducer。 Reducer其实就是合并代码的。Combiner是作用在Map端的。

这个结果不是最终的结果,而是一个临时的小统计。 最终reduce是会将所有的map结果再次进行汇总才是我们最终想要的统计结果。

Combiner 只能用于对统计结果没有影响的场景下。
一般只用于  统计之和,统计最大值最小值的场景下。统计平均值等情况是不能用的。

在代码中如何使用?

Combiner起作用的地方:

Combiner 其实作用于两个地方,一个是环形缓冲区溢写磁盘的时候,除了分区,排序之外,还可以做合并操作,将内存中的 hello 1 hello 1 hello 1 会合并为 hello 3

第二个位置是小文件合并为MapTask的大文件的时候,会将多个 hello 的值相加 hello 19,但是这个不是最终的答案,最终答案是将多个MapTask任务中的hello 进行合并才是最终的结果。

标签:map,Combiner,Shuffle,分区,reduce,Hadoop,MapTask,排序,数据
From: https://blog.csdn.net/Yz9876/article/details/142381094

相关文章

  • centos789手动无脑用sh脚本安装hadoop3.3.5
    和上篇一样的操作查看代码#!/bin/bash#确保以root用户运行if["$(id-u)"-ne"0"];thenecho"请以root用户运行此脚本!"exit1fi#定义目录和文件路径SOFTWARE_DIR="/export/software"SERVER_DIR="/export/servers"HADOOP_TAR="......
  • Hadoop Windows下安装
    1.下载winutils:https://github.com/s911415/apache-hadoop-3.1.0-winutils,替换bin文件hadoop:3.2.0tar解压2.环境变量hadoopversion验证问题:Error:java_homeisincorrectlyset解决:由于jdk在C盘引起:hadoop-env.cmd加入setJAVA_HOME=C:\PROGRA~1\Java\jdk1.8.0_112......
  • Hadoop(二十二)CM Kerberos基本原理
    基于CDP的CM维护Hadoop生态组件CDPClouderaDataPlatform(CDP)是Cloudera公司推出的一个全面的数据平台,它支持在多云环境中部署和管理数据CDP提供了一种统一的方法来管理数据和分析工作负载,无论是在本地、公有云还是私有云环境中CDP包括多种服务和工具,如ClouderaDataHub(CDH)......
  • Hadoop(二十)Yarn工作原理
    Yarn资源调度器Yarn是一个资源调度平台,负责为运算程序提供服务器运算资源,相当于一个分布式的操作系统平台,而MapReduce等运算程序则相当于运行于操作系统之上的应用程序一、基础架构YARN主要由ResourceManager、NodeManager、ApplicationMaster和Container等组件构成二、Y......
  • Hadoop(十九)MapReduce OutputFormat 数据压缩
    OutputFormatOutputFormat是MapReduce输出的基类,所有实现MapReduce输出都实现了OutputFormat接口几种常见的OutputFormat实现类:NullOutputFormat、MapFileOutputFormat、TextOutputFormat等自定义OutputFormat应用场景:输出数据到MySQL/HBase/Elasticsearch等存储框架中步......
  • Hadoop(十八)MapReduce Shuffle机制
    MapReduce工作流程上面的流程是整个MapReduce最全工作流程,但是Shuffle过程只是从第7步开始到第16步结束,具体Shuffle过程详解,如下:MapTask收集map()方法输出的kv对,放到内存缓冲区中从内存缓冲区不断溢出本地磁盘文件,可能会溢出多个文件多个溢出文件会被合并成大的溢出文件在......
  • Hadoop(十七)MapReduce 切片机制 InputFormat
    切片与MapTask并行度决定机制MapTask的并行度决定Map阶段的任务处理并发度,进而影响到整个Job的处理速度数据块:Block是HDFS物理上把数据分成一块一块,数据块是HDFS存储数据单位数据切片:数据切片只是在逻辑上对输入进行分片,并不会在磁盘上将其切分成片进行存储,数据切片是MapReduc......
  • Hadoop
    Hadoop目录Hadoop一、简介Hadoop生态Hadoop组成hdfsyarnmapreduce:HDFS、YARN、MapReduce三者关系:常用端口号:常用的配置文件:部署二、HDFS详解HDFS概述HDFS成员HDFS读写流程HDFS常用命令HDFS编程三、MapReduce详解MapReduce概述MapReduce架构MapTask详解ReduceTask详解shuffle详解M......
  • Hadoop(十五)项目考核 WordCount案例
    一、需求分析需求:在给定的文本文件中统计输出每一个单词出现的总次数SEVENTEEN.txt文本内容如下:saythenameseventeenhelloweareseventeennicetomeetyouyouverynice按照MapReduce编程规范,分别编写Mapper,Reducer,Driver1、Mapper(1)将MapTask传过来的文本内容......
  • Hadoop
    1.概念分布式系统基础架构。主要包括分布式文件系统HDFS(HadoopDistributedFileSystem)​、分布式计算系统MapReduce和分布式资源管理系统YARN2.构成2.1HDFSHDFS提供了高可靠性(主要通过多副本来实现)​、高扩展性(通过添加机器来达到线性扩展)和高吞吐率的数据存储服务HDFS的......