0

次の RecordReader の問題を解決しようとしています。入力ファイルの例:

1,1
2,2
3,3
4,4
5,5
6,6
7,7
.......
.......

RecordReader を返してほしい

key | Value 
0   |1,1:2,2:3,3:4,4:5,5
4   |2,2:3,3:......6,6
6   |3,3:4,4......6,6,7,7

(最初の値の場合は最初の 5 行、2 番目の値の場合は 2 行目から始まる 5 行、3 番目の値の場合は 3 行目から始まる 5 行など)

public class MyRecordReader extends RecordReader<LongWritable, Text> {

@Override
public boolean nextKeyValue() throws IOException, InterruptedException {

    while (pos < end) {
      key.set(pos);
      // five line logic 
      Text nextLine=new Text();



      int newSize = in.readLine(value, maxLineLength,
                                Math.max((int)Math.min(Integer.MAX_VALUE, end-pos),
                                         maxLineLength));
      fileSeek+=newSize;

      for(int n=0;n<4;n++)
      {
          fileSeek+=in.readLine(nextLine, maxLineLength,
                  Math.max((int)Math.min(Integer.MAX_VALUE, end-pos),
                           maxLineLength));
          value.append(":".getBytes(), 0,1);
          value.append(nextLine.getBytes(), 0, nextLine.getLength());
      }
      if (newSize == 0) {

        return false;

      }
      pos += newSize;
      if (newSize < maxLineLength) {

        return true;
      }

      // line too long. try again
      LOG.info("Skipped line of size " + newSize + " at pos " + (pos - newSize));
    }

    return false;
}

}

しかし、これは値を次のように返しています

key | Value 
0   |1,1:2,2:3,3:4,4:5,5
4   |6,6:7,7.......10,10
6   |11,11:12,12:......14,14

誰かがこのコードで私を助けることができますか、または RecodeReader の新しいコードも同様に行いますか? 問題の要件 (ユースケースを理解するのに役立つ場合があります) ありがとう

4

1 に答える 1

3

私は質問を理解していると思います...これが私がすることです：別のRecordReaderをラップし、そこからキー/値をローカルキューにバッファリングします。

public class MyRecordReader extends RecordReader<LongWritable, Text> {
    private static final int BUFFER_SIZE = 5;
    private static final String DELIMITER = ":";

    private Queue<String> valueBuffer = new LinkedList<String>();
    private Queue<Long> keyBuffer = new LinkedList<Long>();
    private LongWritable key = new LongWritable();
    private Text value = new Text();

    private RecordReader<LongWritable, Text> rr;
    public MyRecordReader(RecordReader<LongWritable, Text> rr) {
        this.rr = rr;
    }

    @Override
    public void close() throws IOException {
        rr.close();
    }

    @Override
    public LongWritable getCurrentKey() throws IOException, InterruptedException {
        return key;
    }

    @Override
    public Text getCurrentValue() throws IOException, InterruptedException {
        return value;
    }

    @Override
    public float getProgress() throws IOException, InterruptedException {
        return rr.getProgress();
    }

    @Override
    public void initialize(InputSplit arg0, TaskAttemptContext arg1)
            throws IOException, InterruptedException {
        rr.initialize(arg0, arg1);
    }

    @Override
    public boolean nextKeyValue() throws IOException, InterruptedException {
        if (valueBuffer.isEmpty()) {
            while (valueBuffer.size() < BUFFER_SIZE) {
                if (rr.nextKeyValue()) {
                    keyBuffer.add(rr.getCurrentKey().get());
                    valueBuffer.add(rr.getCurrentValue().toString());
                } else {
                    return false;
                }
            }
        } else {
            if (rr.nextKeyValue()) {
                keyBuffer.add(rr.getCurrentKey().get());
                valueBuffer.add(rr.getCurrentValue().toString());
                keyBuffer.remove();
                valueBuffer.remove();
            } else {
                return false;
            }
        }
        key.set(keyBuffer.peek());
        value.set(getValue());
        return true;
    }

    private String getValue() {
        StringBuilder sb = new StringBuilder();
        Iterator<String> iter = valueBuffer.iterator();
        while (iter.hasNext()) {
            sb.append(iter.next());
            if (iter.hasNext()) sb.append(DELIMITER);
        }
        return sb.toString();
    }

}

次に、たとえば、次のように、TextInputFormat を拡張し、createRecordReaderメソッドをオーバーライドして、その結果を でラップして呼び出しsuper.createRecordReaderて返すカスタム InputFormat を作成できますMyRecordReader。

public class MyTextInputFormat extends TextInputFormat {
    @Override
    public RecordReader<LongWritable, Text> createRecordReader(
                InputSplit arg0, TaskAttemptContext arg1) {
        return new MyRecordReader(super.createRecordReader(arg0, arg1));
    }
}
于 2012-09-14T23:16:19.380 に答える