文件系统

此页面提供有关设置和配置与Flink一起使用的分布式文件系统的详细信息。

Flink使用文件系统作为流/批处理应用程序中的源和接收器,以及作为检查点的目标。例如,这些文件系统可以是Unix / Windows文件系统HDFS,甚至是S3等对象存储

用于特定文件的文件系统由文件URI的方案确定。例如,file:///home/user/text.txt指的是本地文件系统中的文件,而hdfs://namenode:50010/data/user/text.txt指的是特定HDFS集群中的文件。

文件系统通过org.apache.flink.core.fs.FileSystem表示,该类捕获访问和修改该文件系统中的文件和对象的方法。FileSystem实例每个进程实例化一次,然后缓存/池化,以避免每个流创建的配置开销,并强制执行某些约束,如连接/流限制。

内置文件系统

Flink直接实现以下文件系统:

  • local:当方案为“file://”时使用此文件系统,它表示本地计算机的文件系统,包括挂载到该本地文件系统的任何NFS或SAN。

  • S3:Flink直接提供与Amazon S3通信的文件系统,在方案“s3://”下注册有两种替代实现,flink-s3-fs-presto并且flink-s3-fs-hadoop基于Presto项目Hadoop项目的代码两种实现都是自包含的,没有依赖性足迹。要在使用Flink作为库时使用它们,请添加相应的maven依赖项(org.apache.flink:flink-s3-fs-presto:1.7-SNAPSHOTorg.apache.flink:flink-s3-fs-hadoop:1.7-SNAPSHOT)。从Flink二进制文件启动Flink应用程序时,将相应的jar文件从opt文件夹复制或移动到该lib文件夹。请参阅AWS设置了解详细信息

  • MapR FS当MapR库在类路径中时,MapR文件系统“maprfs://”自动可用。

  • OpenStack Swift FS:Flink直接提供一个文件系统,与OpenStack Swift文件系统通信,该系统在“swift://”方案下注册该实现flink-swift-fs-hadoop基于Hadoop项目,但是是自包含的,没有依赖性足迹。要在使用Flink作为库时使用它,请添加相应的maven依赖项(org.apache.flink:flink-swift-fs-hadoop:1.7-SNAPSHOT 从Flink二进制文件启动Flink应用程序时,将相应的jar文件从opt文件夹复制或移动到该lib文件夹。

HDFS和Hadoop文件系统支持

对于Flink未实现文件系统本身的方案,Flink将尝试使用Hadoop为相应的方案实例化文件系统。所有Hadoop文件系统都可自动使用一次flink-runtime,相关的Hadoop库位于类路径中。

这样,Flink无缝支持所有Hadoop文件系统,以及所有与Hadoop兼容的文件系统(HCFS),例如:

  • HDFS
  • FTP
  • s3ns3a
  • har
  • ...

常见文件系统配置

不同文件系统中存在以下配置设置。

默认文件系统

如果文件路径未明确指定文件系统方案(和权限),则将使用默认方案(和权限)。

fs.default-scheme: <default-fs>

例如,如果将默认文件系统配置为fs.default-scheme: hdfs://localhost:9000/,则将文件路径 /user/hugo/in.txt解释为hdfs://localhost:9000/user/hugo/in.txt

连接限制

您可以限制文件系统可以同时打开的连接总数。当文件系统无法同时处理大量并发读/写或打开连接时,这很有用。

例如,具有少量RPC处理程序的非常小的HDFS集群有时会被一个大的Flink作业所淹没,这个作业试图在检查点期间建立许多连接。

要限制特定文件系统的连接,请将以下条目添加到Flink配置中。要限制的文件系统由其方案标识。

fs.<scheme>.limit.total: (number, 0/-1 mean no limit)
fs.<scheme>.limit.input: (number, 0/-1 mean no limit)
fs.<scheme>.limit.output: (number, 0/-1 mean no limit)
fs.<scheme>.limit.timeout: (milliseconds, 0 means infinite)
fs.<scheme>.limit.stream-timeout: (milliseconds, 0 means infinite)

如果输入/输出连接(流)单独(fs.<scheme>.limit.inputfs.<scheme>.limit.output),您可以限制数量,并对并发流(fs.<scheme>.limit.total的总数施加限制如果文件系统尝试打开更多流,则 算子操作将阻塞,直到某些流关闭。如果fs.<scheme>.limit.timeout流的开放时间超过,则流开放将失败。

要防止非活动流占用整个池(阻止打开新连接),可以为流添加不活动超时: fs.<scheme>.limit.stream-timeout如果流至少在该时间内没有读/写任何字节,则强制关闭它。

这些限制是根据TaskManager强制执行的,因此Flink应用程序或集群中的每个TaskManager都将打开该数量的连接。此外,还仅对每个FileSystem实例强制执行限制。由于文件系统是按照方案和权限创建的,因此不同的权限将拥有自己的连接池。例如hdfs://myhdfs:50010/hdfs://anotherhdfs:4399/将有单独的池。

添加新的文件系统实现

Flink通过Java服务抽象发现文件系统实现,从而可以轻松添加其他文件系统实现。

要添加新文件系统,需要执行以下步骤:

  • 添加文件系统实现,它是的子类org.apache.flink.core.fs.FileSystem
  • 添加一个实例化该文件系统的工厂,并声明注册FileSystem的方案。这必须是的子类org.apache.flink.core.fs.FileSystemFactory
  • 添加服务条目。创建一个META-INF/services/org.apache.flink.core.fs.FileSystemFactory包含文件系统工厂类的类名的文件。

有关服务加载器如何工作的更多详细信息,请参阅Java Service Loader文档

回到顶部