PAYATHON 2026

Apache Beam 报错 Unable to find registrar for hdfs 如何解决？

排障赵师傅

结论

Unable to find registrar for hdfs 表示 Beam 在处理 hdfs:// 路径时，没有找到已注册的 HDFS 文件系统实现。fat jar 中有 HadoopFileSystemRegistrar.class，说明类已打包；它是否被发现、获得正确配置并完成注册，还需要分别确认。

对于题目中的 Beam 2.0.0，应依次检查：

  1. fat jar 是否保留了 FileSystemRegistrar 的 SPI 服务声明。
  2. HadoopFileSystemOptions 中是否实际设置了 fs.defaultFS。
  3. 执行 TextIO.write().to(...) 前，是否已使用这些 options 初始化 Beam 文件系统。

题目已经配置了 fs.defaultFS 和 ServicesResourceTransformer，仅用“缺少 HDFS 依赖”无法解释这个错误。接下来需要核查最终运行的 jar，以及构建 pipeline 时的初始化顺序。

为什么包含 registrar 仍然报错

类文件存在，不代表 SPI 能发现它

Beam 通过 Java SPI 发现文件系统 registrar。除类文件外，fat jar 中还应包含：

META-INF/services/org.apache.beam.sdk.io.FileSystemRegistrar

该文件应包含以下实现类名：

org.apache.beam.sdk.io.hdfs.HadoopFileSystemRegistrar

多个依赖可能提供同名服务文件，打包时必须合并。保留题目中的 ServicesResourceTransformer，用它合并这些声明，并检查其他过滤规则是否排除了服务文件。Maven Shade 文档

registrar 需要根据配置创建文件系统

HadoopFileSystemRegistrar.fromOptions(...) 根据 PipelineOptions 创建文件系统，可以返回零个或多个实现。发现 registrar 后，还需要根据配置创建实现，才能注册 hdfs。Beam 2.0.0 API

在这个版本中，应显式设置默认文件系统：

fs.defaultFS=hdfs://192.168.1.214:9000

缺少配置时，可能无法创建 HDFS 实现；如果配置最终指向本地文件系统，还可能出现：

Scheme: [file] has conflicting filesystems

Beam 2.0.0 的问题记录中也有这种配置情况。BEAM-2429

异常发生在构建 pipeline 时

异常堆栈指向：

TextIO$Write.to
FileBasedSink.convertToFileResourceIfPossible
FileSystems.matchNewResource

此时还没有执行 p.run()，但 TextIO.write().to(...) 已经需要解析输出路径。文件系统必须在这之前初始化，不能等到 runner 开始执行。

原代码还调用了 .fromArgs(args1)，忽略了 main(String[] args) 收到的参数。因此，即使在提交命令中修改输入、输出或 HDFS 配置，这些修改也不会生效。

修改代码：先配置并初始化文件系统

保留原来的 WordCountOptions、CountWords 和 FormatAsTextFn，增加以下 import，并替换 main 方法：

import java.util.Collections;

import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.io.FileSystems;
import org.apache.beam.sdk.io.TextIO;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.hadoop.conf.Configuration;
public static void main(String[] args) {
    WordCountOptions options = PipelineOptionsFactory
        .fromArgs(args)
        .withValidation()
        .as(WordCountOptions.class);

    Configuration hdfsConfiguration = new Configuration();
    hdfsConfiguration.set(
        "fs.defaultFS",
        "hdfs://192.168.1.214:9000"
    );

    options.setHdfsConfiguration(
        Collections.singletonList(hdfsConfiguration)
    );

    // 在构建 TextIO、解析 HDFS 路径之前初始化。
    FileSystems.setDefaultPipelineOptions(options);

    Pipeline p = Pipeline.create(options);

    p.apply("ReadLines",
            TextIO.read().from(options.getInputFile()))
        .apply(new CountWords())
        .apply(MapElements.via(new FormatAsTextFn()))
        .apply("WriteCounts",
            TextIO.write().to(options.getOutput()));

    p.run().waitUntilFinish();
}

Beam 2.0.0 中，setHdfsConfiguration(...) 的参数类型是 List<Configuration>，不能直接传入单个 Configuration。HadoopFileSystemOptions API

提前调用 FileSystems.setDefaultPipelineOptions(options)，是因为 driver 在构建 pipeline 时就需要访问文件系统注册表。该方法在 Beam 2.0.0 中标记为 @Internal，这里是针对旧版本的处理；升级 Beam 后，需要按目标版本重新确认初始化机制。FileSystems API

如果希望通过命令行管理 HDFS 配置，可以删除上述创建 Configuration 和调用 setHdfsConfiguration(...) 的代码，改为传入：

'--hdfsConfiguration=[{"fs.defaultFS":"hdfs://192.168.1.214:9000"}]'

这两种方式选一种即可。代码中的赋值会覆盖命令行解析出的 hdfsConfiguration。

检查依赖和最终 fat jar

POM 中的 beam-sdks-java-io-hadoop-file-system 重复声明了两次，应合并为一项。源代码直接引用了 HadoopFileSystemOptions，使用默认的 compile scope 即可：

<dependency>
  <groupId>org.apache.beam</groupId>
  <artifactId>beam-sdks-java-io-hadoop-file-system</artifactId>
  <version>${beam.version}</version>
</dependency>

保留 Shade 插件中的服务合并配置：

<transformers>
  <transformer implementation=
      "org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
</transformers>

重新打包后，检查服务声明的实际内容：

mvn clean package

unzip -p target/word-count-ck-0.1.jar \
  META-INF/services/org.apache.beam.sdk.io.FileSystemRegistrar

输出中应包含 HadoopFileSystemRegistrar 的完整类名。确认后，将新 jar 上传到提交命令使用的 HDFS 路径，避免继续运行旧包。

依赖还需要检查以下几项：

  • 所有 Beam 模块保持同一版本。
  • 只使用 Spark runner 时，可以移除无关的 Flink runner。
  • hadoop-common 并不包含完整的 HDFS 客户端，需要确认运行环境提供了 HDFS 实现。
  • 将题目额外指定的 hadoop-common:3.0.0-alpha2 与 Spark 发行包中的 Hadoop 库核对，避免混入两套客户端版本。

如果 Spark 集群已经提供匹配的 Spark、Scala 和 Hadoop 库，就应避免将另一套版本打入 fat jar。Spark 1.6.3 的提交文档建议在这种情况下将 Spark 和 Hadoop 依赖设为 provided，前提是 driver 和 executor 的运行环境确实提供这些库。Spark 1.6.3 提交文档

调整提交命令并确认结果

排查期间可以先去掉 --supervise，避免 driver 异常退出后自动重启，反复报出相同错误。提交命令继续使用题目中已成功提交请求的 REST 地址：

./spark-submit \
  --class org.apache.beam.examples.WordCount \
  --master spark://192.168.1.214:6066 \
  --deploy-mode cluster \
  --executor-memory 2G \
  --total-executor-cores 4 \
  hdfs://192.168.1.214:9000/input/word-count-ck-0.1.jar \
  --runner=SparkRunner \
  --inputFile=hdfs://192.168.1.214:9000/beamWorks/kinglear.txt \
  --output=hdfs://192.168.1.214:9000/beamWorks/ckoutput/ck

success: true 表示提交请求成功，日志中的 RUNNING 表示 driver 当时已经启动，两者都不足以确认 pipeline 已完成。--supervise 会在 driver 以非零退出码结束后尝试重启；是否反复重启，需要查看后续 driver 日志。Spark 1.6.3 Standalone 文档

这些地址的用途也需要区分：

地址用途
192.168.1.214:6066题目中的 Spark REST 提交端口
Spark Master RPC 地址SparkContext 连接集群所用地址，默认端口通常为 7077
192.168.1.214:9000题目中的 HDFS NameNode 地址

如果还需要通过 Beam 的 --sparkMaster 指定执行集群，应填写实际的 Spark Master RPC 地址，不要直接复用 REST 提交端口。

注册错误消失后，再检查 driver 和 executor 是否都能访问 NameNode、DataNode，以及是否具备输入读取和输出写入权限。复杂集群还需要为运行节点提供正确的 core-site.xml、hdfs-site.xml 及认证配置。能从 HDFS 下载作业 jar，并不足以确认所有 executor 都能正常读写数据。

TextIO.write().to(...) 接收的是输出文件前缀，作业可能生成多个以 ck 开头的分片文件。判断作业是否完成，需要结合 driver 最终状态、executor 日志和 HDFS 输出文件。后续如果出现连接、权限或类版本异常，应继续排查新的异常，不能据此认定 registrar 修复无效。

备注：内容仅供参考。