Java 連接 HDFS HA 與提交 MapReduce Job 完整範例

場景

生產環境 Hadoop 採用 **NameNode HA (High Availability)** 架構，Java 客戶端需正確配置 `dfs.ha.namenodes`、`failover proxy` 才能自動切換 Active/Standby NameNode。同時提交 MapReduce Job 需設定 YARN ResourceManager 與 HDFS 相關參數。

1. HDFS HA 客戶端配置與列表根目錄

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

public class HdfsHaClient {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();

        // --- 核心 HA 配置 ---
        conf.set("fs.defaultFS", "hdfs://mycluster");          // 邏輯名稱
        conf.set("dfs.nameservices", "mycluster");             // 服務名
        conf.set("dfs.ha.namenodes.mycluster", "nn1,nn2");     // 兩個 NameNode ID
        conf.set("dfs.namenode.rpc-address.mycluster.nn1", "nn1-host:8020");
        conf.set("dfs.namenode.rpc-address.mycluster.nn2", "nn2-host:8020");
        // 自動故障轉移代理
        conf.set("dfs.client.failover.proxy.provider.mycluster",
                "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider");

        // 可選：超時與重試
        conf.setInt("ipc.client.connect.timeout", 5000);
        conf.setInt("ipc.client.connect.max.retries", 3);

        try (FileSystem fs = FileSystem.get(conf)) {
            FileStatus[] statuses = fs.listStatus(new Path("/"));
            for (FileStatus status : statuses) {
                System.out.println(status.getPath().getName() + "\t" + 
                        (status.isDirectory() ? "DIR" : "FILE") + "\t" + status.getLen());
            }
        }
    }
}

2. 程式化提交 MapReduce Job (WordCount 範例)

使用 `RunJar` 模擬 `hadoop jar` 指令，將配置透過 `-D` 參數傳入。

public class SubmitMrJob {
    public static void main(String[] args) throws Exception {
        String[] params = new String[] {
            // 1. Jar 包路徑 (含 WordCount 範例)
            "/opt/hadoop/share/hadoop/mapreduce/hadoop-mapreduce-examples-3.3.6.jar",
            "wordcount",                    // 主類別名稱 (範例類)

            // 2. YARN ResourceManager
            "-D", "yarn.resourcemanager.address=rm-host:8032",
            "-D", "yarn.resourcemanager.scheduler.address=rm-host:8030",

            // 3. HDFS HA (同檔案系統配置)
            "-D", "fs.defaultFS=hdfs://mycluster/",
            "-D", "dfs.nameservices=mycluster",
            "-D", "dfs.ha.namenodes.mycluster=nn1,nn2",
            "-D", "dfs.namenode.rpc-address.mycluster.nn1=nn1-host:8020",
            "-D", "dfs.namenode.rpc-address.mycluster.nn2=nn2-host:8020",
            "-D", "dfs.client.failover.proxy.provider.mycluster=org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider",

            // 4. 執行引擎
            "-D", "mapreduce.framework.name=yarn",
            "-D", "fs.hdfs.impl=org.apache.hadoop.hdfs.DistributedFileSystem",

            // 5. 輸入/輸出路徑
            "/user/input/text",   // HDFS 輸入目錄
            "/user/output/wc1"    // HDFS 輸出目錄 (必須不存在)
        };

        // 執行 (RunJar 會啟動新 JVM，阻塞至 Job 完成)
        org.apache.hadoop.util.RunJar.main(params);
    }
}

3. Maven 依賴 (Hadoop 3.3.x)

    
        org.apache.hadoop
        hadoop-client
        3.3.6
    
    
        org.apache.hadoop
        hadoop-hdfs
        3.3.6
    
    
        org.apache.hadoop
        hadoop-mapreduce-client-jobclient
        3.3.6
    

關鍵配置說明表

進階：使用 FileSystem API 直接提交 Job (非 RunJar)

import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class SubmitJobApi {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        // ... 同樣設定 HA/YARN 參數 ...

        Job job = Job.getInstance(conf, "MyWordCount");
        job.setJarByClass(SubmitJobApi.class); // 打包含 Mapper/Reducer 的 Jar
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        FileInputFormat.addInputPath(job, new Path("/user/input/text"));
        FileOutputFormat.setOutputPath(job, new Path("/user/output/wc_api"));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}

常見坑點

• ❌ **Jar 衝突**：客戶端 Hadoop 版本**必須**與叢集版本一致 (如皆為 3.3.6)，否則序列化/協議不匹配。
• ❌ **輸出目錄已存在**：MapReduce 預設拋錯，需先刪除或設定 `mapreduce.output.fileoutputformat.outputdir.overwrite=true` (Hadoop 3.2+)。
• ❌ **Kerberos 認證**：安全叢集需 `UserGroupInformation.loginUserFromKeytab()` 或設定 `hadoop.security.authentication=kerberos`。
• ❌ **網路不通**：客戶端機器需能解析並連線 NameNode、ResourceManager、DataNode、NodeManager 所有埠口。

掌握 HA 配置與程式化提交，即可在 CI/CD 流水線、資料同步排程、微服務中無縫整合 Hadoop 生態。