Apache Beam KafkaIO配置：Spark Yarn模式下JKS信任库路径指定失败

问题核心原因 #

Kafka客户端的SSL配置依赖Java本地文件IO读取证书文件，只能识别本地文件系统路径，而你提供的是HDFS（或MapRFS）分布式文件系统路径，执行器上的Kafka客户端无法直接访问分布式文件系统中的文件，因此抛出文件找不到的异常。

具体解决方法 #

1. 用Spark的--files参数分发证书到执行器本地

这是最直接的方案，利用Spark的文件分发机制，在提交应用时将JKS文件上传到每个执行器的当前工作目录：

• 提交命令添加--files参数：

  spark-submit --class com.your.package.YourApp \
    --master yarn \
    --deploy-mode client \
    --files /本地路径/myapp1.jks \
    your-app.jar
  

• 修改KafkaIO配置，直接使用文件名即可：

  props.put("ssl.keystore.location", "myapp1.jks");
  

• 验证：可以在消费数据的DoFn中打印new File("myapp1.jks").getAbsolutePath()，确认文件存在于执行器本地。

2. 手动从HDFS拷贝证书到执行器临时目录

如果必须从HDFS读取证书，可在Beam任务初始化阶段，通过Hadoop API将文件拷贝到执行器本地临时目录：

public class KafkaConsumeFn extends DoFn<KV<String, String>, String> {
    private String localJksPath;

    @Override
    public void setup() throws Exception {
        // 初始化HDFS文件系统客户端
        Configuration hdfsConf = new Configuration();
        FileSystem hdfsFs = FileSystem.get(new URI("hdfs://your-nn-host:8020"), hdfsConf);
        
        // 定义HDFS路径和本地临时路径
        Path hdfsJksPath = new Path("/users/<user-id>/.staging/application-<id>/myapp1.jks");
        Path localTmpPath = new Path(System.getProperty("java.io.tmpdir"), "myapp1.jks");
        
        // 拷贝文件到本地
        hdfsFs.copyToLocalFile(false, hdfsJksPath, localTmpPath);
        localJksPath = localTmpPath.toString();
        
        // 动态更新Kafka SSL配置（如果是在消费者初始化前执行）
        System.setProperty("ssl.keystore.location", localJksPath);
    }

    @ProcessElement
    public void processElement(ProcessContext c) {
        // 处理Kafka数据
        c.output(c.element().getValue());
    }
}

• 注意：需要确保执行器进程有HDFS访问权限和临时目录写入权限。

3. 修正Kafka SSL配置的拼写错误

你的配置中ssl.key.passwords是错误的，正确的配置项为**ssl.key.password（单数）**，拼写错误会导致SSL认证失败，先修正该配置：

props.put("ssl.key.password", "mypassword");

4. 验证文件权限与路径

• 检查JKS文件在HDFS上的权限：确保执行器的Yarn用户（通常是yarn或提交用户）有读取权限，可通过hdfs dfs -ls /users/<user-id>/.staging/application-<id>/myapp1.jks查看。
• 在执行器日志中搜索myapp1.jks，确认文件是否被正确分发到执行器本地。

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