基于CDC实现Amazon S3新增文件事件推送至Apache Flink并处理

整体流程 #

当S3存储桶中有新文件创建时，通过S3的事件通知机制（即CDC捕获文件变更事件）将事件推送到消息中间件（如Kinesis Data Streams/Kafka），再由Apache Flink消费这些事件，最终读取对应S3文件并执行处理逻辑。

具体实现步骤 #

• 配置S3事件通知（捕获文件新增事件）

  1. 登录AWS控制台，进入目标S3存储桶的「属性」页面，找到「事件通知」选项
  2. 创建新通知：
     • 填写通知名称，选择事件类型为 s3:ObjectCreated:*（涵盖所有文件创建/上传场景）
     • 设置过滤规则（可选）：比如指定前缀data/或后缀.parquet，只监听特定目录或文件类型
     • 选择目标投递服务：推荐使用Kinesis Data Streams或Apache Kafka，作为Flink和S3之间的事件中转层

• Flink消费事件流
  使用Flink对应的连接器消费事件中间件中的S3事件，解析出文件的存储路径等关键信息。示例代码（以Kinesis为例）：

  StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
  
  Properties kinesisProps = new Properties();
  kinesisProps.setProperty("aws.region", "us-east-1");
  kinesisProps.setProperty("flink.stream.initpos", "LATEST");
  
  DataStream<String> s3Events = env.addSource(new FlinkKinesisConsumer<>(
      "your-kinesis-stream-name",
      new SimpleStringSchema(),
      kinesisProps
  ));
  
  // 解析S3事件，提取对象键
  DataStream<String> s3ObjectKeys = s3Events.map(event -> {
      JSONObject eventJson = JSONObject.parseObject(event);
      return eventJson.getJSONArray("Records")
                      .getJSONObject(0)
                      .getJSONObject("s3")
                      .getJSONObject("object")
                      .getString("key");
  });
  

• Flink读取并处理S3文件
  基于解析出的S3对象键，使用Flink的FileSystem连接器读取文件并执行业务处理：

  s3ObjectKeys.flatMap((String key, Collector<String> out) -> {
      String s3Path = "s3://your-bucket-name/" + key;
      // 读取S3文件，这里以文本文件为例，可替换为Parquet/CSV等格式的InputFormat
      DataStream<String> fileContent = env.readTextFile(s3Path);
      // 执行自定义处理逻辑，比如数据清洗、转换
      fileContent.map(line -> line.toUpperCase()).print();
  }).setParallelism(1);
  
  env.execute("S3 File Processing Job");
  

关键注意事项 #

• 权限配置：确保Flink集群拥有S3存储桶的读取权限，以及事件中间件的消费权限（如Kinesis的kinesis:DescribeStream、kinesis:GetRecords权限）
• 幂等处理：S3事件可能存在重复推送的情况，需在Flink处理逻辑中实现幂等性（比如基于文件MD5或唯一ID去重）
• 性能优化：针对大文件场景，可使用Flink的并行读取机制，或结合文件拆分策略提升处理效率

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