尽管有关于此主题的所有资源,但我在刷新磁盘上的hdfs文件时仍然遇到问题(hadoop 2.6)
调用FSDataOutputStream.hsync()
应该可以解决问题,但实际上由于未知原因它只能运行一次。
这是一个失败的简单单元测试:
@Test
public void test() throws InterruptedException, IOException {
final FileSystem filesys = HdfsTools.getFileSystem();
final Path file = new Path("myHdfsFile");
try (final FSDataOutputStream stream = filesys.create(file)) {
Assert.assertEquals(0, getSize(filesys, file));
stream.writeBytes("0123456789");
stream.hsync();
stream.hflush();
stream.flush();
Thread.sleep(100);
Assert.assertEquals(10, getSize(filesys, file)); // Works
stream.writeBytes("0123456789");
stream.hsync();
stream.hflush();
stream.flush();
Thread.sleep(100);
Assert.assertEquals(20, getSize(filesys, file)); // Fails, still 10
}
Assert.assertEquals(20, getSize(filesys, file)); // works
}
private long getSize(FileSystem filesys, Path file) throws IOException {
return filesys.getFileStatus(file).getLen();
}
知道为什么吗?
最佳答案
实际上,hsync()
在内部调用不带标志的私有(private)flushOrSync(boolean isSync, EnumSet<SyncFlag> syncFlags)
,如果提供了SyncFlag.UPDATE_LENGTH
,则长度仅在namenode上更新。
在上述测试中,将getSize()
替换为实际读取文件的代码即可。
private long getSize(FileSystem filesys, Path file) throws IOException {
long length = 0;
try (final FSDataInputStream input = filesys.open(file)) {
while (input.read() >= 0) {
length++;
}
}
return length;
}
要更新大小,您可以选择调用(不进行正确的类类型检查):
((DFSOutputStream) stream.getWrappedStream())).hsync(EnumSet.of(SyncFlag.UPDATE_LENGTH));