如何通过Java代码获取HDFS写入时连接的NameNode及复制确认

一、获取当前连接的HA NameNode信息 #

在HA模式下，HDFS客户端会自动路由到Active NameNode。要在代码中获取当前连接的节点信息，可以通过FileSystem实例的API或配置信息实现：

代码示例 #

// 假设fs是已初始化的FileSystem实例
// 1. 获取当前Active NameNode的服务标识/RPC地址
String activeNnServiceName = fs.getCanonicalServiceName();
System.out.println("当前连接的Active NameNode: " + activeNnServiceName);

// 2. 获取集群中所有NameNode的配置信息
Configuration conf = fs.getConf();
String nameService = conf.get("dfs.nameservices");
String[] nnIds = conf.getTrimmedStrings("dfs.ha.namenodes." + nameService);

for (String nnId : nnIds) {
    // 获取RPC地址
    String rpcAddress = conf.get("dfs.namenode.rpc-address." + nameService + "." + nnId);
    // 获取状态查询地址
    String stateUrl = conf.get("dfs.namenode.http-address." + nameService + "." + nnId + ".state");
    System.out.printf("NameNode %s: RPC地址=%s, 状态查询地址=%s%n", nnId, rpcAddress, stateUrl);
}

说明 #

• getCanonicalServiceName() 返回当前客户端连接的Active NameNode的服务标识（如nn1）或完整RPC地址，具体取决于集群配置。
• 通过Configuration可以批量获取HA集群中所有NameNode的配置项，方便排查多节点状态。

二、接收数据复制成功的确认信息 #

HDFS写入时，客户端默认会等待数据块完成副本复制后才会确认，你可以通过以下方式显式获取复制成功的信号：

1. 使用sync()方法强制等待副本完成 #

sync()方法会阻塞直到当前写入的数据被复制到所有指定副本节点，成功才返回，失败则抛出IOException：

FSDataOutputStream fsDataOutputStream = fs.append(hdfsWritePath);
BufferedWriter bufferedWriter = new BufferedWriter(new OutputStreamWriter(fsDataOutputStream, StandardCharsets.UTF_8));

for (int i=0;i<1000;i++) {
    bufferedWriter.write("append" + i);
    bufferedWriter.newLine();
    System.out.println("wrote : "+i);
    Thread.sleep(100);
}

// 全部写入完成后，调用sync确保所有数据完成副本复制
fsDataOutputStream.sync();
System.out.println("所有数据已完成副本复制确认");

bufferedWriter.flush();
bufferedWriter.close();
fs.close();

2. 通过close()方法自动确认 #

close()方法会自动触发sync()操作，确保所有未刷新的数据都被写入并完成副本复制，关闭流成功则代表所有数据已持久化：

// ... 写入逻辑 ...

bufferedWriter.flush();
bufferedWriter.close(); // 关闭时自动等待所有副本完成
System.out.println("文件关闭成功，所有数据已完成集群副本复制");
fs.close();

3. 检查文件副本数（事后验证） #

如果需要确认最终文件的副本数是否符合预期，可以在写入完成后查询文件状态：

// 写入完成后获取文件状态
FileStatus fileStatus = fs.getFileStatus(hdfsWritePath);
short actualReplication = fileStatus.getReplication();
short expectedReplication = fs.getConf().getShort("dfs.replication", (short)3);

if (actualReplication == expectedReplication) {
    System.out.printf("文件已达到预期副本数: %d%n", expectedReplication);
} else {
    System.out.printf("当前副本数: %d，未达到预期的%d%n", actualReplication, expectedReplication);
}

内容的提问来源于stack exchange，提问作者Abhishek Karigar