flink-scheme:hdfs没有文件系统

col17t5w  于 2021-05-29  发布在  Hadoop
关注(0)|答案(1)|浏览(692)

我目前正在开发一个flink1.4应用程序,从hadoop集群读取avro文件。然而,在我的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 .
我错过什么了吗?官方文件帮不了我。
提前谢谢

b4lqfgs4

b4lqfgs41#

首先,您需要一个hdf集群。
其次,您需要检查flink\u home/lib下的flink-shaded-hadoop-2-uber-.xx.jar。
如果您计划将apache flink与apache hadoop一起使用(在yarn上运行flink、连接到hdfs、连接到hbase或使用一些基于hadoop的文件系统连接器),则选择捆绑匹配hadoop版本的下载,下载与您的版本匹配的可选预捆绑hadoop并将其放在flink的lib文件夹中,或者导出hadoop\u类路径。

相关问题