MapReduce中的map与reduce

￡神魔★判官ぃ 2022-09-24 04:29 187阅读 0赞

> ![20160725170835356][]

> 本文主要介绍MapReduce的map与reduce所包含的各各阶段

>         MapReduce中的每个map任务可以细分4个阶段：**record reader、mapper、combiner和partitioner**。map任务的输出被称
> 
> 为中间键和中间值，会被发送到reducer做后续处理。reduce任务可以分为4个阶段：**混排（shuffle）、****排序（sort）、reducer**
> 
> **和输****出格式（output format）**。map任务运行的节点会优先选择在数据所在的节点，因此，一般可以通过在本地机器上进行计算
> 
> 来减少数据的网络传输。

>         Mapreduce作业的输入是一系列存储在Hadoop分布式文件系统（HDFS）上的文件。在MapReduce中，这些文件通过输入格
> 
> 式（input format）被分成了一系列的输入split（input split）。输入split可以看做是文件在字节层面的分块表示，每个split由一
> 
> 个map任务负责处理。
> 
>   
> 
> 
> **record reader**
> 
>         record reader通过输入格式将输入split解析成记录。record reader的目的只是将输入数据解析成记录，但不负责解析记录
> 
> 本身。它将数据转化为键/值（key/value）对的形式，并传递给mapper处理。通常键是数据在文件中的位置，值是组成记录的
> 
> 数据块。
> 
> **map**
> 
>         在mapper中，用户定义的map代码通过处理record reader解析的每个键/值对来产生0个或多个新的键/值对结果。键/值的
> 
> 选择对MapReduce作业的完成效率来说非常重要。键是数据在reducer中处理时被分组的依据，值是reducer需要分析的数
> 
> 据。
> 
> **combiner**
> 
>         combiner是一个可选的本地reducer，可以在map阶段聚合数据。combiner通过执行用户指定的来自mapper的中间键对
> 
> map的中间结果做单个map范围内的聚合。
> 
>         例如，一个聚合的计数是每个部分计数的总和，用户可以先将每个中间结果取和，再将中间结果的和相加，从而得到最终
> 
> 结果。
> 
>         在很多情况下，这样可以明显地减少通过网络传输的数据量。在网络上发送一次（hello,3）要比三次（hello，1）节省更
> 
> 多的字节量。通过combiner可以产生特别大的性能提升，并且没有副作用，因此combiner的应用非常广泛。
> 
> **partitioner**
> 
>         partitioner的作用是将mapper（如果使用了combiner的话就是combiner）输出的键/值对拆分为分片（shard），每reducer
> 
> 对应一个分片。默认情况下，partitioner先计算目标的散列值（通常为md5值）。然后，通过reducer个数执行取模运算
> 
> key.hashCode()%(reducer的个数)。这种方式不仅能够随机地将整个键空间平均分发给每个reducer，同时也能确保不同mapper
> 
> 产生的相同键能被分发至同一个reducer。用户可以定制partitioner的默认行为，并可以使用更高级的模式，如排序。当然，一
> 
> 般情况下是不需要改写partitioner的。对于每个map任务，其分好区的数据最终会写入本地文件系统，等待其各自的reducer拉
> 
> 取。
> 
>   
> **混排和排序**
> 
>         reduce任务开始于混排和排序这一步骤。该步骤主要是将所有partitioner写入的输出文件拉取到运行reducer的本地机器
> 
> 上，然后将这些数据按照键排序并写到一个较大的数据列表中。排序的目的是将相同键的记录聚合在一起，这样其所对应的值
> 
> 就可以很方便地在reduce任务中进行迭代处理。这个过程完全不可定制，而且是由框架自动处理的。开发人员只能通过自定义
> 
> Comparator对象来确定键如何排序和分组。  
> 
> 
> **reduce**
> 
>         reducer将已经分好组的数据作为输入，并依次为每个键对应分组执行reduce函数。reduce函数的输入是键以及包含与该键
> 
> 对应的所有值的迭代器。在后文介绍的模式中，我们将看到在这个函数中有很多种处理方法。这些数据可以被聚合、过滤或以
> 
> 多种方式合并。当reduce函数执行完毕后，会将0个或多个键/值对发送到最后的处理步骤——输出格式。和map函数一样，因
> 
> 为reduce函数是业务处理逻辑的核心部分，所以不同作业的reduce函数也是不相同。
> 
> **输出格式**
> 
>         输出格式获取reduce函数输出的最终键/值对，并通过record write将它写入到输出文件中。每条记录的键和值默认通过tab
> 
> 分隔，不同记录通过换行符分隔。虽然一般情况下可以通过自定义实现非常多的输出格式，但是，不管什么格式，最终的结果
> 
> 都将写到HDFS上。
> 
>   
> 
> 
> **附：**
> 
> Mapper、Reducer类的源码，通过源码我们可以了解到在map、reduce阶段也可以做一些自定义处理。
> 
> Mapper类中有4个主要的函数分别是：setup、map、cleanup、run，源码如下：
> 
>     public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
>     
>       /**
>     	* The <code>Context</code> passed on to the {@link Mapper} implementations.
>     	*/
>       public abstract class Context
>         implements MapContext<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {
>       }
>       
>      /**
>        * Called once at the beginning of the task.
>        */
>       protected void setup(Context context
>                            ) throws IOException, InterruptedException {
>         // NOTHING
>       }
>     
>       /**
>        * Called once for each key/value pair in the input split. Most applications
>        * should override this, but the default is the identity function.
>        */
>       @SuppressWarnings("unchecked")
>       protected void map(KEYIN key, VALUEIN value, 
>                          Context context) throws IOException, InterruptedException {
>         context.write((KEYOUT) key, (VALUEOUT) value);
>       }
>     
>       /**
>        * Called once at the end of the task.
>        */
>       protected void cleanup(Context context
>                              ) throws IOException, InterruptedException {
>         // NOTHING
>       }
>       
>       /**
>        * Expert users can override this method for more complete control over the
>        * execution of the Mapper.
>        * @param context
>        * @throws IOException
>        */
>       public void run(Context context) throws IOException, InterruptedException {
>         setup(context);
>         try {
>           while (context.nextKeyValue()) {
>             map(context.getCurrentKey(), context.getCurrentValue(), context);
>           }
>         } finally {
>           cleanup(context);
>         }
>       }
>     }
> 
>         由源码可知，当调用到map时，通常会先执行一个setup函数，最后会执行一个cleanup函数。而默认情况下，这两个函数 的内容都是nothing。因此，
> 
> 当map方法不符合应用要求时，可以试着通过增加setup和cleanup的内容来满足应用的需求。
> 
> Recuder类中也有4个主要的函数分别是：setup、reduce、cleanup、run，源码如下：
> 
>     public class Reducer<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {
>     
>     	public abstract class Context
>     			implements ReduceContext<KEYIN,VALUEIN,KEYOUT,VALUEOUT> {
>     	}
>     
>     	protected void setup(Context context
>     	                    ) throws IOException, InterruptedException {
>     		// NOTHING
>     	}
>     
>     	@SuppressWarnings("unchecked")
>     	protected void reduce(KEYIN key, Iterable<VALUEIN> values, Context context
>     	                     ) throws IOException, InterruptedException {
>     		for(VALUEIN value: values) {
>     			context.write((KEYOUT) key, (VALUEOUT) value);
>     		}
>     	}
>     	
>     	protected void cleanup(Context context
>     	                      ) throws IOException, InterruptedException {
>     		// NOTHING
>     	}
>     
>     	public void run(Context context) throws IOException, InterruptedException {
>     		setup(context);
>     		try {
>     			while (context.nextKey()) {
>     				reduce(context.getCurrentKey(), context.getValues(), context);
>     				// If a back up store is used, reset it
>     				Iterator<VALUEIN> iter = context.getValues().iterator();
>     				if(iter instanceof ReduceContext.ValueIterator) {
>     					((ReduceContext.ValueIterator<VALUEIN>)iter).resetBackupStore();
>     				}
>     			}
>     		} finally {
>     			cleanup(context);
>     		}
>     	}
>     }
> 
>        由源码可知，当调用到reduce时，通常会先执行一个setup函数，最后会执行一个cleanup函数。而默认情况下，这两个函数的内容都是nothing。因此
> 
> 一般情况下reduce是中的setup和cleanup是没有内容的，当处理结果不符合或需要更多处理时我们要编写setup和cleanup的内容。  
>

[20160725170835356]: /images/20220719/c93e6e56e2ea40b3b37ae26ab881a2c8.png