使用JQ从Kafka Topic消费JSON并生成格式化CSV的行输出问题

我来帮你解决这个CSV输出的问题！你已经搞定了键值筛选，现在就差把结果转成每行一条的标准CSV格式对吧？

你可以直接在Kafka ConsoleConsumer的命令后管道到jq工具，它能轻松把JSON数据转换成每行一条、字段用逗号分隔的CSV格式，完美适配你的需求。

具体命令示例 #

假设你需要提取的三个目标键是field1、field2、field3（替换成你实际的键名即可），可以使用这条命令：

./kafka-run-class.sh kafka.tools.ConsoleConsumer --bootstrap-server kafka1.example.net:9092 --topic prod.example.v1 --max-messages 2 | jq -r '[.field1, .field2, .field3] | @csv'

关键参数解释： #

• -r：让jq输出原始字符串，避免给字段额外添加不必要的引号（如果你的字段包含逗号、空格等特殊字符，@csv会自动用引号包裹，保证CSV格式合规）
• [.field1, .field2, .field3]：将你需要的三个键值打包成一个数组
• | @csv：把数组直接转换成标准CSV格式，每条记录自动占一行

适配已有筛选逻辑 #

如果你已经有自己的jq筛选规则（比如只保留符合特定条件的记录），只需要在筛选后追加CSV转换逻辑就行。比如你之前的筛选命令是jq '. | select(.status == "valid")'，可以改成：

./kafka-run-class.sh kafka.tools.ConsoleConsumer --bootstrap-server kafka1.example.net:9092 --topic prod.example.v1 --max-messages 2 | jq -r '. | select(.status == "valid") | [.field1, .field2, .field3] | @csv'

无引号格式可选 #

如果你的字段不需要引号包裹，也可以把@csv换成join(",")，输出更简洁的格式：

./kafka-run-class.sh kafka.tools.ConsoleConsumer --bootstrap-server kafka1.example.net:9092 --topic prod.example.v1 --max-messages 2 | jq -r '[.field1, .field2, .field3] | join(",")'

比如用你提到的两条JSON记录测试，输出会是：

value1,100,desc-1
value2,200,desc-2

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