Hadoop MapReduce

我会带着你远行 2021-10-14 16:51 425阅读 0赞

### 一、简介 ###

一种分布式的计算方式指定一个Map（映射）函数，用来把一组键值对映射成一组新的键值对，指定并发的Reduce（归约）函数，用来保证所有映射的键值对中的每一个共享相同的键组

*  **Pattern**

![aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXBhdHRlcm4ucG5n][]

map: (K1, V1) → list(K2, V2) combine: (K2, list(V2)) → list(K2, V2) reduce: (K2, list(V2)) → list(K3, V3)

Map输出格式和Reduce输入格式一定是相同的

*  **基本流程**

MapReduce主要是先读取文件数据，然后进行Map处理，接着Reduce处理，最后把处理结果写到文件中

![aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXByb2Nlc3Mtb3ZlcnZpZXcucG5n][]

*  **详细流程**

![aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXByb2Nlc3MucG5n][]

*  **多节点下的流程**

![aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXByb2Nlc3MtY2x1c3Rlci5wbmc][]

*  **主要过程**

****![aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLWRhdGEtcHJvY2Vzcy5wbmc][]

### 二、读取数据 ###

通过InputFormat决定读取的数据的类型，然后拆分成一个个InputSplit，每个InputSplit对应一个Map处理，RecordReader读取InputSplit的内容给Map

*  **InputFormat**

决定读取数据的格式，可以是文件或数据库等

*  **功能**

1.  验证作业输入的正确性，如格式等
2.  将输入文件切割成逻辑分片(InputSplit)，一个InputSplit将会被分配给一个独立的Map任务
3.  提供RecordReader实现，读取InputSplit中的"K-V对"供Mapper使用

*  **方法**

**List getSplits():** 获取由输入文件计算出输入分片(InputSplit)，解决数据或文件分割成片问题

**RecordReader <k,v>createRecordReader():</k,v>** 创建RecordReader，从InputSplit中读取数据，解决读取分片中数据问题

*  **InputSplit**

代表一个个逻辑分片，并没有真正存储数据，只是提供了一个如何将数据分片的方法

Split内有Location信息，利于数据局部化

一个InputSplit给一个单独的Map处理

public abstract class InputSplit {
          /**
           * 获取Split的大小，支持根据size对InputSplit排序.
           */
          public abstract long getLength() throws IOException, InterruptedException;
    
          /**
           * 获取存储该分片的数据所在的节点位置.
           */
          public abstract String[] getLocations() throws IOException, InterruptedException;
    }

*  **RecordReader**

将InputSplit拆分成一个个<key,value>对给Map处理，也是实际的文件读取分隔对象</key,value>

*  **大量小文件如何处理**

CombineFileInputFormat可以将若干个Split打包成一个，目的是避免过多的Map任务（因为Split的数目决定了Map的数目，大量的Mapper Task创建销毁开销将是巨大的）

*  **怎么计算split的**

通常一个split就是一个block（FileInputFormat仅仅拆分比block大的文件），这样做的好处是使得Map可以在存储有当前数据的节点上运行本地的任务，而不需要通过网络进行跨节点的任务调度

通过mapred.min.split.size， mapred.max.split.size, block.size来控制拆分的大小

如果mapred.min.split.size大于block size，则会将两个block合成到一个split，这样有部分block数据需要通过网络读取

如果mapred.max.split.size小于block size，则会将一个block拆成多个split，增加了Map任务数（Map对split进行计算且上报结果，关闭当前计算打开新的split均需要耗费资源）

先获取文件在HDFS上的路径和Block信息，然后根据splitSize对文件进行切分（ splitSize = computeSplitSize(blockSize, minSize, maxSize) ），默认splitSize 就等于blockSize的默认值（64m）

public List<InputSplit> getSplits(JobContext job) throws IOException {
        // 首先计算分片的最大和最小值。这两个值将会用来计算分片的大小
        long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
        long maxSize = getMaxSplitSize(job);
    
        // generate splits
        List<InputSplit> splits = new ArrayList<InputSplit>();
        List<FileStatus> files = listStatus(job);
        for (FileStatus file: files) {
            Path path = file.getPath();
            long length = file.getLen();
            if (length != 0) {
                  FileSystem fs = path.getFileSystem(job.getConfiguration());
                // 获取该文件所有的block信息列表[hostname, offset, length]
                  BlockLocation[] blkLocations = fs.getFileBlockLocations(file, 0, length);
                // 判断文件是否可分割，通常是可分割的，但如果文件是压缩的，将不可分割
                  if (isSplitable(job, path)) {
                    long blockSize = file.getBlockSize();
                    // 计算分片大小
                    // 即 Math.max(minSize, Math.min(maxSize, blockSize));
                    long splitSize = computeSplitSize(blockSize, minSize, maxSize);
    
                    long bytesRemaining = length;
                    // 循环分片。
                    // 当剩余数据与分片大小比值大于Split_Slop时，继续分片， 小于等于时，停止分片
                    while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
                          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
                          splits.add(makeSplit(path, length-bytesRemaining, splitSize, blkLocations[blkIndex].getHosts()));
                          bytesRemaining -= splitSize;
                    }
                    // 处理余下的数据
                    if (bytesRemaining != 0) {
                        splits.add(makeSplit(path, length-bytesRemaining, bytesRemaining, blkLocations[blkLocations.length-1].getHosts()));
                    }
                } else {
                    // 不可split，整块返回
                    splits.add(makeSplit(path, 0, length, blkLocations[0].getHosts()));
                }
            } else {
                // 对于长度为0的文件，创建空Hosts列表，返回
                splits.add(makeSplit(path, 0, length, new String[0]));
            }
        }
    
        // 设置输入文件数量
        job.getConfiguration().setLong(NUM_INPUT_FILES, files.size());
        LOG.debug("Total # of splits: " + splits.size());
        return splits;
    }

*  **分片间的数据如何处理**

split是根据文件大小分割的，而一般处理是根据分隔符进行分割的，这样势必存在一条记录横跨两个split

![aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXNwbGl0LnBuZw][]

解决办法是只要不是第一个split，都会远程读取一条记录。不是第一个split的都忽略到第一条记录

### 三、Mapper ###

主要是读取InputSplit的每一个Key,Value对并进行处理

public class Mapper<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
        /**
         * 预处理，仅在map task启动时运行一次
         */
        protected void setup(Context context) throws  IOException, InterruptedException {
        }
    
        /**
         * 对于InputSplit中的每一对<key, value>都会运行一次
         */
        @SuppressWarnings("unchecked")
        protected void map(KEYIN key, VALUEIN value, Context context) throws IOException, InterruptedException {
            context.write((KEYOUT) key, (VALUEOUT) value);
        }
    
        /**
         * 扫尾工作，比如关闭流等
         */
        protected void cleanup(Context context) throws IOException, InterruptedException {
        }
    
        /**
         * map task的驱动器
         */
        public void run(Context context) throws IOException, InterruptedException {
            setup(context);
            while (context.nextKeyValue()) {
                map(context.getCurrentKey(), context.getCurrentValue(), context);
            }
            cleanup(context);
        }
    }
    
    public class MapContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> extends TaskInputOutputContext<KEYIN, VALUEIN, KEYOUT, VALUEOUT> {
        private RecordReader<KEYIN, VALUEIN> reader;
        private InputSplit split;
    
        /**
         * Get the input split for this map.
         */
        public InputSplit getInputSplit() {
            return split;
        }
    
        @Override
        public KEYIN getCurrentKey() throws IOException, InterruptedException {
            return reader.getCurrentKey();
        }
    
        @Override
        public VALUEIN getCurrentValue() throws IOException, InterruptedException {
            return reader.getCurrentValue();
        }
    
        @Override
        public boolean nextKeyValue() throws IOException, InterruptedException {
            return reader.nextKeyValue();
        }
    }

### 四、Shuffle ###

对Map的结果进行排序并传输到Reduce进行处理 Map的结果并不直接存放到硬盘,而是利用缓存做一些预排序处理 Map会调用Combiner，压缩，按key进行分区、排序等，尽量减少结果的大小 每个Map完成后都会通知Task，然后Reduce就可以进行处理

![aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXByb2Nlc3MucG5n][]

*  **Map端**

当Map程序开始产生结果的时候，并不是直接写到文件的，而是利用缓存做一些排序方面的预处理操作

每个Map任务都有一个循环内存缓冲区（默认100MB），当缓存的内容达到80%时，后台线程开始将内容写到文件，此时Map任务可以继续输出结果，但如果缓冲区满了，Map任务则需要等待

写文件使用round-robin方式。在写入文件之前，先将数据按照Reduce进行分区。对于每一个分区，都会在内存中根据key进行排序，如果配置了Combiner，则排序后执行Combiner（Combine之后可以减少写入文件和传输的数据）

每次结果达到缓冲区的阀值时，都会创建一个文件，在Map结束时，可能会产生大量的文件。在Map完成前，会将这些文件进行合并和排序。如果文件的数量超过3个，则合并后会再次运行Combiner（1、2个文件就没有必要了）

如果配置了压缩，则最终写入的文件会先进行压缩，这样可以减少写入和传输的数据

一旦Map完成，则通知任务管理器，此时Reduce就可以开始复制结果数据

*  **Reduce端**

Map的结果文件都存放到运行Map任务的机器的本地硬盘中

如果Map的结果很少，则直接放到内存，否则写入文件中

同时后台线程将这些文件进行合并和排序到一个更大的文件中（如果文件是压缩的则需要先解压）

当所有的Map结果都被复制和合并后，就会调用Reduce方法

Reduce结果会写入到HDFS中

*  **调优**

一般的原则是给shuffle分配尽可能多的内存，但前提是要保证Map、Reduce任务有足够的内存

对于Map，主要就是避免把文件写入磁盘，例如使用Combiner，增大io.sort.mb的值

对于Reduce，主要是把Map的结果尽可能地保存到内存中，同样也是要避免把中间结果写入磁盘。默认情况下，所有的内存都是分配给Reduce方法的，如果Reduce方法不怎么消耗内存，可以mapred.inmem.merge.threshold设成0，mapred.job.reduce.input.buffer.percent设成1.0

在任务监控中可通过Spilled records counter来监控写入磁盘的数，但这个值是包括map和reduce的

对于IO方面，可以Map的结果可以使用压缩，同时增大buffer size（io.file.buffer.size，默认4kb）

[aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXBhdHRlcm4ucG5n]: /images/20211015/f6f69d6b364546de817c953ef5899556.png
[aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXByb2Nlc3Mtb3ZlcnZpZXcucG5n]: /images/20211015/836d622b987146488a40d6bf352b113a.png
[aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXByb2Nlc3MucG5n]: /images/20211015/ec0c1dee7b234d9d9fb8178cf437bf3e.png
[aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXByb2Nlc3MtY2x1c3Rlci5wbmc]: /images/20211015/91107fb19e644fed9a8ca3b348182218.png
[aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLWRhdGEtcHJvY2Vzcy5wbmc]: /images/20211015/3176305254a94658a75d36203094f706.png
[aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXNwbGl0LnBuZw]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9hdHRzLnczY3NjaG9vbC5jbi9hdHRhY2htZW50cy9pbWFnZS93ay9oYWRvb3AvbWFwcmVkdWNlLXNwbGl0LnBuZw