文件系统

译者：flink.sojb.cn

此页面提供有关设置和配置与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
  • …

常见文件系统配置

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

默认文件系统

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

  1. 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配置中。要限制的文件系统由其方案标识。

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

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

要防止非活动流占用整个池（阻止打开新连接），可以为流添加不活动超时： fs.&lt;scheme&gt;.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文档。