Hadoop(十四)——hadoop之MapReduce理论篇(五)——MapReduce详细工作流程

Stella981
• 阅读 716

一、Shuffle机制

Mapreduce确保每个reducer的输入都是按键排序的。系统执行排序的过程(即将map输出作为输入传给reducer)称为shuffle。

二、MapReduce工作流程

1.图示流程

Hadoop(十四)——hadoop之MapReduce理论篇(五)——MapReduce详细工作流程

2.流程详解

上面的流程是整个mapreduce最全工作流程,但是shuffle过程只是从第7步开始到第16步结束,具体shuffle过程详解,如下:

  • 1)maptask收集我们的map()方法输出的kv对,放到内存缓冲区中
  • 2)从内存缓冲区不断溢出本地磁盘文件,可能会溢出多个文件
  • 3)多个溢出文件会被合并成大的溢出文件
  • 4)在溢出过程中,及合并的过程中,都要调用partitoner进行分组和针对key进行排序
  • 5)reducetask根据自己的分区号,去各个maptask机器上取相应的结果分区数据
  • 6)reducetask会取到同一个分区的来自不同maptask的结果文件,reducetask会将这些文件再进行合并(归并排序)
  • 7)合并成大文件后,shuffle的过程也就结束了,后面进入reducetask的逻辑运算过程(从文件中取出一个一个的键值对group,调用用户自定义的reduce()方法)

3.注意

Shuffle中的缓冲区大小会影响到mapreduce程序的执行效率,原则上说,缓冲区越大,磁盘io的次数越少,执行速度就越快。 缓冲区的大小可以通过参数调整,参数:io.sort.mb 默认100M

三、partition分区

参考:大数据案例(三)——MapReduce实现流量统计案例-分区

注意:

  • 如果reduceTask的数量> getPartition的结果数,则会多产生几个空的输出文件part-r-000xx;
  • 如果1<reduceTask的数量<getPartition的结果数,则有一部分分区数据无处安放,会Exception;
  • 如果reduceTask的数量=1,则不管mapTask端输出多少个分区文件,最终结果都交给这一个reduceTask,最终也就只会产生一个结果文件 part-r-00000;
    • 例如:假设自定义分区数为5,则
    • (1)job.setNumReduceTasks(1);会正常运行,只不过会产生一个输出文件
    • (2)job.setNumReduceTasks(2);会报错
    • (3)job.setNumReduceTasks(6);大于5,程序会正常运行,会产生空文件

四、排序

1.排序的分类:

  • (1)部分排序:MapReduce根据输入记录的键对数据集排序。保证输出的每个文件内部排序。
  • (2)全排序:如何用Hadoop产生一个全局排序的文件?最简单的方法是使用一个分区。但该方法在处理大型文件时效率极低,因为一台机器必须处理所有输出文件,从而完全丧失了MapReduce所提供的并行架构。
    • 替代方案:首先创建一系列排好序的文件;其次,串联这些文件;最后,生成一个全局排序的文件。主要思路是使用一个分区来描述输出的全局排序。例如:可以为上述文件创建3个分区,在第一分区中,记录的单词首字母a-g,第二分区记录单词首字母h-n, 第三分区记录单词首字母o-z。
  • (3)辅助排序:(GroupingComparator分组)
    • Mapreduce框架在记录到达reducer之前按键对记录排序,但键所对应的值并没有被排序。甚至在不同的执行轮次中,这些值的排序也不固定,因为它们来自不同的map任务且这些map任务在不同轮次中完成时间各不相同。一般来说,大多数MapReduce程序会避免让reduce函数依赖于值的排序。但是,有时也需要通过特定的方法对键进行排序和分组等以实现对值的排序。

2.自定义排序WritableComparable

原理分析:bean对象实现WritableComparable接口重写compareTo方法,就可以实现排序

五、GroupingComparator分组

六、 Combiner合并

  1. combiner是MR程序中Mapper和Reducer之外的一种组件
  2. combiner组件的父类就是Reducer
  3. combiner和reducer的区别在于运行的位置:
    • Combiner是在每一个maptask所在的节点运行
    • Reducer是接收全局所有Mapper的输出结果;
  4. combiner的意义就是对每一个maptask的输出进行局部汇总,以减小网络传输量
  5. 自定义Combiner实现步骤:
  • (1)自定义一个combiner继承Reducer,重写reduce方法

    public class WordcountCombiner extends Reducer<Text, IntWritable, Text, IntWritable>{ @Override protected void reduce(Text key, Iterable values, Context context) throws IOException, InterruptedException { int count = 0; for(IntWritable v :values){ count = v.get(); } context.write(key, new IntWritable(count)); } }

  • (2)在job中设置:

    job.setCombinerClass(WordcountCombiner.class);

  1. combiner能够应用的前提是不能影响最终的业务逻辑,而且,combiner的输出kv应该跟reducer的输入kv类型要对应起来

七、数据倾斜&Distributedcache

  1. 数据倾斜原因:如果是多张表的操作都是在reduce阶段完成,reduce端的处理压力太大,map节点的运算负载则很低,资源利用率不高,且在reduce阶段极易产生数据倾斜。
  2. 实际案例: 大数据案例(六)——MapReduce之reduce端表合并(数据倾斜)
  3. 解决方案: 在map端缓存多张表,提前处理业务逻辑,这样增加map端业务,减少reduce端数据的压力,尽可能的减少数据倾斜。
  4. 具体办法:采用distributedcache
    • 在mapper的setup阶段,将文件读取到缓存集合中
    • 在驱动函数中加载缓存。
      • job.addCacheFile(new URI("file:/j:/pd.txt"));// 缓存普通文件到task运行节点
  5. 实际案例:大数据案例(七)——MapReduce之map端表合并(Distributedcache)
点赞
收藏
评论区
推荐文章
blmius blmius
3年前
MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1
文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s
皕杰报表之UUID
​在我们用皕杰报表工具设计填报报表时,如何在新增行里自动增加id呢?能新增整数排序id吗?目前可以在新增行里自动增加id,但只能用uuid函数增加UUID编码,不能新增整数排序id。uuid函数说明:获取一个UUID,可以在填报表中用来创建数据ID语法:uuid()或uuid(sep)参数说明:sep布尔值,生成的uuid中是否包含分隔符'',缺省为
待兔 待兔
4个月前
手写Java HashMap源码
HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22
Stella981 Stella981
3年前
MapReduce编程模型和计算框架架构原理
Hadoop解决大规模数据分布式计算的方案是MapReduce。MapReduce既是一个编程模型,又是一个计算框架。也就是说,开发人员必须基于MapReduce编程模型进行编程开发,然后将程序通过MapReduce计算框架分发到Hadoop集群中运行。我们先看一下作为编程模型的MapReduce。MapReduce编程模型
Stella981 Stella981
3年前
Hadoop MapReduce执行过程详解(带hadoop例子)
分析MapReduce执行过程MapReduce运行的时候,会通过Mapper运行的任务读取HDFS中的数据文件,然后调用自己的方法,处理数据,最后输出。Reducer任务会接收Mapper任务输出的数据,作为自己的输入数据,调用自己的方法,最后输出到HDFS的文件中。整个流程如图:!image
Stella981 Stella981
3年前
Hadoop学习之路(二十三)MapReduce中的shuffle详解
概述1、MapReduce中,mapper阶段处理的数据如何传递给reducer阶段,是MapReduce框架中最关键的一个流程,这个流程就叫Shuffle2、Shuffle:数据混洗——(核心机制:数据分区,排序,局部聚合,缓存,拉取,再合并排序)3、具体来说:就是将MapTask输出的处理结果数据,按照Par
Stella981 Stella981
3年前
Hadoop源代码分析(包hadoop.mapred中的MapReduce接口)
前面已经完成了对org.apache.hadoop.mapreduce的分析,这个包提供了HadoopMapReduce部分的应用API,用于用户实现自己的MapReduce应用。但这些接口是给未来的MapReduce应用的,目前MapReduce框架还是使用老系统(参考补丁HADOOP1230(https://www.oschina.net/act
Stella981 Stella981
3年前
Hadoop之Mapreduce详解
1、什么是Mapreduce   Mapreduce是一个分布式运算程序的编程框架,是用户开发“基于hadoop的数据分析应用”的核心框架;Mapreduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序,并发运行在一个hadoop集群上;2、Mapreduce框架结构及核心运行机制
Wesley13 Wesley13
3年前
MySQL部分从库上面因为大量的临时表tmp_table造成慢查询
背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_
Python进阶者 Python进阶者
10个月前
Excel中这日期老是出来00:00:00,怎么用Pandas把这个去除
大家好,我是皮皮。一、前言前几天在Python白银交流群【上海新年人】问了一个Pandas数据筛选的问题。问题如下:这日期老是出来00:00:00,怎么把这个去除。二、实现过程后来【论草莓如何成为冻干莓】给了一个思路和代码如下:pd.toexcel之前把这