如何配置Flink File Sink向FS、HDFS、S3写入数据时设置全权限

问题原因 #

Flink File Sink 生成的文件、目录权限默认由Flink进程umask、底层存储系统默认配置共同决定，父目录的权限不会自动继承给新生成的子资源，所以单独设置父目录777不会生效。 #

解决方案 #

方案1：修改Flink集群全局配置（无需改代码，所有作业生效） #

1. 编辑Flink安装目录下的conf/flink-conf.yaml，添加如下配置：

# umask设为000，对应生成的目录权限为777，文件默认权限为666
fs.permissions.umask-mode: 000

2. 重启Flink集群后所有新提交的作业都会生效。
3. 如果是写入HDFS，还需要检查HDFS服务端的hdfs-site.xml配置，确保没有强制权限限制：

<property>
    <name>dfs.permissions.enabled</name>
    <value>false</value>
</property>

如果需要文件默认带执行权限，添加如下配置：

<property>
    <name>dfs.webhdfs.create-permission</name>
    <value>777</value>
</property>

方案2：作业代码内配置（仅当前作业生效，灵活度更高） #

在你的代码中添加全局umask配置即可，修改后的完整代码如下：

import org.apache.flink.api.common.serialization.SimpleStringEncoder
import org.apache.flink.core.fs.Path
import org.apache.flink.connector.file.sink.FileSink
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.DefaultRollingPolicy
import org.apache.flink.configuration.GlobalConfiguration
import org.apache.flink.core.fs.FileSystem
import java.util.concurrent.TimeUnit

val input: DataStream[String] = ...

// 为当前作业设置文件系统umask为000
val flinkConf = GlobalConfiguration.loadConfiguration()
flinkConf.setString("fs.permissions.umask-mode", "000")
FileSystem.initialize(flinkConf, null)

val sink: FileSink[String] = FileSink
    .forRowFormat(new Path(outputPath), new SimpleStringEncoder[String]("UTF-8"))
    .withRollingPolicy(
        DefaultRollingPolicy.builder()
            .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
            .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
            .withMaxPartSize(1024 * 1024 * 1024)
            .build())
    .build()

input.sinkTo(sink)

注意事项 #

• 写入本地文件系统时，需要在启动Flink集群/作业的终端先执行umask 000，避免系统默认umask覆盖Flink配置
• S3对象存储没有POSIX权限体系，权限控制依赖bucket策略和IAM身份配置，无需设置上述umask参数
• 如果需要生成的普通文件带执行权限，可以在文件滚动完成后调用底层文件系统的chmod接口批量修改，也可以自定义FileLifeCycleListener在文件提交时自动修改权限为777

内容的提问来源于stack exchange，提问作者Vadim