如何配置Kafka S3 Sink Connector将多条记录合并写入单个S3对象？

你需要用S3 Sink Connector自身的文件滚动与刷新参数来实现批量写入，Kafka Consumer的fetch参数不控制S3文件的生成逻辑，以下是具体配置方案：

• 按记录数/内存大小批量写入
  用flush.size设置累计多少条记录后刷新到单个S3对象，搭配buffer.memory控制内存缓冲区上限（缓冲区满时也会触发写入）。示例配置：

  flush.size=1000          # 累计1000条记录就写入一个S3文件
  buffer.memory=67108864   # 内存缓冲区达64MB时强制写入
  batch.size=500           # 每次从Kafka拉取500条记录，提升批量处理效率
  

• 按时间间隔批量写入
  用rotate.interval.ms设置每隔多久生成一个新的S3文件，合并该时间段内的所有记录。示例配置：

  rotate.interval.ms=300000  # 每5分钟生成一个S3文件
  

• 混合触发（任一条件满足即写入）
  同时配置flush.size和rotate.interval.ms，只要达到记录数阈值或时间阈值，就会把内存中的所有记录合并到单个S3对象中，之后开启新的文件。

额外注意点：

• 确保使用默认的storage.class=io.confluent.connect.s3.storage.S3Storage，它原生支持文件滚动合并功能。
• 处理CDC数据时，选择合适的format.class（比如io.confluent.connect.s3.format.json.JsonFormat），保证多条CDC记录能正确序列化到同一文件，避免格式错误。
• 你之前关注的fetch.min.bytes和fetch.max.wait.ms只是控制Connector从Kafka拉取数据的时机，和S3文件的合并逻辑无关，无需调整这些参数。

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