一般来说，基于Hadoop的MapReduce框架来处理数据，主要是面向海量大数据，对于这类数据，Hadoop能够使其真正发挥其能力。对于海量小文件，不是说不能使用Hadoop来处理，只不过直接进行处理效率不会高，而且海量的小文件对于HDFS的架构设计来说，会占用NameNode大量的内存来保存文件的元数据（Bookkeeping）。另外，由于文件比较小，我们是指远远小于HDFS默认Block大小（64M），比如1k~2M，都很小了，在进行运算的时候，可能无法最大限度地充分Locality特性带来的优势，导致大量的数据在集群中传输，开销很大。
但是，实际应用中，也存在类似的场景，海量的小文件的处理需求也大量存在。那么，我们在使用Hadoop进行计算的时候，需要考虑将小数据转换成大数据，比如通过合并压缩等方法，可以使其在一定程度上，能够提高使用Hadoop集群计算方式的适应性。Hadoop也内置了一些解决方法，而且提供的API，可以很方便地实现。
下面，我们通过自定义InputFormat和RecordReader来实现对海量小文件的并行处理。
基本思路描述如下：
在Mapper中将小文件合并，输出结果的文件中每行由两部分组成，一部分是小文件名称，另一部分是该小文件的内容。

编程实现

我们实现一个WholeFileInputFormat，用来控制Mapper的输入规格，其中对于输入过程中处理文本行的读取使用的是自定义的WholeFileRecordReader。当Map任务执行完成后，我们直接将Map的输出原样输出到HDFS中，使用了一个最简单的IdentityReducer。
现在，看一下我们需要实现哪些内容：

1. 读取每个小文件内容的WholeFileRecordReader
2. 定义输入小文件的规格描述WholeFileInputFormat
3. 用来合并小文件的Mapper实现WholeSmallfilesMapper
4. 输出合并后的文件Reducer实现IdentityReducer
5. 配置运行将多个小文件合并成一个大文件

接下来，详细描述上面的几点内容。

• WholeFileRecordReader类

输入的键值对类型，对小文件，每个文件对应一个InputSplit，我们读取这个InputSplit实际上就是具有一个Block的整个文件的内容，将整个文件的内容读取到BytesWritable，也就是一个字节数组。

实现RecordReader接口，最核心的就是处理好迭代多行文本的内容的逻辑，每次迭代通过调用nextKeyValue()方法来判断是否还有可读的文本行，直接设置当前的Key和Value，分别在方法getCurrentKey()和getCurrentValue()中返回对应的值。
另外，我们设置了”map.input.file”的值是文件名称，以便在Map任务中取出并将文件名称作为键写入到输出。

• WholeFileInputFormat类

这个类实现比较简单，继承自FileInputFormat后需要实现createRecordReader()方法，返回用来读文件记录的RecordReader，直接使用前面实现的WholeFileRecordReader创建一个实例，然后调用initialize()方法进行初始化。

• WholeSmallfilesMapper

• IdentityReducer类

这个是Reduce任务的实现，只是将Map任务的输出原样写入到HDFS中。

• WholeCombinedSmallfiles

这是是程序的入口，主要是对MapReduce任务进行配置，只需要设置好对应的配置即可。我们设置了5个Reduce任务，最终会有5个输出结果文件。
这里，我们的Reduce任务执行的输出格式为SequenceFileOutputFormat定义的，就是SequenceFile，二进制文件。

运行程序

• 准备工作

• 运行MapReduce程序

• 验证程序运行结果

可以看到，Reducer阶段生成了5个文件，他们都是将小文件合并后的得到的大文件，如果需要对这些文件进行其他处理，可以再实现满足实际处理的Mapper，将输入路径指定的前面Reducer的输出路径即可。这样一来，对于大量小文件的处理，转换成了数个大文件的处理，就能够充分利用Hadoop MapReduce计算集群的优势。