如何解决Python通过Kafka向NiFi传输数据时ConvertRecord失败问题？

1. 修正Python发送的数据结构 #

你当前发送的JSON包含了Avro Schema的元信息（type、name、namespace、fields），但NiFi的JsonTreeReader期望直接读取业务数据本身，也就是你实际需要的{"Id": 0, "Value": 10.1}。修改Python脚本如下：

server_address = "foobar:1234"
producer = KafkaProducer(bootstrap_servers=server_address,
                         value_serializer=lambda v: json.dumps(v).encode('utf-8'))
# 仅发送目标业务数据
data = {"Id": 0, "Value": 10.1}
producer.send("outputs", data)
producer.flush()

注：用value_serializer可以简化JSON序列化逻辑，若数据无特殊类型（如日期、自定义对象），无需额外调用json_util.default。

2. 确认NiFi ConvertRecord的配置 #

确保JsonTreeReader和CSVRecordSetWriter的配置匹配业务数据：

• JsonTreeReader：
  • 关联你配置的AvroSchemaRegistry，选择包含Id和Value字段的Schema
  • 把Schema Access Strategy设为Use Schema from Schema Registry，并指定对应的Schema标识（名称或ID）
• CSVRecordSetWriter：
  • 同样关联同一个AvroSchemaRegistry的Schema
  • 按需配置CSV格式（如是否包含表头、分隔符），默认配置即可正常生成CSV

3. 验证流程 #

修改后重新发送数据，ConsumeKafka_2_6获取到正确的业务JSON后，ConvertRecord会基于Avro Schema解析数据并转换为CSV格式，最终通过PutFile完成保存。

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