Configure HDFS high availability

Author: blootsvoets

src/main/java/org/radarcns/hdfs/RestructureAvroRecords.java

@@ -109,6 +109,26 @@ public static void main(String [] args) {
                 .doDeduplicate(commandLineArgs.deduplicate).format(commandLineArgs.format)
                 .doStage(!commandLineArgs.noStage);
 
+        // Configure high availability
+        if (commandLineArgs.hdfsUri1 != null || commandLineArgs.hdfsUri2 != null || commandLineArgs.hdfsHa != null) {
+            if (commandLineArgs.hdfsUri1 == null || commandLineArgs.hdfsUri2 == null || commandLineArgs.hdfsHa == null) {
+                logger.error("HDFS High availability name node configuration is incomplete. Configure --namenode-1, --namenode-2 and --namenode-ha");
+                System.exit(1);
+            }
+
+            String[] haNames = commandLineArgs.hdfsHa.split(",");
+            if (haNames.length != 2) {
+                logger.error("Cannot process HDFS High Availability mode for other than two name nodes.");
+                System.exit(1);
+            }
+
+            builder.putHdfsConfig("dfs.nameservices", commandLineArgs.hdfsUri);
+            builder.putHdfsConfig("dfs.ha.namenodes." + commandLineArgs.hdfsUri, commandLineArgs.hdfsHa);
+            builder.putHdfsConfig("dfs.namenode.rpc-address." + commandLineArgs.hdfsUri + "." + haNames[0], commandLineArgs.hdfsUri1);
+            builder.putHdfsConfig("dfs.namenode.rpc-address." + commandLineArgs.hdfsUri + "." + haNames[1], commandLineArgs.hdfsUri2);
+            builder.putHdfsConfig("dfs.client.failover.proxy.provider." + commandLineArgs.hdfsUri,"org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");
+        }
+
         RestructureAvroRecords restr = builder.build();
 
         try {
@@ -156,7 +176,8 @@ private RestructureAvroRecords(RestructureAvroRecords.Builder builder) {
     }
 
     public void setInputWebHdfsURL(String fileSystemURL) {
-        conf.set("fs.defaultFS", fileSystemURL);
+        conf.set("fs.defaultFS", "hdfs://" + fileSystemURL);
+        conf.set("fs.default.name", conf.get("fs.defaultFS"));
     }
 
     public void setOutputPath(String path) {

src/main/java/org/radarcns/hdfs/util/commandline/CommandLineArgs.java

@@ -20,9 +20,18 @@ public class CommandLineArgs {
     @Parameter(names = { "-d", "--deduplicate" }, description = "Boolean to define if to use deduplication or not.")
     public boolean deduplicate;
 
-    @Parameter(names = { "-u", "--hdfs-uri" }, description = "The HDFS uri to connect to. Eg - 'hdfs://<HOST>:<RPC_PORT>/<PATH>'.", required = true, validateWith = { HdfsUriValidator.class, PathValidator.class })
+    @Parameter(names = { "-n", "--nameservice"}, description = "The HDFS name services to connect to. Eg - '<HOST>' for single configurations or <CLUSTER_ID> for high availability web services.", required = true, validateWith = { PathValidator.class })
     public String hdfsUri;
 
+    @Parameter(names = { "--namenode-1" }, description = "High availability HDFS name node to also connect to. Eg - '<HOST>:<RPC_PORT>'.", validateWith = { PathValidator.class })
+    public String hdfsUri1;
+
+    @Parameter(names = { "--namenode-2" }, description = "High availability HDFS name node to also connect to. Eg - '<HOST>:<RPC_PORT>'.", validateWith = { PathValidator.class })
+    public String hdfsUri2;
+
+    @Parameter(names = { "--namenode-ha"}, description = "High availability HDFS name node names. Eg - 'nn1,nn2'.", validateWith = { PathValidator.class })
+    public String hdfsHa;
+
     @Parameter(names = { "-o", "--output-directory"}, description = "The output folder where the files are to be extracted.", required = true, validateWith = PathValidator.class)
     public String outputDirectory;