是否存在可将Kafka主题内容持久化为CSV文件的Sink Connector？

嘿，我刚好踩过类似的坑，给你整理几个能直接用或者低成本实现的方案：

1. 社区成熟项目：kafka-connect-csv #

这个项目是专门为CSV输出打造的Sink Connector，已经实现了完整的任务逻辑，不是那种半吊子的半成品。它支持：

• 自定义CSV分隔符、引号规则
• 按主题、时间窗口拆分输出文件
• 配置字段映射，把Kafka消息的键/值字段映射到CSV列
• 支持Avro、JSON等常见消息格式的解析

举个简单的配置示例：

name=csv-sink-connector
connector.class=io.github.jcustenborder.kafka.connect.csv.CsvSinkConnector
tasks.max=1
topics=your-input-topic
csv.directory=/path/to/output/csvs
csv.file.name.format=${topic}-${timestamp:yyyyMMdd}.csv
csv.fields=id,name,value
csv.field.mappings=id=key.id,name=value.name,value=value.amount

2. 扩展Confluent FileSink Connector #

如果不想引入第三方依赖，你可以基于Confluent的FileStreamSinkConnector做轻量扩展：

• 自定义一个Format实现类，重写write方法，把Kafka的SinkRecord转换成CSV格式的字符串
• 把这个类打包成JAR，放到Kafka Connect的插件目录
• 在Connector配置里指定format.class为你自定义的类

这种方式适合有一定开发能力的场景，能完全掌控CSV的生成逻辑。

3. 用Transform + FileSink做无代码配置 #

如果你的消息结构本身比较扁平，也可以用Kafka Connect自带的Transforms先把消息转换成CSV字符串，再用FileStreamSinkConnector输出：

• 用ExtractField提取消息里的字段，再用ReplaceField重命名或筛选字段
• 用Transform的拼接逻辑把字段组合成CSV格式的字符串
• 最后用FileSink输出到文件

这种方式不需要写代码，但只适合结构简单的消息，复杂嵌套结构处理起来会比较麻烦。

内容的提问来源于stack exchange，提问作者spi-x-i