为什么 SequenceFile writer 的附加操作会用最后一个值覆盖所有值？

Question

首先，考虑这个 CustomWriter class:

public final class CustomWriter {

  private final SequenceFile.Writer writer;

  CustomWriter(Configuration configuration, Path outputPath) throws IOException {
    FileSystem fileSystem = FileSystem.get(configuration);
    if (fileSystem.exists(outputPath)) {
      fileSystem.delete(outputPath, true);
    }

    writer = SequenceFile.createWriter(configuration,
        SequenceFile.Writer.file(outputPath),
        SequenceFile.Writer.keyClass(LongWritable.class),
        SequenceFile.Writer.valueClass(ItemWritable.class),
        SequenceFile.Writer.compression(SequenceFile.CompressionType.BLOCK, new DefaultCodec()),
        SequenceFile.Writer.blockSize(1024 * 1024),
        SequenceFile.Writer.bufferSize(fileSystem.getConf().getInt("io.file.buffer.size", 4 * 1024)),
        SequenceFile.Writer.replication(fileSystem.getDefaultReplication(outputPath)),
        SequenceFile.Writer.metadata(new SequenceFile.Metadata()));
  }

  public void close() throws IOException {
    writer.close();
  }

  public void write(Item item) throws IOException {
    writer.append(new LongWritable(item.getId()), new ItemWritable(item));
  }
}

我想做的是使用 Item 类型对象的异步流。消费者有一个对 CustomWriter 实例的引用。然后，它会为收到的每个项目调用 CustomWriter#write 方法。当流结束时，调用 CustomWriter#close 方法关闭写入器。

如您所见，我只创建了一个编写器，它开始附加到一个全新的文件。所以，毫无疑问 this 不是原因。

我还应该注意，我目前运行按照说明 here 使用 MiniDFSCluster 在单元测试环境中进行此操作。如果我运行在非单元测试环境中（即没有 MiniDFSCluster），它似乎工作得很好。

当我尝试读回文件时，我看到的只是最后写入的 Item 对象 N 次（其中 N 是流中接收到的项目总数）。这是一个例子：

sparkContext.hadoopFile(path, SequenceFileInputFormat.class, LongWritable.class, ItemWritable.class)
    .collect()
    .forEach(new BiConsumer<>() {
      @Override
      public void accept(Tuple2<LongWritable, ItemWritable> tuple) {
        LongWritable id = tuple._1();
        ItemWritable item = tuple._2();
        System.out.print(id.get() + " -> " + item.get());
      }
    });

这将打印如下内容：

...
1234 -> Item[...]
1234 -> Item[...]
1234 -> Item[...]
...

我是不是做错了什么，或者这是使用 MiniDFSCluster 的副作用吗？

Answer 1

Writable（如LongWritable, ItemWritable）在处理数据时重复使用。当接收到一条记录时，Writable 通常只是替换它的内容，你将只接收到相同的 Writable 对象。如果你想将它们收集到一个数组中，你应该将它们复制到一个新对象中。

为什么 SequenceFile writer 的附加操作会用最后一个值覆盖所有值？

Why does SequenceFile writer's append operation overwrites all values with the last value?

java

hadoop

apache-spark

hdfs

sequencefile