转自：http://www.linuxidc.com/Linux/2012-04/57831.htm
系统默认的LineRecordReader是按照每行的偏移量做为map输出时的key值，每行的内容作为map的value值，默认的分隔符是回车和换行。
现在要更改map对应的输入的<key,value>值，key对应的文件的路径（或者是文件名），value对应的是文件的内容（content）。
那么我们需要重写InputFormat和RecordReader，因为RecordReader是在InputFormat中调用的，当然重写RecordReader才是重点！
下面看代码InputFormat的重写：
- public class chDicInputFormat extends FileInputFormat<Text,Text>
- implements JobConfigurable{
- private CompressionCodecFactory compressionCodecs = null;
- public void configure(JobConf conf) {
- compressionCodecs = new CompressionCodecFactory(conf);
- }
-
- protected boolean isSplitable(FileSystem fs, Path file) {
-
- return false;
- }
-
- public RecordReader<Text,Text> getRecordReader(InputSplit genericSplit,
- JobConf job, Reporter reporter) throws IOException{
- reporter.setStatus(genericSplit.toString());
- return new chDicRecordReader(job,(FileSplit)genericSplit);
- }
-
- }
下面来看RecordReader的重写：
- public class chDicRecordReader implements RecordReader<Text,Text> {
- private static final Log LOG = LogFactory.getLog(chDicRecordReader.class.getName());
- private CompressionCodecFactory compressionCodecs = null;
- private long start;
- private long pos;
- private long end;
- private byte[] buffer;
- private String keyName;
- private FSDataInputStream fileIn;
-
- public chDicRecordReader(Configuration job,FileSplit split) throws IOException{
- start = split.getStart();
- end = split.getLength() + start;
- final Path path = split.getPath();
- keyName = path.toString();
- LOG.info("filename in hdfs is : " + keyName);
- final FileSystem fs = path.getFileSystem(job);
- fileIn = fs.open(path);
- fileIn.seek(start);
- buffer = new byte[(int)(end - start)];
- this.pos = start;
-
- }
-
- public Text createKey() {
- return new Text();
- }
-
- public Text createValue() {
- return new Text();
- }
-
- public long getPos() throws IOException{
- return pos;
- }
-
- public float getProgress() {
- if (start == end) {
- return 0.0f;
- } else {
- return Math.min(1.0f, (pos - start) / (float)(end - start));
- }
- }
-
- public boolean next(Text key, Text value) throws IOException{
- while(pos < end) {
- key.set(keyName);
- value.clear();
- fileIn.readFully(pos,buffer);
- value.set(buffer);
-
- pos += buffer.length;
- LOG.info("end is : " + end + " pos is : " + pos);
- return true;
- }
- return false;
- }
-
- public void close() throws IOException{
- if(fileIn != null) {
- fileIn.close();
- }
-
- }
-
- }
通过上面的代码，然后再在main函数中设置InputFormat对应的类，就可以使用这种新的读入格式了。
【转载】Hadoop自定义RecordReader
原文：http://www.cnblogs.com/YangtzeYu/p/6271211.html
