Flink集群配置不同S3端点用于Checkpoint与FileSink

完全可以为Checkpoint和FileSink配置不同的S3端点，无需替换FileSink或自行实现WindowFunction，Flink本身提供了更简洁的原生方案：

方案1：注册多个同类型不同配置的文件系统（推荐） #

Flink支持通过自定义scheme注册多个同类型但参数不同的文件系统实例，以此隔离Checkpoint和FileSink的S3端点：

1. 在集群配置flink-conf.yaml中新增一套S3配置，用新的scheme区分：

# 原有FileSink使用的S3端点配置（scheme: s3p）
s3p.endpoint: http://your-filesink-s3-endpoint
s3p.access-key: xxx
s3p.secret-key: xxx
s3p.path.style.access: true

# 新增Checkpoint专用的S3端点配置（scheme: s3c，可自定义不冲突的名称）
s3c.endpoint: http://your-checkpoint-s3-endpoint
s3c.access-key: yyy
s3c.secret-key: yyy
s3c.path.style.access: true

# 注册两个文件系统实例
fs.s3p.impl: org.apache.flink.fs.s3p.common.S3FileSystem
fs.s3c.impl: org.apache.flink.fs.s3p.common.S3FileSystem

2. 代码中分别指定对应scheme的路径：
   • Checkpoint路径设为 s3c://bckt/checkpoints
   • FileSink输出路径设为 s3p://bckt/sink-output

两者会自动使用各自对应的S3端点，配置完全隔离。

方案2：单独为FileSink指定自定义文件系统配置 #

若不想新增scheme，可在代码中为FileSink单独注入自定义配置的文件系统，不影响集群全局的Checkpoint配置：

// 构建FileSink专用的S3配置
Configuration sinkFsConfig = new Configuration();
sinkFsConfig.setString("s3p.endpoint", "http://your-filesink-s3-endpoint");
sinkFsConfig.setString("s3p.access-key", "xxx");
sinkFsConfig.setString("s3p.secret-key", "xxx");

// 获取自定义配置的文件系统实例
FileSystem sinkFs = FileSystem.get(new URI("s3p:///"), sinkFsConfig);

// 创建FileSink时注入该文件系统
FileSink<String> sink = FileSink
    .forBulkFormat(new Path("s3p://bckt/sink-output"), BulkFormats.toString())
    .withBucketAssigner(new DateTimeBucketAssigner<>("yyyy-MM-dd"))
    .setFileSystem(sinkFs) // 指定自定义文件系统
    .build();

// Checkpoint继续使用集群全局配置的s3p端点
env.getCheckpointConfig().setCheckpointStorage("s3p://bckt/checkpoints");

关于你提出的方案补充 #

• 用WindowFunction替代FileSink：虽可行，但需自行实现批量写入的容错、分桶、文件滚动等生产级特性，重复造轮子会大幅增加维护成本，不推荐。
• 修改FileSink源码：无需操作，Flink原生提供的setFileSystem()方法已支持注入自定义配置的文件系统，完全满足需求。

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