Mapreduce中maptask过程详解

Posted 2021-11-06 hellobigtable

一、Maptask并行度与决定机制

　　1.一个job任务的map阶段的并行度默认是由该任务的大小决定的；

　　2.一个split切分分配一个maprask来并行处理；

　　3.默认情况下，split切分的大小等于blocksize大小；

　　4.切片不是mapper类中对单词的切片，而是对每一个处理文件的单独切片。

　　eg.  默认情况下，一个maptask处理的文件大小为128M，比如一个400M的数据文件，就需要4个maptask并行来处理，而500M的数据文件也是需要4个maptask。

 

二、Maptask运行机制

　　1.读数据文件：执行类Driver通过InputFormat类读取文件中的数据；

　　2.mapper阶段：通过文件的大小决定了maptask的数量，然后mapper进行逻辑运行(读数据、切分、封装)；

　　3.OutputCollector阶段：mapper方法通过OutputCollector接口将KV对写入到环形缓冲区中(这个过程不需要我们处理我们)；

　　4.溢写阶段：环形缓冲区默认的大小为100M，当环形缓冲区中数据量到达阈值的80%的时候发生溢写，溢写的过程中会保证数据KV对使用默认的分区和排序(HashPartitioner分区、字典排序，而环形缓冲区大小和阈值的大小都是可以通过配置来修改的)；

　　5.归并排序：将溢写的数据进行合并排序。

 

三、MR的小文件优化案例

　　当许多个小文件上传到HDFS集群上时，每个小文件都将会占用一个blocksize的大小(128M)，而且在对它们进行MR计算时，一个文件就会开启一个maptask，这样会浪费很多的资源，下面有两种解决方案：

　　1.在文件上传到HDFS集群前，先将文件进行合并成一个大的文件，再上传到HDFS集群进行存储和计算；

　　2.若文件已经上传到HDFS集群，需要直接进行计算时，

　　可以再Driver类中设置输入流之前设置InputFormatClass属性为CombinerTextInputFormat(它的默认为TextInputFormat)，

　　原理是：CombineTextInputFormat类可以将多个小文件交给一个split切片，然后交给一个maptask来处理，即再Driver类中设置输入流FileInputFormat前加入代码：

job.setInputFormatClass(CombinerTextInputFormat.class);
CombinerTextInputFormat.setMaxInputSplitSize(job,4194304);    //设置切片最大值为4M
CombinerTextInputFormat.setMinInputSplitSize(job,3145725);    //设置切片最大值为3M

　　表示大小在3M~4M的文件会被方法一个切片中，那么如果有无数的小文件，一个maptask中大概会有28~42个小文件一起处理。

 

四、自定义分区Partitioner

　　在MR程序中，默认分区为HashPartitioner，以下为源码：

public class HashPartitioner<K, V> extends Partitioner<K, V> {
    public HashPartitioner() {
    }

    public int getPartition(K key, V value, int numReduceTasks) {
        return (key.hashCode() & 2147483647) % numReduceTasks;
    }
}

　　HashPartitioner继承了父类Partitioner，其中getPartition方法返回int值0（注释：分区数量决定了reducetask的数量，不分区reducetask值为1，所以一直返回int值0，也就只会产生一个结果文件！！！）

　　而如果我们想要进行自定义分区，就要重新定义一个分区类继承Partitioner类：

public class FlowPartitioner extends Partitioner<Text,FlowBean> {
    @Override
    public int getPartition(Text key, FlowBean value, int i) {
        //获取用来分区的电话号码前三位
        String phoneNum = key.toString().substring(0, 3);
        //设置分区逻辑
        int partitionNum = 4;
        if ("135".equals(phoneNum)){
            return 0;
        }else if ("137".equals(phoneNum)){
            return 1;
        }else if ("138".equals(phoneNum)){
            return 2;
        }else if ("139".equals(phoneNum)){
            return 3;
        }
        return partitionNum;
    }
}

　　我在流量统计案例中也写了该分区类，然后再Driver类中的InputFormat类之前加入设置的自定义分区代码：

 

job.setPartitionClass(PhoneNumPartitioner.class);
job.setNumReduceTasks(5);		(注意：输出文件数量要大于partitioner分区的数量)

 

　　总结：MR程序运算过程中，决定maptask个数的有块大小(blocksize)、数据文件大小、文件输入方式(小文件优化)；而决定reducetask个数的是分区(无分区时reducetask个数为1，生成一个结果文件)。