Apache Beam连接Kerberos启用HDFS集群Java示例代码求助

配置后仍报访问控制异常的核心踩坑点 #

• 仅在Driver端执行Kerberos认证：Beam分布式运行时计算逻辑会分发到Worker节点执行，Worker未加载认证逻辑会直接抛出权限异常
• 配置文件未全节点分发：krb5.conf、core-site.xml、hdfs-site.xml、认证用keytab文件必须保证所有Worker节点可访问，不能仅存放在Driver本地路径
• Hadoop FileSystem缓存复用未认证实例：未关闭FS缓存时，节点会复用之前初始化的无认证FileSystem对象，触发访问控制异常

第一步：引入适配依赖 #

注意Hadoop相关依赖版本必须和目标HDFS集群版本完全一致，避免版本不兼容报错：

<dependencies>
    <!-- Beam核心依赖 -->
    <dependency>
        <groupId>org.apache.beam</groupId>
        <artifactId>beam-sdks-java-core</artifactId>
        <version>2.52.0</version>
    </dependency>
    <!-- 本地调试用Direct Runner，提交集群替换为对应Runner（Flink/Spark等） -->
    <dependency>
        <groupId>org.apache.beam</groupId>
        <artifactId>beam-runners-direct-java</artifactId>
        <version>2.52.0</version>
    </dependency>
    <!-- Hadoop HDFS依赖 -->
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-common</artifactId>
        <version>3.3.4</version>
    </dependency>
    <dependency>
        <groupId>org.apache.hadoop</groupId>
        <artifactId>hadoop-hdfs</artifactId>
        <version>3.3.4</version>
    </dependency>
    <!-- Beam HDFS文件系统适配 -->
    <dependency>
        <groupId>org.apache.beam</groupId>
        <artifactId>beam-sdks-java-io-hadoop-file-system</artifactId>
        <version>2.52.0</version>
    </dependency>
</dependencies>

第二步：实现可序列化的Kerberos认证工具类 #

认证逻辑必须支持序列化，保证分发到Worker节点后可正常执行：

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.security.UserGroupInformation;
import java.io.IOException;
import java.io.Serializable;

public class HdfsKerberosUtil implements Serializable {
    // 以下参数可抽离到配置文件统一管理
    private static final String KRB5_CONF_PATH = "/etc/krb5.conf";
    private static final String KEYTAB_PATH = "/opt/resources/hdfs_auth.keytab";
    private static final String PRINCIPAL = "hdfs_user@YOUR_KERBEROS_REALM";
    private static final String CORE_SITE_PATH = "/opt/resources/core-site.xml";
    private static final String HDFS_SITE_PATH = "/opt/resources/hdfs-site.xml";

    public static Configuration getAuthenticatedHdfsConf() throws IOException {
        // 全局加载krb5配置，必须在Hadoop配置初始化前设置
        System.setProperty("java.security.krb5.conf", KRB5_CONF_PATH);
        System.setProperty("javax.security.auth.useSubjectCredsOnly", "false");

        Configuration conf = new Configuration();
        conf.addResource(new Path(CORE_SITE_PATH));
        conf.addResource(new Path(HDFS_SITE_PATH));
        // 强制关闭FileSystem缓存，避免复用未认证实例
        conf.setBoolean("fs.hdfs.impl.disable.cache", true);
        conf.setBoolean("fs.file.impl.disable.cache", true);

        // 执行Kerberos keytab认证
        UserGroupInformation.setConfiguration(conf);
        UserGroupInformation.loginUserFromKeytab(PRINCIPAL, KEYTAB_PATH);
        return conf;
    }
}

第三步：Beam作业读写HDFS示例 #

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.io.hdfs.HadoopFileSystemOptions;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.values.PCollection;
import org.apache.hadoop.conf.Configuration;
import java.io.IOException;
import java.util.Collections;

public class BeamHdfsKerberosJob {
    public static void main(String[] args) throws IOException {
        // 初始化带Kerberos认证的HDFS配置
        Configuration authHdfsConf = HdfsKerberosUtil.getAuthenticatedHdfsConf();

        // 构建Beam配置，注入认证后的HDFS参数
        HadoopFileSystemOptions options = PipelineOptionsFactory
                .create()
                .as(HadoopFileSystemOptions.class);
        options.setHdfsConfiguration(Collections.singletonList(authHdfsConf));

        Pipeline pipeline = Pipeline.create(options);

        // 读Kerberos HDFS文件
        PCollection<String> inputData = pipeline.apply(
                TextIO.read().from("hdfs://namenode_host:8020/user/hdfs_user/input/demo.txt")
        );

        // 写Kerberos HDFS文件
        inputData.apply(
                TextIO.write()
                        .to("hdfs://namenode_host:8020/user/hdfs_user/output/")
                        .withSuffix(".txt")
                        .withNumShards(1)
        );

        pipeline.run().waitUntilFinish();
    }
}

常见报错排查 #

• 报GSS initiate failed：先在节点本地执行kinit -kt 你的keytab路径 你的principal手动验证Kerberos认证是否正常，检查krb5.conf中KDC地址、REALM配置是否正确，节点和KDC服务器网络是否连通
• 报Permission denied：检查认证对应的用户对目标HDFS路径是否有对应读写权限，确认keytab未过期、principal填写正确
• 本地调试正常、集群运行报错：检查作业打包时是否将所有配置文件、keytab纳入资源包，Worker节点是否能访问代码中写的配置文件路径，不要使用仅Driver本地存在的绝对路径
• 注意不要将UGI认证逻辑写在静态代码块中，静态代码块仅会在Driver端执行，Worker节点不会触发该逻辑

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