hadoop各种输入方法(InputFormat)汇总

引自http://www.blogjava.net/shenh062326/archive/2012/07/03/hadoop.html

mapreduce 中 , 一个 job 的 map 个数 , 每个 map 处理的数据量是如何决定的呢 ? 另外每个 map 又是如何读取输入文件的内容呢 ? 用户是否可以自己决定输入方式 , 决定 map 个数呢 ? 这篇文章将详细讲述 hadoop 中各种 InputFormat 的功能和如何编写自定义的InputFormat.

 

简介 : mapreduce 作业会根据输入目录产生多个 map 任务 , 通过多个 map 任务并行执行来提高作业运行速度 , 但如果 map 数量过少 , 并行量低 , 作业执行慢 , 如果 map 数过多 , 资源有限 , 也会增加调度开销 . 因此 , 根据输入产生合理的 map 数 , 为每个 map 分配合适的数据量 , 能有效的提升资源利用率 , 并使作业运行速度加快 .

    在 mapreduce 中 , 每个作业都会通过 InputFormat 来决定 map 数量 . InputFormat 是一个接口, 提供两个方法:

InputSplit[] getSplits(JobConf job, int numSplits) throws IOException;

RecordReader<K, V> getRecordReader(InputSplit split,

                                     JobConf job,

                                     Reporter reporter) throws IOException;

    其中 getSplits 方法会根据输入目录产生 InputSplit 数组 , 每个 InputSplit 会相应产生一个 map 任务 , map 的输入定义在 InputSplit 中 . getRecordReader 方法返回一个 RecordReader 对象 , RecordReader 决定了 map 任务如何读取输入数据 , 例如一行一行的读取还是一个字节一个字节的读取 , 等等 .

    下图是 InputFormat 的实现类:

       (暂时无法上传)

    这理详细介绍 FileInputFormat 和 CombineFileInputFormat, 其它不常用 , 有兴趣的可以自己查看 hadoop 源码 .

 

FileInputFormat( 旧接口 org.apache.hadoop.mapred )

 

 

mapreduce 默认使用 TextInputFormat ， TextInputFormat 没有实现自己的 getSplits 方法 , 它 继承于 FileInputFormat, 因此使用了 FileInputFormat的.

org.apache.hadoop.mapred. FileInputFormat 的 getSplits 流程 :

两个配置

mapred.min.split.size         ( 一个 map 最小输入长度 ),

mapred.map.tasks                 ( 推荐 map 数量 )

如何决定每个 map 输入长度呢 ? 首先获取输入目录下所有文件的长度和 , 除以 mapred.map.tasks 得到一个推荐长度 goalSize, 然后通过式子 : Math.max (minSize, Math.min (goalSize, blockSize)) 决定 map 输入长度 . 这里的 minSize 为 mapred.min.split.size, blockSize 为相应文件的 block 长度 . 这式子能保证一个 map 的输入至少大于 mapred.min.split.size, 对于推荐的 map 长度 , 只有它的长度小于 blockSize 且大于 mapred.min.split.size 才会有效果 . 由于 mapred.min.split.size 默认长度为 1, 因此通常情况下只要小于 blockSize 就有效果 , 否则使用 blockSize 做为 map 输入长度 .

因此 , 如果想增加 map 数 , 可以把 mapred.min.split.size 调小 ( 其实默认值即可 ), 另外还需要把 mapred.map.tasks 设置大.

如果需要减少 map 数 , 可以把 mapred.min.split.size 调大 , 另外把 mapred.map.tasks 调小 .

这里要特别指出的是 FileInputFormat 会让每个输入文件至少产生一个 map 任务 , 因此如果你的输入目录下有许多文件 , 而每个文件都很小 , 例如几十 kb, 那么每个文件都产生一个 map 会增加调度开销 . 作业变慢 .

那么如何防止这种问题呢 ? CombineFileInputFormat 能有效的减少 map 数量 .

FileInputFormat( 新接口 org.apache.hadoop.mapreduce.lib.input )

Hadoop 0.20 开始定义了一套新的 mapreduce 编程接口 , 使用新的 FileInputFormat, 它与旧接口下的 FileInputFormat 主要区别在于 , 它不再使用 mapred.map.tasks, 而使用 mapred.max.split.size 参数代替 goalSize, 通过 Math.max (minSize, Math.min (maxSize, blockSize)) 决定 map 输入长度 , 一个 map 的输入要大于 minSize, 小于

Math.min (maxSize, blockSize).

    若需增加 map 数 , 可以把 mapred.min.split.size 调小 , 把 mapred.max.split.size 调大 . 若需减少 map 数 , 可以把 mapred.min.split.size 调大 , 并把 mapred.max.split.size 调小 .

 

CombineFileInputFormat

顾名思义 , CombineFileInputFormat 的作用是把许多文件合并作为一个 map 的输入 .

在它之前 , 可以使用 MultiFileInputFormat , 不过其功能太简单 , 它 以文件为单位，一个文件至多分给一个 map处理 , 如果某个目录下有许多小文件 , 另外还有一个超大文件 , 处理大文件的 map会严重偏慢 .

CombineFileInputFormat 是一个被推荐使用的 InputFormat. 它有三个配置 :

mapred.min.split.size.per.node ， 一个节点上 split 的至少的大小

mapred.min.split.size.per.rack   一个交换机下 split 至少的大小

mapred.max.split.size             一个 split 最大的大小

它的主要思路是把输入目录下的大文件分成多个 map 的输入 , 并合并小文件 , 做为一个 map 的输入 . 具体的原理是下述三步 :

1. 根据输入目录下的每个文件 , 如果其长度超过 mapred.max.split.size, 以 block 为单位分成多个 split( 一个 split 是一个 map 的输入 ), 每个 split 的长度都大于 mapred.max.split.size, 因为以 block 为单位 , 因此也会大于 blockSize, 此文件剩下的长度如果大于 mapred.min.split.size.per.node, 则生成一个 split, 否则先暂时保留 .

2. 现在剩下的都是一些长度效短的碎片 , 把每个 rack 下碎片合并 , 只要长度超过 mapred.max.split.size 就合并成一个 split, 最后如果剩下的碎片比 mapred.min.split.size.per.rack 大 , 就合并成一个 split, 否则暂时保留 .

3. 把不同 rack 下的碎片合并 , 只要长度超过 mapred.max.split.size 就合并成一个 split, 剩下的碎片无论长度 , 合并成一个 split.

举例 : mapred.max.split.size=1000

     mapred.min.split.size.per.node=300

      mapred.min.split.size.per.rack=100

输入目录下五个文件 ,rack1 下三个文件 , 长度为 2050,1499,10, rack2 下两个文件 , 长度为 1010,80. 另外 blockSize 为 500.

经过第一步 , 生成五个 split: 1000,1000,1000,499,1000. 剩下的碎片为 rack1 下 :50,10; rack2 下 10:80

由于两个 rack 下的碎片和都不超过 100, 所以经过第二步 , split 和碎片都没有变化 .

第三步 , 合并四个碎片成一个 split, 长度为 150.

 

如果要减少 map 数量 , 可以调大 mapred.max.split.size, 否则调小即可 .

其特点是 : 一个块至多作为一个 map 的输入，一个文件可能有多个块，一个文件可能因为块多分给做为不同 map 的输入， 一个 map 可能处理多个块，可能处理多个文件。

 

 

编写自己的 InputFormat

 

    待续