0

我正在使用以下代码使用 XMLStreamReader 在 hadoop RecordReader 中读取大型 xml 文件（以 GB 为单位）

public class RecordReader {
   int progressCouunt = 0;
   public RecordReader() {
    XMLInputFactory factory = XMLInputFactory.newInstance();
    FSDataInputStream fdDataInputStream = fs.open(file); //hdfs file
    try {
          reader = factory.createXMLStreamReader(fdDataInputStream);
    } catch (XMLStreamException exception) {
           throw new RuntimeException("XMLStreamException exception : ", exception);
    }
   }
   @Override
  public float getProgress() throws IOException, InterruptedException {
     return progressCouunt; 
   }
}

我的问题是如何使用 XMLStreamReader 获取文件的读取进度，因为它不提供任何开始或结束位置来计算进度百分比。我参考了如何在 StAX 中跟踪大文件的解析进度？，但不能用户 filterReader。请在这里帮助我。

4

1 回答 1

2

InputStream您可以通过扩展来包装FilterInputStream.

public interface InputStreamListener {
    void onBytesRead(long totalBytes);
}

public class PublishingInputStream extends FilterInputStream {
    private final InputStreamListener;
    private long totalBytes = 0;

    public PublishingInputStream(InputStream in, InputStreamListener listener) {
       super(in);
       this.listener = listener;
    }

    @Override
    public int read(byte[] b) {
       int count = super.read(b);
       this.totalBytes += count;
       this.listener.onBytesRead(totalBytes);
    }

    // TODO: override the other read() methods
}

用法

XMLInputFactory factory = XMLInputFactory.newInstance();
InputStream in = fs.open(file);
final long fileSize = someHadoopService.getFileLength(file);
InputStremListener listener = new InputStreamListener() {
    public void onBytesRead(long totalBytes) {
        System.out.println(String.format("Read %s of %s bytes", totalBytes, fileSize));
    }
};
InputStream publishingIn = new PublishingInputStream(in, listener);
try {
    reader = factory.createXMLStreamReader(publishingIn);
    // etc
于 2016-06-10T14:57:12.273 回答