Hadoop自定义记录读取器实现

Question

我很难理解下面链接中解释的 nextKeyValue() 方法中发生的事情的流程：

http://analyticspro.org/2012/08/01/wordcount-with-custom-record-reader-of-textinputformat/

尤其是 nextKeyValue() 中的 for 循环

任何帮助都将是可观的

提前致谢

Answer 1

nextKeyValue() 是为特定映射调用设置键值对的核心函数。因此，从您的链接中，下面的代码（在 for 循环之前）它只是使用 pos 设置键，这不过是起始偏移量key.set(pos) 并且它缓冲了先前设置的值。对应代码：

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;

for 循环之后。我为每一行添加了足够的 cmets。

       for(int i=0;i<NLINESTOPROCESS;i++){ //Since this is NLineInputFormat they want to read 3 lines at a time and set that as value,
so this loop will continue until that is satisfied.
            Text v = new Text();
            while (pos < end) { //This is to prevent the recordreader from reading the second split, if it is currently reading the first split. pos would be start
of the split and end would be end offset of the split. 
                newSize = in.readLine(v, maxLineLength,Math.max((int)Math.min(Integer.MAX_VALUE, end-pos),maxLineLength));
//This calls the linereader readline function which reads until it encounters a newline(default delim for TextInputformat and maxlinelength would be max integer size
just to ensure the whole line doesn''t go beyond the maxlinelength and the line read would be stored in Text variable v)
                value.append(v.getBytes(),0, v.getLength());
//Reads from v(whole line) and appends it to the value,append is necessary because we are going to read 3 lines.
                value.append(endline.getBytes(),0, endline.getLength());
//appends newline to each line read
                if (newSize == 0) {
                    break;//If there is nothing to read then come out.
                }
                pos += newSize;
                if (newSize < maxLineLength) {//There is a flaw here it should be >=, to imply if the read line is greater than max integer size then come out
                    break;
                }
            }
        }
        if (newSize == 0) {
            key = null;//If there is nothing to read assign key and value as null else continue the process by returning true to map call.
            value = null;
            return false;
        } else {
            return true;
        }
    }
}

Answer 2

每个映射器将使用方法 nextKeyValue() 在所有拆分记录之间进行迭代。

NLinesRecordReader 类定义每条记录有 3 行。

private final int NLINESTOPROCESS = 3;

nextKeyValue() 中循环的主要作用是为每条记录获取 3 行。该记录将用作 map() 方法的输入值。

Answer 3

每当需要新数据时，都会发生两件事。向读者提出的第一个问题是

你有任何数据吗？？？

如果读者回答是，那么调用者可以从 getCurrentValue 方法中获取数据。

现在 nextKeyValue 方法完成了这个任务，它只是回答了你有任何数据要给我吗？

由于防火墙问题，我无法访问该链接，但我使用的示例实现是

HashMap<Integer, Invoice> allData= new HashMap<Integer, Invoice>();

    @Override
public boolean nextKeyValue() throws IOException, InterruptedException {
    if(key == null) {
        this.key = new LongWritable();
    }
    this.key.set(startPos);

    if(value == null) {
        this.value = new Invoice();
    }
    if(startPos >= endPos) {
        key = null;
        value = null;
        return false;
    } else {
        this.value = allData.get(startPos);
        startPos = startPos + 1;
        return true;
    }
}

这里发票只是一个 POJO。在初始化方法中，我什么也没做，只是解析整个文档并存储在哈希图中。 在 nextKeyValue 方法中检查下一个键是否存在，如果它确实返回相应的值，否则返回该键不存在。