MapReduce-深度解析

MR Workflow
图片摘自于Alexey Grishchenko博文

Map-side

InputFormat Class :

• getSplits: the set of input data splits 返回一组输入数据的拆分文件

• getRecordReader: iterable interface for reading all the records from a single input

  ​ 从单个输入文件提供可迭代的接口用于读取输入数据

文件块大小取决于InputFormat（自定义输入文件拆分需要继承此FileInputFormat）

• 输入文件是文本类型需要配置dfs.blocksize的大小（hdfs-site.xml）

• gzip类型的压缩文件，不可拆分需要用输入整个文件等。

每个mapper处理一个输入拆分块文件，大多数时处理128MB大小文件若输入文件是以GB或PB或更大。

map 函数运用于输入拆分文件的每对键值对（k，v），它们的每对键值对都有RecordReader 返回。

根据业务逻辑需求，在mapper对每对键值做处理输出结果，将结果由Context文本类传给reducer端。

负责收集map的输出数据(如文件)是mapreduce.job.map.output.collector 属性 （mapreduce-default.xml），默认是由org.apache.hadoop.mapred.MapTask$MapOutputBuffer 实现。

Map 函数的输出结果首先调用从Partitioner类getPartition方法。 getPartition需要传入键值对（k，v） 还有reduce的任务数量（numReduceTasks），然后返回这些键值对匹配的分区。

public class HashPartitioner<K, V> extends Partitioner<K, V> {
    public int getPartition(K key, V value, int numReduceTasks) {
        return (key.hashCode() & Integer.MAX_VALUE) % numReduceTasks;
    }
}

接着把输出数据键值对和分区号一同写入环形缓冲区 （ring buffer）, 此缓冲区大小由mapreduce.task.io.sort.mb （mapreduce-default.xml) 定义， 默认为100MB， 最大map输出数据信息允许在ring buffer占用的大小. 若输出数据大小大于此值，数据会被溢出到（写入）硬盘。

注意，map 输出的环形缓冲区默认是比输入文件的拆分块（默认是128MB） 要小，所以多余的会被写到硬盘中。

溢出的操作是由新的线程执行，起初环形缓冲区的大小为0.8mapreduce.map.sort.spill.percent （mapreduce-default.xml）， 所以最初的缓冲区大小为80MB。 map task的输出溢出文件大小默认是大于80MB的，多余的是会被写到硬盘里。

溢出用开启新线程来处理输出文件数据是为了让mapper在处理溢出同时也能继续执行处理输入文件数据。

注意： 当处理输入数据（FileInputFormat）的数率比溢出的数率快时，Mapper函数会停止工作 因为 内存的环形缓冲区可能会达到100%。在这种情况，mapper函数会阻塞并等待溢出线程为下一批输出数据处理清空一些内存空间。

溢出线程会把环形缓冲区数据写到mapper函数调用的服务器的本地的文件里。溢出线程写出本地的路劲是由mapreduce.job.local.dir （mapred-default.xml）定义，此配置属性包含了一组由集群上MapReduce任务用到的路径来存储零时数据。 文件夹被一个接一个使用。 写入前，数据会以快速排序 进行排序：comparator函数先对比partition分区号然后再对比key值，以至于先排分区，再每个分区排序key值。

排序完成后，Combiner被调用用于减少输入硬盘的数据量。Combiner的输出会被写入硬盘。 有个边际情况是当mapper产生的输出数据过于大不能容入内存时（大于输出缓冲区大小），sorter和combiner时不会被调用； 这时，mapper的输出数据就会直接写入硬盘。

无论mapper输出数据是多大，输出完成时“Spill”是肯定会至少调用一次。

（详细信息查看：ShuffleHandler.class#sendMapOutput）

protected ChannelFuture sendMapOutput(ChannelHandlerContext ctx, Channel ch, ​ String user, String mapId, int reduce, MapOutputInfo mapOutputInfo)

同样的map任务会执行，如sort和combine，还有写入本地磁盘文件。

每个由溢出线程溢出文件有个索引包含了每个溢出文件的分区信息：分区从哪开始于哪结束。 这些索引被存在内存中，此内存块大小由mapreduce.task.index.cache.limit.bytes决定，默认为1MB。 内存不足时，所有的下一批生成的溢出文件的索引会与溢出文件一起被写入到磁盘。

当mapper处理输出文件与最后溢出结束时，溢出线程完成结束而合并阶段开始。 在合并时，所有的溢出文件应该合并一块为单个map的输出文件。一个合并过程默认可以处理10个溢出文件（由mapreduce.task.io.sort.factor决定）。

<property>
  <name>mapreduce.task.io.sort.factor</name>
  <value>10</value>
  <description>The number of streams to merge at once while sorting
  files.  This determines the number of open file handles.</description>
</property>

若溢出文件大于此属性值， 剩余的文件会被合并为一个大文件。

合并期间，若被合并的文件大于（min.num.spills.for.combine , 默认为3), 写入磁盘前，combiner会在merge结果上执行。

MapTask的结果是一个包含了所有的Mapper输出结果与描述分区开始-结束的索引信息的输出文件， 这些分区开始-结束索引信息便于ReduceTask能够从磁盘中索取每个reducer任务的运行相关数据。

---

Reduce-Side

Map的任务数量由拆分块决定的，然而reduce的任务用户自己设置的（mapreduce.job.reduces， 默认为1）。 Reduce端的shuffle的实现由mapreduce.job.reduce.shuffle.consumer.plugin.class 属性决定，默认为org.apache.hadoop.mapreduce.task.reduce.Shuffle。

Reduce端做的第一件事就开启”Event Fetcher" 线程，从Application Master得到Mapper的状态并监听mapper的事件是否执行结束。 当mapper结束自己的shuffle过程，mapper的输出文件数据传送到多个“Fetcher”线程的其中一个。 “Fetcher”线程是由mapreduce.reduce.shuffle.parallelcopies 决定的，默认为5个，这意味着单个reduce任务, 有五个线程来从mapper端并行拷贝数据。Fetch的节点间的传输是通过HTTP或HTTPS的协议连接fetcher到相应的DataNode URL。

<property>
  <name>mapreduce.reduce.shuffle.parallelcopies</name>
  <value>5</value>
  <description>The default number of parallel transfers run by reduce
  during the copy(shuffle) phase.
  </description>
</property>

“Fetcher”从Mapper端下载的数据都被存在内存中，此内存大小占用reducer内存比例由mapreduce.reduce.shuffle.input.buffer.percent决定，总的reducer内存是mapreduce.reduce.memory.totalbytes。

<property>
  <name>mapreduce.reduce.shuffle.merge.percent</name>
  <value>0.66</value>
  <description>The usage threshold at which an in-memory merge will be
  initiated, expressed as a percentage of the total memory allocated to
  storing in-memory map outputs, as defined by
  mapreduce.reduce.shuffle.input.buffer.percent.
  </description>
</property>

<property>
  <name>mapreduce.reduce.memory.mb</name>
  <value>1024</value>
  <description>The amount of memory to request from the scheduler for each
  reduce task.
  </description>
</property>

若这些内存不够容纳，reducer会把数据存入reducer端的本地磁盘中的mapreduce.job.local.dir文件夹。

<property>
  <name>mapreduce.cluster.local.dir</name>
  <value>${hadoop.tmp.dir}/mapred/local</value>
  <description>The local directory where MapReduce stores intermediate
  data files.  May be a comma-separated list of
  directories on different devices in order to spread disk i/o.
  Directories that do not exist are ignored.
  </description>
</property>

fetcher索取相应文件数据后，merger线程开启工作。它们不会等待整个fetching过程完成而是开新线程与其并行执行。 Hadoop有三种merger线程。

1. InMemory merger （内存线程）
   1. 不能关闭。 由Reduce任务索取MapTask的输出数据占用的Reducer内存缓存超出了总内存允许的占用百分比reduce.shuffle.merge.percent而启动。 合并后执行combiner。输出写入硬盘，总会被调用至少一次。
2. MemToMem merger （内存到内存）
   1. 默认为关闭。 可由reduce.merge.memtomem.enabled 开启。 此线程合并内存中mapper的输出文件数据并写reduce输出到内存。当不同的MapTask输出文件大小达到mapreduce.reduce.merge.memtomem.threshold （默认为1000）， 线程会被启动。
3. OnDisk （硬盘）
   1. 在一次线程执行中， 当文件数量以（2 * task.io.sort.factor - 1) 上升 , 但是合并不超过mapreduce.task.io.sort.factor 的文件数量而启动。 OnDisk Merger线程合并本地磁盘的文件。

最后个线程， finalMerge 在reducer的主线程中运行，合并所有由InMemory和OnDisk在本地磁盘产生剩余的文件。 最终合并输出结果分布在RAM和硬盘之间。RAM最大允许使用为reduce 输入大小是由总的栈大小mapred.job.reduce.markreset.buffer.percent百分比来决定的，默认为0.

在这些所有的线程启动完成后，reducer会把输出写入HDFS文件系统中。

---

原文出自于，Alexey Grishchenko的Hadoop MapReduce Comprehensive Description

译者： 迈大_阿李同学