hadoop map reduce录音阅读器

eoigrqb6 于 2021-06-03 发布在 Hadoop

关注(0)|答案(1)|浏览(365)

我正在尝试解决以下recordreader问题。输入文件示例：

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

我要我的读卡器回来

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

（对于第一个值第一个五行，对于第二个值从第二行开始五行，对于第三个值从第三行开始五行，依此类推） 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

有人能帮我用这个代码或一个新的代码重编机也可以吗？问题的要求（可能帮助您理解用例）谢谢

Java hadoop mapreduce

来源：https://stackoverflow.com/questions/12418847/hadoop-map-reduce-recordreader

1条答案

按热度按时间

kh212irz1#

我想我理解这个问题。。。我要做的是： Package 另一个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();
    }

}

例如，您可以有一个自定义inputformat来扩展textinputformat并覆盖 createRecordReader 要调用的方法 super.createRecordReader 并返回一个 MyRecordReader ，如下所示：

public class MyTextInputFormat extends TextInputFormat {
    @Override
    public RecordReader<LongWritable, Text> createRecordReader(
                InputSplit arg0, TaskAttemptContext arg1) {
        return new MyRecordReader(super.createRecordReader(arg0, arg1));
    }
}

赞(0）回复(0）举报 2021-06-03

我来回答

hadoop map reduce录音阅读器

1条答案

相关问题

热门标签

最新问答