Flink基于当前时间消费带时间戳S3文件并支持Checkpoint可行性咨询

该需求完全可以通过Flink现有基于文件的Source实现，不需要自定义开发数据源，且原生支持checkpoint机制，方案具备生产级可用性。

你可以直接使用Flink 1.12及以上版本提供的官方FileSource组件，具体配置要点如下：

• 首先引入Flink S3文件系统依赖，使用官方兼容Hadoop的S3实现即可，无需额外适配S3访问逻辑。
• 配置增量监听规则：设置FileSource为持续监听模式，扫描间隔可根据你的延迟要求设置（建议不超过1分钟），同时新增文件过滤逻辑：
  • 优先选择解析你路径中自带的时间戳进行筛选，精度更高，可避免S3对象元数据的LastModified时间和文件实际生成时间不一致的问题
  • 也可以直接读取S3对象元数据的LastModified属性，筛选最近5分钟内更新的对象
    参考代码片段：

  FileSource<String> s3Source = FileSource
    .forRecordStreamFormat(new TextLineFormat(), new Path("s3://你的Bucket根路径"))
    .monitorContinuously(Duration.ofMinutes(1)) // 每分钟扫描一次新增文件
    .setFileFilter(fileStatus -> {
        // 方案1：基于路径时间戳过滤
        String fileName = fileStatus.getPath().getName();
        // 此处补充你自己的路径时间戳解析逻辑，判断是否在最近5分钟内即可
        // 方案2：基于S3对象修改时间过滤
        // return System.currentTimeMillis() - fileStatus.getLastModifiedTime() < 5 * 60 * 1000;
    })
    .build();
  

• Checkpoint适配无需额外开发：FileSource原生和Flink checkpoint机制对齐，已处理的文件列表、文件读取偏移量都会自动存入checkpoint快照，任务故障恢复后会自动从上次中断的位置继续处理，不会出现数据重复或丢失的问题。

• 如果你的S3 Bucket下文件量级很大，可以适当调大扫描间隔，避免频繁调用S3 List接口触发云厂商的接口限流
• 可开启file.source.cleanup-processed-files配置，自动标记/清理已经处理完成的文件，减少后续扫描的无效校验开销

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