Flink对接HDFS报错Wrong FS expected file:///问题咨询

你遇到的报错核心是手动创建的Hadoop Configuration未正确加载HDFS配置，导致FileSystem默认初始化了本地文件系统实例，和你传入的HDFS路径不匹配。以下是具体的可能原因和对应解决方法：

可能的问题点 #

• 配置未正确加载：Flink配置的fs.hdfs.hadoopconf属性是供给Flink内置的FileSystem实现使用的，你代码中直接通过Job.getInstance().getConfiguration()获取的配置对象，只会加载JVM classpath下的Hadoop配置文件，不会自动读取Flink配置的Hadoop配置目录内容。如果你的Hadoop配置文件没有打到作业Jar包、也不在节点的默认classpath中，该配置对象就不会包含fs.defaultFS、fs.hdfs.impl等HDFS相关配置，最终返回LocalFileSystem实例。
• FileSystem获取方式错误：直接调用FileSystem.get(hadoopConf)获取的是配置中fs.defaultFS对应的默认文件系统实例，当你的路径前缀和默认文件系统不匹配时就会抛出对应错误。正确的做法是用待操作的Path对象获取对应文件系统实例。
• 依赖缺失或冲突：如果你的作业没有引入hadoop-hdfs依赖，或者依赖版本和集群Hadoop版本不兼容，会导致Hadoop的SPI机制无法找到DistributedFileSystem实现类，最终 fallback 到本地文件系统。
• 配置被覆盖：如果作业Jar包中自带了core-site.xml、或者classpath中存在优先级更高的配置文件，将fs.defaultFS覆盖为file:///，也会触发该错误。

解决步骤 #

1. 正确获取Flink托管的Hadoop配置，不要手动创建Job实例拿配置，在Sink的open方法中通过Flink的上下文获取配置：

// 在RichSinkFunction的open方法中调用
val flinkConf = getRuntimeContext.getConfiguration
val hadoopConf = org.apache.flink.runtime.util.HadoopUtils.getHadoopConfiguration(flinkConf)

通过该方式拿到的配置会自动加载fs.hdfs.hadoopconf指向目录下的所有Hadoop配置。

2. 修改FileSystem实例获取逻辑：

// 优先用Path获取对应文件系统，不要用全局默认的FileSystem
val fs = path.getFileSystem(hadoopConf)
val os = fs.create(path)

如果使用相对路径，建议给Path补全HDFS前缀：new Path("hdfs://hostname:port/path/to/one/hdfs/dir")

3. 检查依赖配置：确保作业引入了和集群版本兼容的hadoop-client、hadoop-hdfs依赖，打包作业时将Hadoop依赖 scope 设为provided，避免和集群内置依赖冲突。

4. 配置校验：可以在代码中打印配置项确认加载结果：

println(hadoopConf.get("fs.defaultFS"))
println(hadoopConf.get("fs.hdfs.impl"))

正常情况下fs.hdfs.impl的取值应该是org.apache.hadoop.hdfs.DistributedFileSystem。

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