Hadoop中MapTask的并行度的决定机制

在MapReduce程序的运行中，并不是MapTask越多就越好。需要考虑数据量的多少及机器的配置。如果数据量很少，可能任务启动的时间都远远超过数据的处理时间。同样可不是越少越好。

那么应该如何切分呢？

假如我们有一个300M的文件，它会在HDFS中被切成3块。0-128M,128-256M,256-300M。并被放置到不同的节点上去了。在MapReduce任务中，这3个Block会被分给3个MapTask。

MapTask在任务切片时实际上也是分配一个范围，只是这个范围是逻辑上的概念，与block的物理划分没有什么关系。但在实践过程中如果MapTask读取的数据不在运行的本机，则必须通过网络进行数据传输，对性能的影响非常大。所以常常采取的策略是就按照块的存储切分MapTask，使得每个MapTask尽可能读取本机的数据。

如果一个Block非常小，也可以把多个小Block交给一个MapTask。

所以MapTask的切分要看情况处理。默认的实现是按照Block大小进行切分。MapTask的切分工作由客户端（我们写的main方法）负责。一个切片就对应一个MapTask实例。

MapTask并行度的决定机制

一个job的map阶段并行度由客户端在提交job时决定。

而客户端对map阶段并行度的规划的基本逻辑为：

将待处理数据执行逻辑切片（即按照一个特定切片大小，将待处理数据划分成逻辑上的多个split），然后每一个split分配一个mapTask并行实例处理

这段逻辑及形成的切片规划描述文件，由FileInputFormat实现类的getSplits()方法完成，其过程如下图：

FileInputFormat切片机制

1、切片定义在InputFormat类中的getSplit()方法

2、FileInputFormat中默认的切片机制：

简单地按照文件的内容长度进行切片

切片大小，默认等于block大小

切片时不考虑数据集整体，而是逐个针对每一个文件单独切片

比如待处理数据有两个文件：

file1.txt    320M
file2.txt    10M

经过FileInputFormat的切片机制运算后，形成的切片信息如下：

file1.txt.split1--  0~128
file1.txt.split2--  128~256
file1.txt.split3--  256~320
file2.txt.split1--  0~10M

3、FileInputFormat中切片的大小的参数配置

通过分析源码，在FileInputFormat中，计算切片大小的逻辑：Math.max(minSize, Math.min(maxSize, blockSize)); 切片主要由这几个值来运算决定

minsize：默认值：1

配置参数：

mapreduce.input.fileinputformat.split.minsize

maxsize：默认值：Long.MAXValue

配置参数：mapreduce.input.fileinputformat.split.maxsize

blocksize

因此，默认情况下，切片大小=blocksize

maxsize（切片最大值）：

参数如果调得比blocksize小，则会让切片变小，而且就等于配置的这个参数的值

minsize （切片最小值）：

参数调的比blockSize大，则可以让切片变得比blocksize还大

选择并发数的影响因素：

运算节点的硬件配置

运算任务的类型：CPU密集型还是IO密集型

运算任务的数据量

Map并行度的经验之谈

如果硬件配置为2*12core + 64G，恰当的map并行度是大约每个节点20-100个map，最好每个map的执行时间至少一分钟。

如果job的每个map或者 reduce task的运行时间都只有30-40秒钟，那么就减少该job的map或者reduce数，每一个task(map|reduce)的setup和加入到调度器中进行调度，这个中间的过程可能都要花费几秒钟，所以如果每个task都非常快就跑完了，就会在task的开始和结束的时候浪费太多的时间。

配置task的JVM重用可以改善该问题：

（mapred.job.reuse.jvm.num.tasks，默认是1，表示一个JVM上最多可以顺序执行的task

数目（属于同一个Job）是1。也就是说一个task启一个JVM）

如果input的文件非常的大，比如1TB，可以考虑将hdfs上的每个block size设大，比如设成256MB或者512MB