请求修正Kafka Connect FileStreamSink Connector按时间戳分区配置或替代方案

你的配置无效的核心原因：

• TimestampRouter这个SMT（单消息转换）的作用是修改消息发送的目标Topic名称，但FileStreamSink Connector是直接将所有消费到的消息写入配置中指定的固定file路径，完全不依赖Topic名称，所以这个配置对实现文件分区没有任何作用。
• FileStreamSink是Kafka Connect提供的极简Sink Connector，本身不支持动态生成分区文件路径，只能写入单个固定文件。

可行替代方案 #

方案1：使用支持分区的第三方File Sink Connector #

比如Confluent提供的File Sink Connector，它原生支持按时间戳、字段等维度动态生成分区文件。配置示例：

name=partitioned-file-sink
connector.class=io.confluent.connect.file.FileSinkConnector
tasks.max=1
topics=your-kafka-topic
file.dir=/path/to/output/directory
# 生成带日期的文件名
filename.format=data-${timestamp:yyyy-MM-dd}.txt
# 配置时间分区器
partitioner.class=io.confluent.connect.file.partitioner.TimeBasedPartitioner
partitioner.timezone=UTC
partitioner.duration.ms=86400000  # 按天分区（24小时）
# 指定时间戳来源：用消息自带的Kafka记录时间戳
partitioner.field=kafka.record.timestamp
# 消息格式可根据需求调整（比如Json、Avro等）
format.class=io.confluent.connect.file.format.JsonFormat

方案2：自定义FileStreamSink实现分区逻辑 #

如果不想依赖第三方组件，可以基于FileStreamSink的源码修改，添加按时间戳分区的逻辑：

• 重写FileStreamSinkTask类的put方法
• 在方法中根据消息的时间戳（record.timestamp()）动态判断当前应写入的文件路径
• 自动创建当日文件并切换输出流，将消息写入对应文件

方案3：用Kafka Streams前置路由实现分区 #

通过Kafka Streams先将原Topic的消息按时间戳路由到不同的Topic，再用FileStreamSink分别写入对应文件：

1. 编写Kafka Streams应用，将原Topic的消息按yyyy-MM-dd格式的时间戳，路由到your-topic-yyyy-MM-dd这类命名的Topic
2. 为每个日期Topic配置一个独立的FileStreamSink，将file参数设为/path/to/output/directory/data-yyyy-MM-dd.txt

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