咱先拆解下你当前遇到的核心问题：flush.size=200000（按记录数触发文件滚动）和rotate.interval.ms=600000（10分钟强制滚动）是逻辑或的关系——满足任意一个条件就会生成新文件。这直接导致了两种极端情况：

• 消息高峰时，10分钟内凑够20万条记录，生成大文件（比如30MB）；
• 消息低谷时，10分钟内没到20万条，到点强制滚动，生成小文件（比如6KB）。

再加上你用了自定义的字段+时间分区器，如果分区维度（时间粒度、字段值）的流量分布不均，会进一步放大文件大小的波动。结合这些情况，给你几个针对性的优化方向：

1. 用字节数替代记录数控制文件大小 #

flush.size按记录数触发，但每条消息的大小可能差异极大（从几十字节到几KB都有可能），这是文件大小波动的核心原因之一。如果你的连接器版本支持（多数新版Confluent S3连接器都支持），建议改用flush.size.bytes按字节数精准控制：

# 比如目标文件大小设为10MB，可根据你的需求调整
flush.size.bytes=10485760
# 保留时间兜底，但调长让字节数条件优先触发
rotate.interval.ms=3600000  # 1小时

这样不管消息条数多少，只要文件字节数接近目标值就滚动，能大幅缩小文件大小的差异。

2. 匹配flush与rotate参数的触发节奏 #

如果必须用记录数控制，先统计你10分钟内的最小/平均/最大消息量：

• 假设10分钟内最少有5万条消息，平均15万条，最多30万条
• 把flush.size设为略高于最小消息量的数值（比如6万），同时把rotate.interval.ms调小（比如5分钟）
• 这样即使在消息低谷，5分钟内也能凑够接近flush.size的记录数，避免生成过小的文件；高峰时也不会因为记录数超标生成过大的文件

3. 优化自定义分区器的粒度 #

你的自定义分区器如果是按时间+字段分区，检查这两个维度的分布：

• 时间粒度：如果按分钟分区，低峰时段的分钟可能只有几条消息，建议把时间粒度调粗（比如按15分钟/小时分区），让每个分区内的消息量更均衡；
• 字段分布：如果用来分区的字段（比如用户ID、业务类型）存在明显冷热差异（某些值的消息量特别大，某些特别小），可以考虑：
  • 合并低流量的字段值（比如把小于N条的归为一个“其他”分区）；
  • 改用哈希后的字段值作为分区键，让流量分布更均匀。
    另外，确保分区器在切换分区时不会强制生成空文件或过小的文件——比如只有当当前分区有消息时才触发滚动。

4. 启用压缩缩小大小波动 #

开启消息压缩能让不同内容的消息大小更均匀，缩小文件尺寸的波动范围：

compression.type=gzip  # 或snappy、lz4，根据性能需求选择

压缩后的文件大小受原始内容的影响会降低，即使原始消息大小波动大，最终生成的文件尺寸会更接近目标值。

5. 监控并动态调整参数 #

建议采集Kafka Connect的监控指标（比如kafka_connect_sink_record_count_total按分区维度的统计），跟踪每个分区的消息量变化：

• 针对高流量分区，适当调小flush.size.bytes或flush.size；
• 针对低流量分区，调大rotate.interval.ms，让文件有足够时间积累到目标大小；
  如果你的架构支持，甚至可以用动态配置工具实时调整不同分区的参数。

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