Flink - 方案没有文件系统：hdfs答案

【问题标题】：Flink - No FileSystem for scheme: hdfsFlink - 方案没有文件系统：hdfs
【发布时间】：2018-02-14 10:12:44
【问题描述】：

我目前正在开发一个从 Hadoop 集群读取 Avro 文件的 Flink 1.4 应用程序。但是，在我的 IDE 上以本地模式运行它可以正常工作。但是当我将它提交给 Jobmanager Flink 时，它总是失败并显示以下消息：

java.io.IOException: Error opening the Input Split hdfs://namenode/topics/CaseLocations/partition=0/CaseLocations+0+0000155791+0000255790.avro [0,16549587]: Could not find a file system implementation for scheme 'hdfs'. The scheme is not directly supported by Flink and no Hadoop file system to support this scheme could be loaded.
at org.apache.flink.api.common.io.FileInputFormat.open(FileInputFormat.java:705)
at org.apache.flink.formats.avro.AvroInputFormat.open(AvroInputFormat.java:110)
at org.apache.flink.formats.avro.AvroInputFormat.open(AvroInputFormat.java:54)
at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:145)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:718)
at java.lang.Thread.run(Thread.java:748)
Caused by: org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Could not find a file system implementation for scheme 'hdfs'. The scheme is not directly supported by Flink and no Hadoop file system to support this scheme could be loaded.
at org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:405)
at org.apache.flink.core.fs.FileSystem.get(FileSystem.java:320)
at org.apache.flink.api.common.io.FileInputFormat$InputSplitOpenThread.run(FileInputFormat.java:864)
Caused by: org.apache.flink.core.fs.UnsupportedFileSystemSchemeException: Hadoop File System abstraction does not support scheme 'hdfs'. Either no file system implementation exists for that scheme, or the relevant classes are missing from the classpath.
at org.apache.flink.runtime.fs.hdfs.HadoopFsFactory.create(HadoopFsFactory.java:102)
at org.apache.flink.core.fs.FileSystem.getUnguardedFileSystem(FileSystem.java:401)
... 2 more
Caused by: java.io.IOException: No FileSystem for scheme: hdfs
at org.apache.hadoop.fs.FileSystem.getFileSystemClass(FileSystem.java:2798)
at org.apache.flink.runtime.fs.hdfs.HadoopFsFactory.create(HadoopFsFactory.java:99)
... 3 more

我正在使用官方 Flink Docker 镜像flink:1.4.0-hadoop28-scala_2.11 运行集群，该镜像应该已经包含一个 Hadoop 发行版。

我也尝试将依赖项添加到我的应用程序 jar 中，但这也无济于事。这是我的 sbt 依赖项：

val flinkVersion = "1.4.0"
val hadoopVersion = "2.8.1"
val providedDependencies = Seq(
    "org.apache.flink" %% "flink-clients" % flinkVersion,
    "org.apache.flink" %% "flink-scala" % flinkVersion,
    "org.apache.flink" %% "flink-streaming-scala" % flinkVersion
)
val compiledDependencies = Seq(
    "org.apache.flink" % "flink-hadoop-fs" % flinkVersion,
    "org.apache.hadoop" % "hadoop-hdfs" % hadoopVersion,
    "org.apache.hadoop" % "hadoop-common" % hadoopVersion,
    "org.apache.flink" % "flink-avro" % flinkVersion,
    "org.apache.flink" %% "flink-table" % flinkVersion,
    "org.scalaj" %% "scalaj-http" % "2.2.1"
)

另外，文件系统类包含在我的META-INF/services/org.apache.hadoop.fs.FileSystem 中。

我错过了什么吗？官方文档帮不了我。

提前致谢

【问题讨论】：

Could not find a file system implementation for scheme 'hdfs'... Flink 是否有用于 Hadoop 配置的 core-site.xml？
包含 hadoop-hdfs 和 hadoop-common 应该可以解决问题。
在我检查了 Jobmanager 和 Taskmanager 的日志之后，在这两种情况下，日志文件都显示：Cannot load filesystem: java.util.ServiceConfigurationError: org.apache.hadoop.fs.FileSystem: Provider org.apache.hadoop.hdfs.DistributedFileSystem not a subtype 知道这是什么原因吗？
请将日志级别更改为调试
同样的问题

标签： hadoop hdfs apache-flink avro

【解决方案1】：

首先，您需要一个 HDFS 集群。

其次，需要检查FLINK_HOME/lib下的flink-shaded-hadoop-2-uber-xxx.xx.jar。

如果您计划将 Apache Flink 与 Apache Hadoop 一起使用（在 YARN 上运行 Flink，连接到 HDFS，连接到 HBase，或者使用一些基于 Hadoop 的文件系统连接器）然后选择捆绑匹配 Hadoop 版本的下载，下载可选的与你的版本匹配的预捆绑 Hadoop 并将其放置在 Flink 的 lib 文件夹中，或者导出你的 HADOOP_CLASSPATH。

【讨论】：

我面临着类似的问题。所以，我用 hadoop 安装了 flink 版本，并添加了 hadoop_conf_dir 作为环境变量。然后我就可以在本地的 flink 实例上提交我的工作了。

【解决方案2】：

我今天也遇到了同样的问题，分两步解决了

检查 HADOOP_CONF_DIR（或 HADOOP_HOME、HADOOP_CLASSPATH）是否正确配置
检查 FLINK_HOME/lib 是否有 flink-shaded-hadoop-2-uber-xxx.jar，如果没有则从here 下载

如果这两个步骤都不ok，你可能需要重启flink集群:)

【讨论】：