Question

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

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-SNAPSHOT 或 org.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
• s3n 和 s3a
• 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.input 和 fs.<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 文档 。