MapReduce中的mapper和reducer以及context

女爷i 2024-02-19 12:53 28阅读 0赞

### 关于MapReduce中的mapper ###

Mapper有setup()，map()，cleanup()和run()四个方法。其中setup()一般是用来进行一些map()前的准备工作，map()则一般承担主要的处理工作，cleanup()则是收尾工作如关闭文件或者执行map()后的K-V分发等。run()方法提供了setup->map->cleanup()的执行模板。

在MapReduce中，Mapper从一个输入分片中读取数据，然后经过Shuffle and Sort阶段，分发数据给Reducer，在Map端和Reduce端我们可能使用设置的Combiner进行合并，这在Reduce前进行。Partitioner控制每个K-V对应该被分发到哪个reducer\[我们的Job可能有多个reducer\]，Hadoop默认使用HashPartitioner，HashPartitioner使用key的hashCode对reducer的数量取模得来。

mapper的核心:

读取HDFS中的文件。每一行解析成一个<k,v>。每一个键值对调用一次map函数。  
<0,hello you> <10,hello me>

覆盖map()，接收1.1产生的<k,v>，进行处理，转换为新的<k,v>输出。  
<hello,1> <you,1> <hello,1> <me,1>

输出的<k,v>进行分区。默认分为一个区。详见《Partitioner》

对不同分区中的数据进行排序（按照k）、分组。分组指的是相同key的value放到一个集合中。 排序后：<hello,1> <hello,1> <me,1> <you,1> 分组后：<hello,\{1,1\}><me,\{1\}><you,\{1\}>

将输入内容经过map后解析成新的键值对.

Shuffle过程，也称Copy阶段。reduce task从各个map task上远程拷贝一片数据，并针对某一片数据，如果其大小超过一定的阀值，则写到磁盘上，否则直接放到内存中。

官方的Shuffle过程如上图所示，不过细节有错乱，官方图并没有说明partition、sort和combiner具体作用于哪个阶段。

注意：Shuffle过程是贯穿于map和reduce两个过程的！可以说shuffle是在map和reduce过程中间进行的.

其中shuffle有很重要的一个阶段就是sort(merge),将各个节点的map的值传过来进行合并,再传给reduce.

下图是我觉得很有代表性的一张图了:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NyZzE4NDM4NjEwNTc3_size_16_color_FFFFFF_t_70][]

其次mapper中有四个泛型,大家还记得吗?Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT>

分别代表的是:

LongWritable key Map函数的输入键

Text value Map函数的输入值

OutputCollector<Text,IntWritable> output 输出键

Report reporter 输出值

接写来mapper中我们要重写一个map方法:里面有一个context参数,是将map后的结果写进去

context.write

这里当时我是纠结了很久的,纠结到底会不会覆盖,事实证明不会覆盖,因为他是实现序列化的,不会覆盖,直接写进去的.

最后大家如果相练习mapreduce的话

推荐三个很经典的例子:

单词统计

天气统计

二度关系(很重要)

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2NyZzE4NDM4NjEwNTc3_size_16_color_FFFFFF_t_70]: https://image.dandelioncloud.cn/pgy_files/images/2024/01/29/9fe64c0febb24d19969d87b190a9a5e1.png