MapReduce-定制Partitioner-使用NLineInputFormat处理大文件-求文件奇偶数行之和

在上一篇《MapReduce-定制Partitioner-求文件奇偶数行之和》博客中有朋友提到“如果文件很大，就被分成了多个record，那么每个record中的文件的奇数和偶数相对于原来的文件来说，就不确定了”这样一个问题，这一篇文章就对这种情况的处理进行说明一下，解决的思路如下：

我们只要固定每一个inputSplit的行数，我们就可以确定某一个inputSplit的某一行在整个文件中是奇数行还是偶数行。而且mapreduce框架已经为我们提供了NLineInputFormat来处理这一个问题，我们只用在启动程序中加入两行代码就能解决这个问题。

处理数据：

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

这里我们假设数据很大被分成了两个split,我们的启动程序如下：
其于的map端代码，reduce端代码完全以及partition的代码没有任何改动，参见《MapReduce-定制Partitioner-求文件奇偶数行之和》

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.NLineInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class JobMain {
public static void main(String[] args) throws Exception {
Configuration configuration = new Configuration();
/**
* 对比前一篇博客的启动程序多了以下设置每个mapper的行数以及设置InputFormat的代码
*/
configuration.setInt("mapreduce.input.lineinputformat.linespermap", 10);
Job job = new Job(configuration, "partitioner-job");
job.setInputFormatClass(NLineInputFormat.class);
job.setJarByClass(JobMain.class);
job.setMapperClass(MyMapper.class);
job.setMapOutputKeyClass(LongWritable.class);
job.setMapOutputValueClass(IntWritable.class);
job.setPartitionerClass(MyPartitioner.class);
job.setReducerClass(MyReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
job.setNumReduceTasks(2);
FileInputFormat.addInputPath(job, new Path(args[0]));
Path outputDir = new Path(args[1]);
FileSystem fs = FileSystem.get(configuration);
if( fs.exists(outputDir)) {
fs.delete(outputDir ,true);
}
FileOutputFormat.setOutputPath(job, outputDir);
System.exit(job.waitForCompletion(true) ? 0: 1);
}
}

运行过程：



从红色框中可以看出测试数据被分成两个split用两个不同的map进行计算。

运行结果：



总结：
这里指定的是每个inputSplit为10行，具体应用时应该根据行的大小以及hdfs的block的大小来设置行数，以使各个map进程能够跑在不同的机器上以提高集群资源的利用率。