0

我编写了一个自定义文件阅读器来不拆分我的输入文件，因为它们是大型 gzip 文件，我希望我的第一个映射器工作只是简单地对它们进行压缩。我按照“Hadoop The Definitive Guide”中的示例进行操作，但在尝试读取 BytesWritable 时出现堆错误。我相信这是因为字节数组的大小为 85713669，但我不确定如何克服这个问题。

这是代码：

public class WholeFileRecordReader extends RecordReader<NullWritable, BytesWritable> {

private FileSplit fileSplit;
private Configuration conf;
private BytesWritable value = new BytesWritable();
private boolean processed = false;

@Override
public void close() throws IOException {
    // do nothing
}

@Override
public NullWritable getCurrentKey() throws IOException,
        InterruptedException {
    return NullWritable.get();
}

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

@Override
public float getProgress() throws IOException, InterruptedException {
    return processed ? 1.0f : 0.0f;
}

@Override
public void initialize(InputSplit split, TaskAttemptContext context)
        throws IOException, InterruptedException {
    this.fileSplit = (FileSplit) split;
    this.conf = context.getConfiguration();
}

@Override
public boolean nextKeyValue() throws IOException, InterruptedException {
    if (!processed) {
        byte[] contents = new byte[(int) fileSplit.getLength()];
        Path file = fileSplit.getPath();
        FileSystem fs = file.getFileSystem(conf);
        FSDataInputStream in = null;
        try {
            in = fs.open(file);
            IOUtils.readFully(in, contents, 0, contents.length);
            value.set(contents, 0, contents.length);
        } finally {
            IOUtils.closeStream(in);
        }
        processed = true;
        return true;
    }
    return false;
}

}

4

1 回答 1

1

一般来说，您不能将整个文件加载到 Java VM 的内存中。您应该找到一些流式解决方案来处理大文件 - 逐块读取数据并将结果保存在内存中而不修复整个数据集
这个特定任务 - 解压缩可能不适合 MR，因为没有将数据逻辑划分为记录。
另请注意，hadoop 会自动处理 gzip - 您的输入流将已经解压缩。

于 2013-02-11T13:19:45.367 回答