使用Avro Schema创建KSQLDB Stream时遭遇VALUE_FORMAT与VALUE_SCHEMA_ID兼容错误求助

我来帮你搞定这个报错！首先咱们先弄明白为什么会出现这个错误：

当你使用VALUE_SCHEMA_ID参数时，ksqlDB要求你的VALUE_FORMAT必须是支持Schema Registry集成的格式——毕竟VALUE_SCHEMA_ID是用来指定Schema Registry中已注册的schema ID的，而普通的JSON格式本身和Schema Registry没有绑定关系，没法通过ID关联到对应的schema，所以才会抛出"VALUE_FORMAT should support schema inference when VALUE_SCHEMA_ID is provided"的错误。

针对你的场景（Kafka主题里是JSON数据，但你已经在Schema Registry注册了Avro schema），解决方案很明确，只需要调整两个地方：

1. 替换VALUE_FORMAT为JSON_SR #

JSON_SR是ksqlDB专门设计的、用于结合Schema Registry解析JSON数据的格式，它原生支持通过VALUE_SCHEMA_ID指定要使用的schema。

2. 处理字段名大小写不匹配的问题 #

你的Avro schema里字段都是小写（比如name、height），但Kafka主题里的JSON数据字段是大写开头（比如Name、Height），这时候需要加上VALUE_FIELD_CASE='LOWERCASE'参数，让ksqlDB自动把JSON字段名转成小写来匹配schema。

修改后的创建流命令 #

CREATE STREAM customer_stream WITH (
  KAFKA_TOPIC='customers',
  VALUE_FORMAT='JSON_SR',
  VALUE_SCHEMA_ID=1,
  VALUE_FIELD_CASE='LOWERCASE'
);

验证结果 #

创建完成后，你可以用下面的命令查看数据是否正确解析：

SELECT * FROM customer_stream EMIT CHANGES LIMIT 2;

额外注意事项 #

• 确保ksqlDB的配置文件里已经正确设置了ksql.schema.registry.url，指向你的Schema Registry地址，这样ksqlDB才能获取到ID=1的Avro schema。
• 如果你的字段名大小写本来就完全匹配，那VALUE_FIELD_CASE参数可以省略。

内容的提问来源于stack exchange，提问作者Ola Hällvall