MapReduce原理

news2024/11/27 19:48:50

MapReduce 编程规范

MapReduce 的开发一共有八个步骤, 其中 Map 阶段分为 2 个步骤，Shuffle 阶段 4 个步骤，Reduce 阶段分为 2 个步骤

Map 阶段 2 个步骤

  1. 设置 InputFormat 类, 将数据切分为 Key-Value(K1和V1) 对, 输入到第二步
  2. 自定义 Map 逻辑, 将第一步的结果转换成另外的 Key-Value（K2和V2） 对, 输出结果

Shuffle 阶段 4 个步骤

  1. 对输出的 Key-Value 对进行分区
  2. 对不同分区的数据按照相同的 Key 排序
  3. (可选) 对分组过的数据初步规约, 降低数据的网络拷贝
  4. 对数据进行分组, 相同 Key 的 Value 放入一个集合中

Reduce 阶段 2 个步骤

  1. 对多个 Map 任务的结果进行排序以及合并, 编写 Reduce 函数实现自己的逻辑, 对输入的 Key-Value 进行处理, 转为新的 Key-Value（K3和V3）输出
  2. 设置 OutputFormat 处理并保存 Reduce 输出的 Key-Value 数据

MapReduce 运行模式

本地运行模式

  1. MapReduce 程序是被提交给 LocalJobRunner 在本地以单进程的形式运行
  2. 处理的数据及输出结果可以在本地文件系统, 也可以在hdfs上
  3. 怎样实现本地运行? 写一个程序, 不要带集群的配置文件, 本质是程序的 conf 中是否有
mapreduce.framework.name=local 以及 
yarn.resourcemanager.hostname=local 参数
  1. 本地模式非常便于进行业务逻辑的 Debug , 只要在 Eclipse 中打断点即可
 1 configuration.set("mapreduce.framework.name","local"); 
 2 configuration.set("yarn.resourcemanager.hostname","local"); 
 3 TextInputFormat.addInputPath(job,new Path("file:///F:\\wordcount\\input"));
 4 TextOutputFormat.setOutputPath(job,new Path("file:///F:\\wordcount\\output"));

集群运行模式

  1. 将 MapReduce 程序提交给 Yarn 集群, 分发到很多的节点上并发执行
  2. 处理的数据和输出结果应该位于 HDFS 文件系统
  3. 提交集群的实现步骤: 将程序打成JAR包，然后在集群的任意一个节点上用hadoop命令启动
hadoop jar hadoop_hdfs_operate-1.0-SNAPSHOT.jar cn.itcast.hdfs.demo1.JobMain

MapReduce 分区

在 MapReduce 中, 通过我们指定分区, 会将同一个分区的数据发送到同一个 Reduce 当中进行处理
例如: 为了数据的统计, 可以把一批类似的数据发送到同一个 Reduce 当中, 在同一个 Reduce 当中统计相同类型的数据, 就可以实现类似的数据分区和统计等其实就是相同类型的数据, 有共性的数据, 送到一起去处理
Reduce 当中默认的分区只有一个

MapReduce 排序和序列化

  • 序列化 (Serialization) 是指把结构化对象转化为字节流
  • 反序列化 (Deserialization) 是序列化的逆过程. 把字节流转为结构化对象. 当要在进程间传递对象或持久化对象的时候, 就需要序列化对象成字节流, 反之当要将接收到或从磁盘读取的字节流转换为对象, 就要进行反序列化
  • Writable 是 Hadoop 的序列化格式, Hadoop 定义了这样一个 Writable 接口. 一个类要支持可序列化只需实现这个接口即可
  • 另外 Writable 有一个子接口是 WritableComparable, WritableComparable 是既可实现序列化, 也可以对key进行比较

MapReduce的运行机制详解

在这里插入图片描述

MapTask 工作机制

  1. 读取数据组件InputFormat(默认TextInputFormat)会通过getSplits方法对输入目录中文件进行逻辑切片规划得到block,有多少个block就对应启动多少个MapTask。
  2. 将输入文件切分为block之后,由RecordReader对象(默认是LineRecordReader)进行读取,以\n作为分隔符,读取一行数据,返回<key，value>。Key表示每行首字符偏移值,Value表示这一行文本内容
  3. 读取block返回<key,value>,进入用户自己继承的Mapper类中，执行用户重写的map函数,RecordReader每读取一行block中的内容，这里调用一次map函数
  4. Mapper逻辑结束之后,将Mapper的每条结果通过context.write进行collect数据收集。在collect中,会先对其进行分区处理，来决定当前的这对输出数据最终应该交由哪个Reduce task处理(默认使用HashPartitioner对Key Hash后再以Reducer数量取模)
  5. 接下来,会将数据写入内存,内存中这片区域叫做环形缓冲区,缓冲区的作用是批量收集
    Mapper结果,减少磁盘IO的影响。我们的Key/Value对以及Partition的结果都会被写入缓冲区。当然,写入之前，Key与Value值都会被序列化成字节数组。
  6. 当溢写线程启动后,需要对这80MB空间内的Key做排序(Sort)。排序是MapReduce模型默认的行为,这里的排序也是对序列化的字节做的排序
  7. 合并溢写文件,每次溢写会在磁盘上生成一个临时文件(写之前判断是否有Combiner),如果Mapper的输出结果真的很大,有多次这样的溢写发生,磁盘上相应的就会有多个临时文件存在。当整个数据处理结束之后开始对磁盘中的临时文件进行Merge合并,因为最终的文件只有一个,写入磁盘,并且为这个文件提供了一个索引文件,以记录每个reduce对应数据的偏移量
环形缓冲区其实是一个数组,数组中存放着Key,Value的序列化数据和Key,Value的元数据信息,包括Partition,Key的起始位置,Value的起始位置以及Value的长度。环形结构是一个抽象概念。
缓冲区是有大小限制,默认是100MB。当Mapper的输出结果很多时,就可能会撑爆内存,所以需要在一定条件下将缓冲区中的数据临时写入磁盘,然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为Spill,中文可译为溢写。这个溢写是由单独线程来完成,不影响往缓冲区写Mapper结果的线程。溢写线程启动时不应该阻止Mapper的结果输出,所以整个缓冲区有个溢写的比例spill。percent。这个比例默认是0。8,也就是当缓冲区的数据已经达到阈值 buffersize*spillpercent=100MB*0.8=80MB,溢写线程启动,锁定这80MB的内存,执行溢写过程。Mapper的输出结果还可以往剩下的20MB内存中写,互不影响

如果Job设置过Combiner,那么现在就是使用Combiner的时候了。将有相同Key的Key/Value对的Value加起来,减少溢写到磁盘的数据量。Combiner会优化MapReduce的中间结果,所以它在整个模型中会多次使用
	
那哪些场景才能使用Combiner呢?从这里分析,Combiner的输出是Reducer的
输入,Combiner绝不能改变最终的计算结果。Combiner只应该用于那种Reduce的输入Key/Value与输出Key/Value类型完全一致,且不影响最终结果的场景。比如累加,最大值等。Combiner的使用一定得慎重,如果用好,它对Job执行效率有帮助,反之会影响Reducer的最终结果

ReduceTask 工作机制

  1. Copy阶段：ReduceTask从各个MapTask上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定阈值，则写到磁盘上，否则直接放到内存中。
  2. Sort阶段：在远程拷贝数据的同时，ReduceTask启动了两个后台线程对内存和磁盘上的文件进行合并，以防止内存使用过多或磁盘上文件过多。按照MapReduce语义，用户编写reduce()函数输入数据是按key进行聚集的一组数据。为了将key相同的数据聚在一起，Hadoop采用了基于排序的策略。由于各个MapTask已经实现对自己的处理结果进行了局部排序，因此，ReduceTask只需对所有数据进行一次归并排序即可。
  3. Reduce阶段：reduce()函数将计算结果写到HDFS上。

Shuffle阶段

map 阶段处理的数据如何传递给 reduce 阶段，是 MapReduce 框架中最关键的一个流程，这个流程就叫 shuffle—（核心机制：数据分区，排序，分组，规约，合并等过程）
shuffle是Mapreduce 的核心，它分布在 Mapreduce的map阶段和reduce 阶段。一般把从Map 产生输出开始到 Reduce 取得数据作为输入之前的过程称作 shuffle。

  1. Collect阶段：将 MapTask 的结果输出到默认大小为 100M 的环形缓冲区，保存的是 key/value，Partition 分区信息等。
  2. Spill阶段：当内存中的数据量达到一定的阀值的时候，就会将数据写入本地磁盘，在将数据写入磁盘之前需要对数据进行一次排序的操作，如果配置了 combiner，还会将有相同分区号和 key 的数据进行排序。
  3. Merge阶段：把所有溢出的临时文件进行一次合并操作，以确保一个 MapTask 最终只产生一个中间数据文件。
  4. Copy阶段：ReduceTask 启动 Fetcher 线程到已经完成 MapTask 的节点上复制一份属于自己的数据，这些数据默认会保存在内存的缓冲区中，当内存的缓冲区达到一定的阀值的时候，就会将数据写到磁盘之上。
  5. Merge阶段：在 ReduceTask 远程复制数据的同时，会在后台开启两个线程对内存到本地的数据文件进行合并操作。
  6. Sort阶段：在对数据进行合并的同时，会进行排序操作，由于 MapTask 阶段已经对数据进行了局部的排序，ReduceTask 只需保证 Copy 的数据的最终整体有效性即可。
    Shuffle中的缓冲区大小会影响到 mapreduce 程序的执行效率，原则上说，缓冲区越大，磁盘io的次数越少，执行速度就越快
    缓冲区的大小可以通过参数调整, 参数：mapreduce.task.io.sort.mb 默认100M

HDFS&Map-Reduce&YARN工作提交全流程

请添加图片描述

HDFSYARNMapReduce
DataNode
NameNode		ResourceManager
NodeManager
ApplicationMaster
Container		MrAppMaster：负责整个程序的过程调度及状态协调。
MapTask：负责Map阶段的整个数据处理流程。
ReduceTask：负责Reduce阶段的整个数据处理流程。
  1. 在客户端执行submit()方法之前，会先去获取一下待读取文件的信息，然后行程一个任务划分的规划
  2. YarnRunner向ResourceManager申请一个Application。RM将该应用程序的资源路径返回给YarnRunner。
  3. 将job提交给yarn，这时候会带着三个信息过去（job.split；.jar;job.xml）程序资源提交完毕后，申请运行mrAppMaster。
  4. yarn 会根据文件的切片信息去计算将要启动的maptask的数量，然后去启动maptask;
  5. maptask会调用inputFormat()方法去HDFS上面读取文件。inputFormat()方法会再去调用RecordRead()方法，将数据以行首字母的偏移量为key，一行数据为value传给mapper()方法；
  6. mapper方法做一下逻辑处理后，将数据传到分区方法中，对数据进行一个分区标注后，发送到环形缓冲区中；
  7. 环形缓冲区默认的大小是100M,达到80%的阈值将会溢写；
  8. 在溢写之前会做一个排序的动作，排序的规则是按照key进行字典序排序，排序的手段是快排；
  9. 在溢写会产生出大量的溢写文件，会再次调用marge()方法，使用归并排序默认10个溢写文件合并一个大文件；
  10. 也可以对溢写文件做一个localReduce也就是combiner的操作，但前提是combiner的结果不能对最终的结果有影响；
  11. 等待所有的maptask结束后，会启动一定数量的reducetask；
  12. reducetask会发起拉取线程到map端拉取数据，拉取到数据会先加载到内存中，内存不够会写到磁盘里，等待所有数据拉取完毕，会将这些输出再进行一次归并排序；
  13. 归并后的文件会在进行一次分组的操作，然后再将数据以组为单位发送到reduce方法（）；
  14. reduce方法做一下逻辑判断后，最终调用OutputFormat（）会再调用RecordWrite()方法将数据以KV的信息写到HDFS上；

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/428827.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

十五周算法训练营——链表专题

十五周算法训练营——链表专题

今天是十五周算法训练营的第三周&#xff0c;主要讲链表专题&#xff0c;包含&#xff1a;反转链表、移除链表、交换链表、链表相交、删除链表中的倒数第N个节点、环形链表II。&#xff08;欢迎加入十五周算法训练营&#xff0c;与小伙伴一起卷算法——文章末尾进群&#xff09…
阅读更多...
【我的创作纪念日】恒川的创作一周年

【我的创作纪念日】恒川的创作一周年

机缘 大家好&#xff0c;我是热爱跑步的恒川&#xff0c;今天是个特殊的日子&#xff08;我的创作纪念日&#xff09;&#xff0c;在去年的今天&#xff0c;我发了第一篇博文。去年的时候&#xff0c;刚接触到CSDN&#xff0c;只想把他当作一个学习的工具&#xff0c;后来&…
阅读更多...
Node.js -- Express路由

Node.js -- Express路由

1.路由的概念 在Express中&#xff0c;路由指的是客户端的请求与服务器处理函数之间的关系。 Express中的路由分为三部分&#xff0c;分别是请求的类型&#xff0c;请求的URL地址和处理函数。格式如下&#xff1a; app.METHOD(PATH,HANDLER)例子如下&#xff1a; // 匹配 GE…
阅读更多...
【计算几何3】博物馆问题和相关证明

【计算几何3】博物馆问题和相关证明

&#xff08; gallery problem or museum problem &#xff09; Art gallery problem - Wikipedia 一、说明 画廊问题是一个基本的计算几何问题&#xff0c;最现实的反映是&#xff0c;到底对一个场景监视会用到几个摄像头。更深刻的层次是&#xff0c;如何对场景生成地图&am…
阅读更多...
C生万物 | 十分钟带你学会位段相关知识

C生万物 | 十分钟带你学会位段相关知识

结构体相关知识可以先看看这篇文章 —— 链接 一、什么是位段 位段的声明和结构是类似的&#xff0c;有两个不同&#xff1a; 位段的成员必须是 int、unsigned int 或signed int位段的成员名后边有一个冒号和一个数字 在下面&#xff0c;我分别写了一个结构体和一个位段&…
阅读更多...
嵌入式51单片机01-开发版介绍与LED流水灯系列

嵌入式51单片机01-开发版介绍与LED流水灯系列

文章目录1. STC89C52单片机介绍2. 单片机预备知识3. LED灯操作系列1. 点亮LED&#xff08;1&#xff09;LED原理图连接方式&#xff08;2&#xff09; proteus接线图&#xff08;3&#xff09;keil代码2. LED灯闪烁&#xff08;1&#xff09;实验电路图&#xff08;2&#xff0…
阅读更多...
每日一题104——转置矩阵

每日一题104——转置矩阵

给你一个二维整数数组 matrix&#xff0c; 返回 matrix 的 转置矩阵 。 矩阵的 转置 是指将矩阵的主对角线翻转&#xff0c;交换矩阵的行索引与列索引。 示例 1&#xff1a; 输入&#xff1a;matrix [[1,2,3],[4,5,6],[7,8,9]] 输出&#xff1a;[[1,4,7],[2,5,8],[3,6,9]] 示…
阅读更多...
大佬们隔空互怼

大佬们隔空互怼

阅读本文大概需要 1.14 分钟。1、最近国内各大互联网公司都在掀起一股大模型热潮&#xff0c;要想让这股热潮变得更持久更精彩&#xff0c;离不开互联网大佬们的添砖加瓦。比如上周搜狗的王小川跟百度的肖阳&#xff0c;两位互联网大佬隔空互怼的场景&#xff0c;一度让大家觉得…
阅读更多...
时序分析 49 -- 贝叶斯时序预测（一）

时序分析 49 -- 贝叶斯时序预测（一）

贝叶斯时序预测&#xff08;一&#xff09; 时序预测在统计分析和机器学习领域一直都是一个比较重要的话题。在本系列前面的文章中我们介绍了诸如ARIMA系列方法&#xff0c;Holt-Winter指数平滑模型等多种常用方法&#xff0c;实际上这些看似不同的模型和方法之间都具有千丝万缕…
阅读更多...
SpringBoot中处理日期的两种方式(消息转换器)

SpringBoot中处理日期的两种方式(消息转换器)

在Spring Boot中,我们通常会使用Jackson来序列化和反序列化Java对象到JSON。在进行日期序列化时,我们需要指定日期格式,否则Jackson会使用默认格式,这可能不是我们想要的。要指定日期格式,有两种方式: 如果没有处理按照默认的结果如下图: 所以我们需要通过一些手段来对日期进行…
阅读更多...
总结一下Redis的缓存雪崩、缓存击穿、缓存穿透

总结一下Redis的缓存雪崩、缓存击穿、缓存穿透

缓存是提高系统性能的一种常见手段&#xff0c;其中Redis是一种常用的高性能缓存数据库。但是在使用缓存时&#xff0c;可能会遇到一些问题&#xff0c;比如缓存击穿、缓存穿透、缓存雪崩等问题&#xff0c;本文将介绍这些问题的概念、原因以及解决方案。 缓存击穿 缓存击穿指…
阅读更多...
Jenkins——用户管理、授权策略配置以及Jenkins安全配置管理

Jenkins——用户管理、授权策略配置以及Jenkins安全配置管理

这里写目录标题一、Jenkins用户管理1、进入系统管理界面2、创建用户3、编辑用户信息4、编辑用户信息5、删除用户信息二、Jenkins授权策略配置1、授权策略插件2、安装插件:Role-based Authorization Strategy3、管理角色a、全局角色 Global rolesb、项目角色 Item rolesc、节点角…
阅读更多...
CVPR | 达摩院开源自监督学习框架CoKe, 单机8卡可训练

CVPR | 达摩院开源自监督学习框架CoKe, 单机8卡可训练

团队模型、论文、博文、直播合集&#xff0c;点击此处浏览 一、论文 论文链接: Unsupervised Visual Representation Learning by Online Constrained K-Means 代码链接&#xff1a;https://github.com/idstcv/CoKe 二、背景 虽然基于instance的自监督学习方法在ImageNet上…
阅读更多...
注册claude AI账号 slack工作区账号

注册claude AI账号 slack工作区账号

Claude 是建立在 slack工作区的一个AI人工助手&#xff0c;更像是将chatgpt集成到了会议模式&#xff0c;一个账号实际上拥有了你的会议室和你的AI助手&#xff0c;你可以让你的朋友和同事进入你的房间体验。 Claude是不是openai的产物&#xff1f;目前还不知道&#xff0c;不…
阅读更多...
RS232/RS485/RS422 接线说明

RS232/RS485/RS422 接线说明

RS232/RS485/RS422 接线 文章目录RS232/RS485/RS422 接线RS232RS485422RS232 设备A 和 设备B 没有硬件流控 设备A设备BTXDRXDRXDTXDGNDGND 设备A 和 设备B 使用RTS/CTS 做硬件流控 一般使用RTS/CTS都会由232芯片自动控制&#xff0c; 比如XR21V1414, 需要设备gpio mode 为00…
阅读更多...
Docker安装+利用docker安装MySQL（保姆级教程）

Docker安装+利用docker安装MySQL（保姆级教程）

前言&#xff1a;本文基于docker安装MySQL&#xff0c;如果还没安装docker的童鞋&#xff0c;点击这里&#xff08;超大声&#xff09;安装卸载老版本docker&#xff08;超详细&#xff09; 一、在docker仓库搜你想要的镜像版本 docker镜像仓库 打开官网&#xff0c;在搜索框…
阅读更多...
Docker 镜像制作 服务编排 私有仓库

Docker 镜像制作 服务编排 私有仓库

DockerNginx部署Redis部署Dockerfile镜像制作容器转为镜像dockerfileDocker服务编排Docker ComposeDocker Compose安装使用docker compose编排nginxspringboot项目Docker私有仓库私有仓库搭建将镜像上传至私有仓库Nginx部署 案例&#xff1a;需求 在Docker容器中部署Nginx&…
阅读更多...
【ARMv8 异常模型入门及渐进2 - 系统寄存器访问方法：op1，CRn，CRm，op2】

【ARMv8 异常模型入门及渐进2 - 系统寄存器访问方法：op1，CRn，CRm，op2】

文章目录1.1 ARMv8 系统寄存器访问概要1.1.1 系统寄存器访问级别1.1.2 ARMv8 系统寄存器编码1.1 ARMv8 系统寄存器访问概要 ARMv8中&#xff0c;取消了协处理器&#xff0c;之前协处理器实现的功能&#xff0c;全部由系统寄存器来是实现。对于系统寄存器的访问&#xff0c;使用…
阅读更多...
javaweb在校大学生贷款管理系统ns08a9

javaweb在校大学生贷款管理系统ns08a9

1系统主要实现&#xff1a;学生注册、填写详细资料、申请贷款、学校审核、银行审核、贷后管理等功能&#xff0c; (1) 学生注册&#xff1a;学生通过注册用户&#xff0c;提交自己的详细个人资料&#xff0c;考虑现实应用中的安全性&#xff0c;资料提交后不可修改&#xff1b;…
阅读更多...
Springboot + MySQL+ JPA Ⅲ delete方法详解

Springboot + MySQL+ JPA Ⅲ delete方法详解

一、deleteById(Id id) 和 delete(T entity) 为什么要把这两个方法放在一起呢&#xff1f;我们先看源码再说 deleteById源码&#xff08;通过id进行删除&#xff09; Transactional Override public void deleteById(ID id) {Assert.notNull(id, ID_MUST_NOT_BE_NULL);delet…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023