首先,考虑这个 CustomWriter 类:
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
调用该方法来关闭编写器。
如您所见,我只创建了一个编写器,它开始附加到一个全新的文件。因此,毫无疑问,这不是原因。
我还应该注意,我目前正在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
吗?