hadoop - 在 Hadoop MapReduce 中将 RecordReader 上下文设置为段落

Question

我想编写自己的 RecordReader，它以整个段落的形式返回上下文，而不是 TextInputFormat 中的一行。

我尝试了以下功能，但绝对是它的方式

public boolean nextKeyValue() throws IOException, InterruptedException {
    if (key == null) {
        key = new LongWritable();
    }
    key.set(pos);
    if (value == null) {
        value = new Text();
    }
    value.clear();
    final Text endline = new Text("\n");
    int newSize = 0;

        Text v = new Text();
        while (v!= endline) {
            value.append(v.getBytes(),0, v.getLength());
            value.append(endline.getBytes(),0, endline.getLength());
            if (newSize == 0) {
                break;
            }
            pos += newSize;
            if (newSize < maxLineLength) {
                break;
            }
        }
    if (newSize == 0) {
        key = null;
        value = null;
        return false;
    } else {
        return true;
    }
}

Answer 1

您实际上不需要努力编写自己的 RecordReader。相反，只需扩展 TextInputFormat 并更改分隔符。以下是 TextInputFormat 的库代码，仅更改了分隔符：

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;

import com.google.common.base.Charsets;

public class ParagraphInputFormat
extends TextInputFormat {
    private static final String PARAGRAPH_DELIMITER = "\r\n\r\n";

    @Override
    protected boolean isSplitable(JobContext context, Path file) {
        return false;
    }

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