Apache Flink写入S3后如何获取或配置CSV生成文件名？

一、是否可以自定义文件名（去掉UUID） #

Flink FileSink默认添加UUID是为了保证分布式环境下的文件唯一性——避免多并行子任务、滚动生成的文件片段因重名导致覆盖或冲突。如果业务场景允许（比如单并行度任务、无并发写入冲突），可通过以下方式自定义文件名：

1. 自定义BucketAssigner与FileNamePolicy
   实现BucketAssigner控制桶划分（比如固定桶），再通过withFileNamePolicy自定义文件名生成逻辑。示例代码：

   // 自定义文件名策略，移除UUID
   FileNamePolicy fileNamePolicy = new FileNamePolicy() {
       @Override
       public Path resolveBucketId(Path basePath, BucketID bucketId) {
           return basePath; // 固定桶路径
       }
   
       @Override
       public Path getPendingPath(Path bucketPath, String pendingId) {
           // 自定义待提交文件路径
           return new Path(bucketPath, "test-pending.csv");
       }
   
       @Override
       public Path getCompletedPath(Path bucketPath, String pendingId) {
           // 最终完成的文件名
           return new Path(bucketPath, "test.csv");
       }
   };
   
   FileSink<RowData> sink = FileSink.forRowFormat(new Path(s3FilePath), new SerializationSchemaAdapter(serSchemaBuilder))
           .withOutputFileConfig(new OutputFileConfig("test", ".csv"))
           .withBucketAssigner(new BucketAssigner<RowData, String>() {
               @Override
               public String getBucketId(RowData element, BucketAssigner.Context context) {
                   return "fixed-bucket"; // 固定桶ID，所有数据写入同一桶
               }
   
               @Override
               public SimpleVersionedSerializer<String> getSerializer() {
                   return SimpleVersionedStringSerializer.INSTANCE;
               }
           })
           .withFileNamePolicy(fileNamePolicy)
           .build();
   

   注意：多并行度任务使用固定文件名会导致文件覆盖，仅适用于单并行度或串行写入场景。

2. 限制滚动策略（减少文件片段）
   通过withRollingPolicy配置滚动条件（比如仅在任务结束时滚动），减少生成的文件数量，但UUID仍会存在，需结合上述自定义文件名策略才能完全移除。

二、如何获取生成的文件名 #

Flink没有直接提供从Sink返回文件名的API（因Sink是分布式异步执行），可通过以下方式实现：

1. 自定义Committer收集文件名 #

在文件提交到S3时，通过自定义Committer捕获完成的文件路径：

// 自定义Committer，记录生成的文件名
class CustomCommitter implements Committer<FileCommittable> {
    @Override
    public List<FileCommittable> commit(List<FileCommittable> committables) throws IOException {
        for (FileCommittable committable : committables) {
            // 获取完整S3文件路径
            String filePath = committable.getPath().toString();
            // 可将文件名写入日志、数据库或其他存储
            System.out.println("生成文件：" + filePath);
        }
        return committables;
    }

    @Override
    public void close() throws Exception {}
}

// 构建Sink时配置自定义Committer
FileSink<RowData> sink = FileSink.forRowFormat(new Path(s3FilePath), new SerializationSchemaAdapter(serSchemaBuilder))
        .withOutputFileConfig(new OutputFileConfig("test", ".csv"))
        .withCommitter(new CustomCommitter())
        .build();

此方式在文件提交时触发，能准确获取所有生成的文件路径。

2. 遍历S3指定前缀（适合离线场景） #

任务结束后，用AWS SDK遍历S3目标前缀下的文件：

// 示例：用AWS SDK列出S3前缀下的文件
AmazonS3 s3Client = AmazonS3ClientBuilder.defaultClient();
ListObjectsV2Result result = s3Client.listObjectsV2("你的存储桶名称", "test");
for (S3ObjectSummary objectSummary : result.getObjectSummaries()) {
    System.out.println("生成文件：" + objectSummary.getKey());
}

注意：此方式存在延迟，需处理重复文件问题。

3. 自定义Metric收集文件名 #

通过Flink Metric系统，在文件写入完成时将文件名作为Metric上报，后续从Metric系统中提取。

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