Sample HDFS HA Client

In any HDP cluster with a HA setup with quorum there are two NameNodes configured with one working as the active and the other as the standby instance. As the standby node does not accept any write requests, for a client try to write to HDFS it is fairly important to know which one of the two NameNodes it the active one at any given time. The discovery process for that is configured through the hdfs-site.xml.

For any custom implementation it’s becomes relevant to set and understand the correct parameters if a current hdfs-site.xml configuration of the cluster is not given. This post gives a sample Java implementation of a HA HDFS client.

The failover mechanism of a client is a plug-able implementation configurable by the dfs.client.failover.proxy.provider.{{serviceId}} property. This is set for each service identify, so be using hdfs://serviceId the client will lookup the appropriate fail-over implementation for that service.

A service id should therefor also not contain the port 8020 as it would in a non HA setup.

Typically the ConfiguredFailoverProxyProvider (link) is being used, which expects in the the hdfs-site.xml a proper configuration to access both NameNodes as illustrated in the picture below. The fail-over typically happens based on a trial and error approach, where the first NN’s state is being validated before the second.

Programmatically using a HA HDFS file system in Java is to configure the defaultFS property with the service id of the clsuter:

Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://" + nameserviceId);

The client needs to know it is supposed to use the fail-over mechanism by setting the proxy provider:

conf.set("dfs.client.failover.proxy.provider."+ nameserviceId, 
        "org.apache.hadoop.hdfs.server.namenode.
ha.ConfiguredFailoverProxyProvider");

For the HA fail-over proxy to work we also need to configure the whole HA setup into the client:

conf.set("dfs.nameservices", nameserviceId);
conf.set("dfs.ha.namenodes."+ nameserviceId, "nn01,nn02");
conf.set("dfs.namenode.rpc-address."+ nameserviceId +".nn01", getProperty("nn1.rpc-address"));
conf.set("dfs.namenode.rpc-address."+ nameserviceId +".nn02", getProperty("nn2.rpc-address"));
conf.set("dfs.namenode.http-address."+ app.nameserviceId +".nn01", getProperty("nn1.http-address"));
conf.set("dfs.namenode.http-address."+ app.nameserviceId +".nn02", getProperty("nn2.http-address"));

The configuration design in an overview:

With this configuration we can now use the FS to list all files under the user directory:

FileSystem fs = FileSystem.get(conf);
String pathStr = "/user/" + System.getProperty("user.name");
System.out.println("Listing files in: " + pathStr);
FileStatus[] fsStatus = fs.listStatus(new Path(pathStr));

for (int i = 0; i < fsStatus.length; i++) {
  System.out.println(fsStatus[i].getPath().toString());
}

Here is the complete Java implementation:

Configuration conf = new Configuration();
conf.set("fs.defaultFS", "hdfs://" + nameserviceId);
            
            
conf.set("dfs.client.failover.proxy.provider."+ nameserviceId, 
        "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

conf.set("dfs.nameservices", nameserviceId);
conf.set("dfs.ha.namenodes."+ nameserviceId, "nn01,nn02");
conf.set("dfs.namenode.rpc-address."+ nameserviceId +".nn01", getProperty("nn1.rpc-address"));
conf.set("dfs.namenode.rpc-address."+ nameserviceId +".nn02", getProperty("nn2.rpc-address"));
conf.set("dfs.namenode.http-address."+ app.nameserviceId +".nn01", getProperty("nn1.http-address"));
conf.set("dfs.namenode.http-address."+ app.nameserviceId +".nn02", getProperty("nn2.http-address"));
            

FileSystem fs = FileSystem.get(conf);
String pathStr = "/user/" + System.getProperty("user.name");
System.out.println("Listing files in: " + pathStr);
FileStatus[] fsStatus = fs.listStatus(new Path(pathStr));

for (int i = 0; i < fsStatus.length; i++) {
  System.out.println(fsStatus[i].getPath().toString());
}

The complete configuration in hdfs-site.xml for a sample HA setup:

<configuration>
  
  <!-- services -->
  <property>
    <name>dfs.nameservices</name>  
    <value>serviceId1</value>
  </property>
  
  <!--  services NameNodes --> 
  <property> 
    <name>dfs.ha.namenodes.serviceId2</name>    
    <value>nn201,nn202</value> 
  </property>

  <!-- NameNode 01 --> 
  <property> 
    <name>dfs.namenode.rpc-address.serviceId2.nn01</name> 
    <value>nn01.hdp:8020</value>
  </property> 
  <property> 
    <name>dfs.namenode.servicerpc-address.serviceId2.nn01</name> 
    <value>nn01.hdp:54321</value> 
  </property>
  <property>
    <name>dfs.namenode.http-address.serviceId2.nn01</name> 
    <value>nn01.hdp:50070</value>
  </property>
  <property> 
    <name>dfs.namenode.https-address.serviceId2.nn01</name> 
    <value>nn01.hdp:50470</value>
  </property>

  <!-- NameNode 02 -->
  <property>
    <name>dfs.namenode.rpc-address.serviceId2.nn02</name> 
    <value>nn02.hdp:8020</value>
  </property>
  <property>
    <name>dfs.namenode.servicerpc-address.serviceId2.nn02</name> 
    <value>nn02.hdp:54321</value>
  </property>
  <property>
    <name>dfs.namenode.http-address.serviceId2.nn02</name>
    <value>nn02.hdp:50070</value>
  </property>
  <property> 
    <name>dfs.namenode.https-address.serviceId2.nn02</name>  
    <value>nn02.hdp:50470</value>
  </property>

  <!-- HA service -->
  <property>   
    <name>dfs.client.failover.proxy.provider.nameservices2</name> 
    <value>org.apache.hadoop.hdfs.server  
               .namenode.ha.ConfiguredFailoverProxyProvider</value>
  </property>
</configuration>

Further Reading

• HDFS High Availability Using the Quorum Journal Manager
• HDFS High Availability with NFS
• A Secure HDFS Client Example
• Distcp between two HA Cluster
• HDFS Namenodes and Datanodes