NLineInputFormat 应用

hadoop数据块默认是64MB，也就是说1G的文件会被split成16块， 分发到16个map任务。

现在假设有一个命令行文本文件，每N行作为一个命令单元，总共有M行(M%N=0)，把这个命令行文件作为hadoop作业的输入, 要求hadoop按每N行进行一次split,最后将M/N个命令单元分发到各个map任务, 要实现这样的功能就需要用到NLineInputFormat类, 将作业的输入类设置为NLineInputFormat, 再将作业属性设置成-Dmapred.line.input.format.linespermap=N, 就可以按N行做一次split了

来分析下NLineInputFormat源码

初始化部分:

public void configure(JobConf conf) {
N = conf.getInt("mapred.line.input.format.linespermap", 1); // 获取参数配置， 默认为1, 也就是每一行对应一个map
}

split部分:

if (numLines == N) {
splits.add(new FileSplit(fileName, begin, length, new String[]{}));
begin += length;
length = 0;
numLines = 0;
}

下面附上一个真实的用例:

目前公司采用hadoop分析日志文件，每小时都会从11台日志服务器上获取日志文件到 hdfs文件系统。

线上总共11台日志服务器，每隔一个小时生成一个日志文件

目录路径为: /..../logs/yyyy-MM-dd/hour/logfile.log

日志获取作业每一个小时执行一次， 将当前小时的前一小时日志从11台服务器拉取到HDFS文件系统, 为了增加获取日志的并行度，在日志获取作业执行前会动态生成一个获取日志文件的命令行配置文件

配置文件大概为这样:

wget host1/logs/yyyy-MM-dd/hour/logfile.log  //wget日志文件
hadoop put ...                               // put到hdfswe
wget host2/logs/yyyy-MM-dd/hour/logfile.log
hadoop put ...
...
...
...
wget host11/logs/yyyy-MM-dd/hour/logfile.log
hadoop put ...

为了达到并行， 需要将命令行文件每两行做一次split, 对一台日志服务器的获取任务分配到一个map, 这样需设置作业参数
-Dmapred.line.input.format.linespermap=2即可

最后， 上面这个例子只说明NLineInputFormat类的使用场景， 与这个日志获取方案是否好无关，就目前来说， 公司日志分析刚刚起步，每小时日志文件大小在3G左右，通过wget获取倒也凑合，但这个方案有明显不足， 日志是定时一次获取的， 随着日志量的日益增大， 实时性和效率都得不到保证，目前正研究可应用于日志收集的开源系统，如 chukwa, kafka等， 以期能实现一个符合公司要求的数据收集方案!