`

Hadoop MapReduce常见问题学习(shuffle-->将map输出转化成reducer输入)

 
阅读更多
MapReduce是什么
Hadoop是一个分布式计算系统,两大核心HDFS,这是存储文件的,MapReduce这是用于计算的。MapReduce是一种编程模型,类比为设计模式,好比最佳实践,编写是先写map函数,再写reduce函数。

Shuffle是什么



这个图来自《Hadoop,The Definitive Guide》,就是“谁说大象不能跳舞”那本书,用来描述shuffle和sort,用这个图理解shuffle,很难学的会跳舞。
shuffle描述的是,数据从map函数输出后,到reduce函数输入前的这段过程,这段过程中所发生的事情,叫做shuffle。

Map端的shuffle:



上图可能是某个map task的运行情况。拿它与官方图的左半边比较,会发现很多不一致。官方图没有清楚地说明partition, sort与combiner到底作用在哪个阶段。我画了这张图,希望让大家清晰地了解从map数据输入到map端所有数据准备好的全过程。

整个流程我分了四步。简单些可以这样说,每个map task都有一个内存缓冲区,存储着map的输出结果,当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘,当整个map task结束后再对磁盘中这个map task产生的所有临时文件做合并,生成最终的正式输出文件,然后等待reduce task来拉数据。

当然这里的每一步都可能包含着多个步骤与细节,下面我对细节来一一说明:
1、在map task执行时,它的输入数据来源于HDFS的block,当然在MapReduce概念中,map task只读取split。Split与block的对应关系可能是多对一,默认是一对一。在WordCount例子里,假设map的输入数据都是像“aaa”这样的字符串。

2、在经过mapper的运行后,我们得知mapper的输出是这样一个key/value对: key是“aaa”, value是数值1。因为当前map端只做加1的操作,在reduce task里才去合并结果集。前面我们知道这个job有3个reduce task,到底当前的“aaa”应该交由哪个reduce去做呢,是需要现在决定的。

MapReduce提供Partitioner接口,它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力,如果用户自己对Partitioner有需求,可以订制并设置到job上。

在我们的例子中,“aaa”经过Partitioner后返回0,也就是这对值应当交由第一个reducer来处理。接下来,需要将数据写入内存缓冲区中,缓冲区的作用是批量收集map结果,减少磁盘IO的影响。我们的key/value对以及Partition的结果都会被写入缓冲区。当然写入之前,key与value值都会被序列化成字节数组。

3、这个内存缓冲区是有大小限制的,默认是100MB。当map task的输出结果很多时,就可能会撑爆内存,所以需要在一定条件下将缓冲区中的数据临时写入磁盘,然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为Spill,中文可译为溢写,字面意思很直观。这个溢写是由单独线程来完成,不影响往缓冲区写map结果的线程。溢写线程启动时不应该阻止map的结果输出,所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8,也就是当缓冲区的数据已经达到阈值(buffer size * spill percent = 100MB * 0.8 = 80MB),溢写线程启动,锁定这80MB的内存,执行溢写过程。Map task的输出结果还可以往剩下的20MB内存中写,互不影响。当溢写线程启动后,需要对这80MB空间内的key做排序(Sort)。排序是MapReduce模型默认的行为,这里的排序也是对序列化的字节做的排序。

在这里我们可以想想,因为map task的输出是需要发送到不同的reduce端去,而内存缓冲区没有对将发送到相同reduce端的数据做合并,那么这种合并应该是体现是磁盘文件中的。从官方图上也可以看到写到磁盘中的溢写文件是对不同的reduce端的数值做过合并。所以溢写过程一个很重要的细节在于,如果有很多个key/value对需要发送到某个reduce端去,那么需要将这些key/value值拼接到一块,减少与partition相关的索引记录。

在针对每个reduce端而合并数据时,有些数据可能像这样:“aaa”/1, “aaa”/1。对于WordCount例子,就是简单地统计单词出现的次数,如果在同一个map task的结果中有很多个像“aaa”一样出现多次的key,我们就应该把它们的值合并到一块,这个过程叫reduce也叫combine。但MapReduce的术语中,reduce只指reduce端执行从多个map task取数据做计算的过程。除reduce外,非正式地合并数据只能算做combine了。其实大家知道的,MapReduce中将Combiner等同于Reducer。

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

4、 每次溢写会在磁盘上生成一个溢写文件,如果map的输出结果真的很大,有多次这样的溢写发生,磁盘上相应的就会有多个溢写文件存在。当map task真正完成时,内存缓冲区中的数据也全部溢写到磁盘中形成一个溢写文件。最终磁盘中会至少有一个这样的溢写文件存在(如果map的输出结果很少,当map执行完成时,只会产生一个溢写文件),因为最终的文件只有一个,所以需要将这些溢写文件归并到一起,这个过程就叫做Merge。Merge是怎样的?如前面的例子,“aaa”从某个map task读取过来时值是5,从另外一个map 读取时值是8,因为它们有相同的key,所以得merge成group。什么是group。对于“aaa”就是像这样的:{“aaa”, [5, 8, 2, …]},数组中的值就是从不同溢写文件中读取出来的,然后再把这些值加起来。请注意,因为merge是将多个溢写文件合并到一个文件,所以可能也有相同的key存在,在这个过程中如果client设置过Combiner,也会使用Combiner来合并相同的key。

至此,map端的所有工作都已结束,最终生成的这个文件也存放在TaskTracker够得着的某个本地目录内。每个reduce task不断地通过RPC从JobTracker那里获取map task是否完成的信息,如果reduce task得到通知,获知某台TaskTracker上的map task执行完成,Shuffle的后半段过程开始启动。

Reduce端shuffle:
reduce task在执行之前的工作就是不断地拉取当前job里每个map task的最终结果,然后对从不同地方拉取过来的数据不断地做merge,也最终形成一个文件作为reduce task的输入文件



Shuffle在reduce端的过程也能用图上标明的三点来概括。当前reduce copy数据的前提是它要从JobTracker获得有哪些map task已执行结束,这段过程不表,有兴趣的朋友可以关注下。Reducer真正运行之前,所有的时间都是在拉取数据,做merge,且不断重复地在做。如前面的方式一样,下面我也分段地描述reduce 端的Shuffle细节:
1、Copy过程,简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher),通过HTTP方式请求map task所在的TaskTracker获取map task的输出文件。因为map task早已结束,这些文件就归TaskTracker管理在本地磁盘中。

2、Merge阶段。这里的merge如map端的merge动作,只是数组中存放的是不同map端copy来的数值。Copy过来的数据会先放入内存缓冲区中,这里的缓冲区大小要比map端的更为灵活,它基于JVM的heap size设置,因为Shuffle阶段Reducer不运行,所以应该把绝大部分的内存都给Shuffle用。这里需要强调的是,merge有三种形式:1)内存到内存  2)内存到磁盘  3)磁盘到磁盘。默认情况下第一种形式不启用,让人比较困惑,是吧。当内存中的数据量到达一定阈值,就启动内存到磁盘的merge。与map 端类似,这也是溢写的过程,这个过程中如果你设置有Combiner,也是会启用的,然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行,直到没有map端的数据时才结束,然后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。

3、Reducer的输入文件。不断地merge后,最后会生成一个“最终文件”。为什么加引号?因为这个文件可能存在于磁盘上,也可能存在于内存中。对我们来说,当然希望它存放于内存中,直接作为Reducer的输入,但默认情况下,这个文件是存放于磁盘中的。当Reducer的输入文件已定,整个Shuffle才最终结束。然后就是Reducer执行,把结果放到HDFS上。

Shuffle产生的意义是什么?
Shuffle过程的期望可以有:
1.完整地从map task端拉取数据到reduce 端。
2.在跨节点拉取数据时,尽可能地减少对带宽的不必要消耗。
3.减少磁盘IO对task执行的影响。

每个map task都有一个内存缓冲区,存储着map的输出结果,当缓冲区快满的时候需要将缓冲区的数据该如何处理?
每个map task都有一个内存缓冲区,存储着map的输出结果,当缓冲区快满的时候需要将缓冲区的数据以一个临时文件的方式存放到磁盘,当整个map task结束后再对磁盘中这个map task产生的所有临时文件做合并,生成最终的正式输出文件,然后等待reduce task来拉数据。

MapReduce提供Partitioner接口,它的作用是什么?
MapReduce提供Partitioner接口,它的作用就是根据key或value及reduce的数量来决定当前的这对输出数据最终应该交由哪个reduce task处理。默认对key hash后再以reduce task数量取模。默认的取模方式只是为了平均reduce的处理能力,如果用户自己对Partitioner有需求,可以订制并设置到job上。

什么是溢写?
在一定条件下将缓冲区中的数据临时写入磁盘,然后重新利用这块缓冲区。这个从内存往磁盘写数据的过程被称为Spill,中文可译为溢写。

溢写为什么不影响往缓冲区写map结果的线程?
溢写线程启动时不应该阻止map的结果输出,所以整个缓冲区有个溢写的比例spill.percent。这个比例默认是0.8,也就是当缓冲区的数据已经达到阈值(buffer size * spill percent = 100MB * 0.8 = 80MB),溢写线程启动,锁定这80MB的内存,执行溢写过程。Map task的输出结果还可以往剩下的20MB内存中写,互不影响。

当溢写线程启动后,需要对这80MB空间内的key做排序(Sort)。排序是MapReduce模型默认的行为,这里的排序也是对谁的排序?
当溢写线程启动后,需要对这80MB空间内的key做排序(Sort)。排序是MapReduce模型默认的行为,这里的排序也是对序列化的字节做的排序。

溢写过程中如果有很多个key/value对需要发送到某个reduce端去,那么如何处理这些key/value值?
如果有很多个key/value对需要发送到某个reduce端去,那么需要将这些key/value值拼接到一块,减少与partition相关的索引记录。

哪些场景才能使用Combiner呢?
Combiner的输出是Reducer的输入,Combiner绝不能改变最终的计算结果。所以从我的想法来看,Combiner只应该用于那种Reduce的输入key/value与输出key/value类型完全一致,且不影响最终结果的场景。比如累加,最大值等。Combiner的使用一定得慎重,如果用好,它对job执行效率有帮助,反之会影响reduce的最终结果。

Merge的作用是什么?
最终磁盘中会至少有一个这样的溢写文件存在(如果map的输出结果很少,当map执行完成时,只会产生一个溢写文件),因为最终的文件只有一个,所以需要将这些溢写文件归并到一起,这个过程就叫做Merge

每个reduce task不断的通过什么协议从JobTracker那里获取map task是否完成的信息?
每个reduce task不断地通过RPC从JobTracker那里获取map task是否完成的信息

reduce中Copy过程采用是什么协议?
Copy过程,简单地拉取数据。Reduce进程启动一些数据copy线程(Fetcher),通过HTTP方式请求map task所在的TaskTracker获取map task的输出文件。

reduce中merge过程有几种方式?
merge有三种形式:1)内存到内存  2)内存到磁盘  3)磁盘到磁盘。默认情况下第一种形式不启用,让人比较困惑,是吧。当内存中的数据量到达一定阈值,就启动内存到磁盘的merge。与map 端类似,这也是溢写的过程,这个过程中如果你设置有Combiner,也是会启用的,然后在磁盘中生成了众多的溢写文件。第二种merge方式一直在运行,直到没有map端的数据时才结束,然后启动第三种磁盘到磁盘的merge方式生成最终的那个文件。


Mapper的缓冲区和Spill是什么
1、每个Mapper任务在内存中都会有一个缓冲区(环形缓冲区)用于存储任务的输出,默认的缓冲区大小为100M,数据量太大后会撑爆缓冲区,所有个阀值,超过缓冲区阀值后数据会被写入到本地磁盘,这个动作称为spill(溢出、流出),阀值为缓冲去大小的80%,如果默认是100M,那么阀值为80M。
2、缓冲区数据在spill到磁盘之前需要sort和combiner(对key进行合并的操作,数据:<a,1><a,1><a,1>,combiner过后合并成:<a,3>),一个Mapper任务执行完成后可能会spill到磁盘几个文件。
3、spill到磁盘的文件还有个分区概念(partition),spill到磁盘的文件内部进行了分区,分区的数量有Reducer的数量决定,所以一个文件内部的分区对应一Reducer,这是就是分区。
4、最后对生成的磁盘文件进行merge(合并),merge后的文件中有N个partition。

partition-->combiner-->merge

combiner是什么
combine分为map端和reduce端,作用是把同一个key的键值对合并在一起,并不是所有的归约工作都可以使用Combiner来做。比如求平均值就不能使用Combiner。因为对于平均数的归约算法不能多次调用。

partition是什么
partition是分割map每个节点的结果,按照key分别映射给不同的reduce。

转自:http://mvplee.iteye.com/blog/2220149
  • 大小: 209.5 KB
  • 大小: 25.5 KB
  • 大小: 62.2 KB
分享到:
评论
发表评论

文章已被作者锁定,不允许评论。

相关推荐

    Hadoop mapreduce实现wordcount

    Map 阶段将原始输入数据(通常是文本文件)拆分成小块,并对每一块进行独立处理;Reduce 阶段则负责聚合 Map 阶段产生的中间结果,最终输出每个单词的总数。 【详细知识点】 1. **Hadoop MapReduce 概述**: ...

    Hadoop MapReduce实战手册(完整版)

    《Hadoop MapReduce实战手册》是一本专注于大数据处理技术的专著,主要针对Apache Hadoop中的MapReduce框架进行了深入的探讨。MapReduce是Hadoop生态系统中的核心组件之一,用于处理和生成大规模数据集。该书旨在...

    hadoop2x-eclipse-plugin

    2. 调试MapReduce任务:通过"Debug As" -&gt; "Hadoop Job",Eclipse提供了调试功能,可以在Mapper和Reducer中设置断点,逐步执行代码,查看变量状态,帮助找出问题所在。 六、优化与维护 Hadoop2x-eclipse-plugin还...

    Hadoop应用系列2--MapReduce原理浅析(上)

    在IT行业中,分布式计算系统是处理大规模数据的关键技术之一,而Hadoop作为开源的分布式计算框架,其核心组件MapReduce则是...通过持续学习和实践,开发者可以更好地利用MapReduce解决实际问题,驾驭海量数据的挑战。

    Java操作Hadoop Mapreduce基本实践源码

    此外,还需要配置Job对象,指定输入输出路径、Mapper和Reducer类以及其他的Hadoop配置参数。 例如,一个简单的WordCount程序可以如下编写: ```java // 定义Mapper类 public static class TokenizerMapper extends...

    hadoop jar包.rar

    - **Shuffle与Sort**:Reduce阶段前,Map输出会被归并和排序,确保相同键的键值对一起处理。 3. **Hadoop生态**: - **Hive**:基于Hadoop的数据仓库工具,用于SQL-like查询和分析大规模数据。 - **Pig**:提供...

    Hadoop-2.8.0-Day04-MapReduce编程案例-课件与资料.zip

    2. 数据输入与输出格式:了解InputFormat和OutputFormat接口,以及自定义输入输出格式的必要性。 3. 键值对处理:深入理解Writable接口,如何实现自定义的键值对类。 4. Shuffle与Sort过程:MapReduce如何自动进行...

    Hadoop_MapReduce教程.doc

    MapReduce的核心思想是将大问题分解为小部分,通过map阶段的并行处理,然后在reduce阶段进行聚合,从而实现高效的数据处理。 1. **Map阶段**: 在Map阶段,输入数据集被分割成多个块,每个块分配给集群中的一个...

    Hadoop_MapReduce教程

    - **输入切分**:首先,Hadoop MapReduce 将输入数据集切分为若干个数据块,每个数据块由一个 map 任务处理。 - **Map 阶段**:每个 map 任务处理一个数据块,产生中间键值对。 - **Shuffle 排序**:中间键值对...

    mapreduce 实现朴素贝叶斯算法-源码

    Map阶段将输入数据分割成多个小块,然后对每个块进行独立处理,生成中间键值对。Reduce阶段则负责聚合这些中间结果,进一步处理并输出最终结果。 **朴素贝叶斯算法:** 朴素贝叶斯算法基于贝叶斯定理,其核心思想是...

    03-Hadoop-MapReduce.docx

    1. Map阶段:将输入数据分割成多个键值对,然后应用用户定义的Map函数,生成中间键值对。 2. Shuffle阶段:对中间键值对进行排序和分区,为Reduce阶段做准备。 3. Reduce阶段:根据相同的键将中间键值对分组,并应用...

    理论部分-Hadoop MapReduce1

    1. Map 阶段:这是并行计算的第一步,它接收输入数据,并将其拆分成键值对的形式。在提供的 WordCount 示例中,`TokenizerMapper` 类实现了 Map 功能。它遍历文本行,使用 `StringTokenizer` 分割单词,并为每个单词...

    hadoop-common-2.6.0-bin-master.zip.rar

    MapReduce是Hadoop处理大数据的核心计算模型,它将大规模数据集分割为小块(称为“map”阶段),并在集群中的多台机器上并行处理,然后将结果合并(称为“reduce”阶段)。Driver在MapReduce程序中扮演着至关重要的...

    hadoop2.7 source code

    - Reducer:源代码在`hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapred/Reducer.java`,对Mapper输出的中间结果进行聚合和处理。...

    Hadoop (十三)Hadoop-MR编程 -- 【模拟qq推荐你可能认识的人】

    在`main`方法中,我们配置Job,并指定输入输出路径以及Mapper和Reducer类: ```java Configuration conf = new Configuration(); Job job = Job.getInstance(conf, "qq_recommendation"); job.setJarByClass...

    hadoop MapReduce教材

    Shuffle和Sort阶段是MapReduce过程中的重要环节,它们在Map任务完成后,对中间结果进行排序和分区,为Reduce任务提供有序的数据输入。 对于MapReduce的优化,有多种策略可以采用,例如减少数据溢出、优化数据压缩、...

    Hadoop (十五)Hadoop-MR编程 -- 【使用hadoop计算网页之间的PageRank值----编程】

    在本篇中,我们将深入探讨如何使用Hadoop MapReduce编程模型来计算网页之间的PageRank值。PageRank是Google搜索引擎算法的核心部分,它通过衡量网页之间的链接关系来评估每个网页的重要性。在Hadoop环境下,我们可以...

    006_hadoop中MapReduce详解_3

    为了提高性能,MapReduce允许用户进行各种优化,比如数据本地化(将数据处理任务调度到数据所在节点,减少网络传输)、Combiner(在Map阶段进行局部聚合,减少数据传输量)和Partitioner(控制数据如何在Reducer之间...

Global site tag (gtag.js) - Google Analytics