MapReduce之RecordReader组件源码解析及实例，mapreduce实例

MapReduce之RecordReader组件源码解析及实例，mapreduce实例

简述

无论我们以怎样的方式从分片中读取一条记录，每读取一条记录都会调用RecordReader类；
系统默认的RecordReader是LineRecordReader，TextInputFormat；
LineRecordReader是用每行的偏移量作为map的key，每行的内容作为map的value；
而SequenceFileInputFormat的RecordReader是SequenceFileRecordReader；
应用场景：自定义读取每一条记录的方式；自定义读入key的类型，如希望读取的key是文件的路径或名字而不是该行在文件中的偏移量。

TextInputFormat源码如下：

package org.apache.hadoop.mapreduce.lib.input;
/** An {@link InputFormat} for plain text files.  Files are broken into lines.
 * Either linefeed or carriage-return are used to signal end of line.  Keys are
 * the position in the file, and values are the line of text.. */
public class TextInputFormat extends FileInputFormat<LongWritable, Text> {

  @Override
  public RecordReader<LongWritable, Text> 
    createRecordReader(InputSplit split,
                       TaskAttemptContext context) {
    String delimiter = context.getConfiguration().get(
        "textinputformat.record.delimiter");
    byte[] recordDelimiterBytes = null;
    if (null != delimiter)
      recordDelimiterBytes = delimiter.getBytes(Charsets.UTF_8);
    return new LineRecordReader(recordDelimiterBytes);
  }

  @Override
  protected boolean isSplitable(JobContext context, Path file) {
    final CompressionCodec codec =
      new CompressionCodecFactory(context.getConfiguration()).getCodec(file);
    if (null == codec) {
      return true;
    }
    return codec instanceof SplittableCompressionCodec;
  }
}

textinputformat.record.delimiter指的是读取一行的数据的终止符号，即遇到textinputformat.record.delimiter所包含的字符时，该一行的读取结束。
可以通过Configuration的set()方法来设置自定义的终止符，如果没有设置 textinputformat.record.delimiter，那么Hadoop就采用以CR，LF或者CRLF作为终止符，这一点可以查看LineReader的readDefaultLine方法 。

LineRecordReader源码如下：

package org.apache.hadoop.mapreduce.lib.input;

/**
 * Treats keys as offset in file and value as line. 
 */
public class LineRecordReader extends RecordReader<LongWritable, Text> {
     public void initialize(InputSplit genericSplit,
                         TaskAttemptContext context) throws IOException {
        ......
        start = split.getStart();
        end = start + split.getLength();
        final Path file = split.getPath();

        // open the file and seek to the start of the split
        final FileSystem fs = file.getFileSystem(job);
        fileIn = fs.open(file);
        ......
        // If this is not the first split, we always throw away first record
        // because we always (except the last split) read one extra line in
        // next() method.
        if (start != 0) {
            start += in.readLine(new Text(), 0, maxBytesToConsume(start));
        }
        this.pos = start;
        ......
    }
    ......
}

自定义RecordReader

1、继承抽象类RecordReader，实现RecordReader的一个实例；
2、实现自定义InputFormat类，重写InputFormat中createRecordReader()方法，返回值是自定义的RecordReader实例；
3、配置job.setInputFormatClass()设置自定义的InputFormat实例；

实例

数据：

10
20
30
40
50
60
70
……

要求：读取整个文件，分别计算奇数行与偶数行数据之和
奇数行之和：10+30+50+70=160
偶数行之和：20+40+60=120

package Recordreader;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
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.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.LineReader;

public class MyRecordReader {

    private final static String INPUT_PATH = "hdfs://liguodong:8020/inputsum";
    private final static String OUTPUT_PATH = "hdfs://liguodong:8020/outputsum";    

    public static class DefRecordReader extends