kSQLDB创建AVRO复合Key流失败：Key格式不匹配求助

问题原因 #

你创建ip_key流时，在WITH子句中仅指定了FORMAT='AVRO'，但该参数实际是指定Value的格式，Key的格式默认使用KEY_FORMAT='KAFKA'（即Kafka原生二进制结构化格式）。因此即便用STRUCT构建了复合Key，KSQL仍会用Kafka默认格式序列化Key，导致print命令无法识别为AVRO格式，最终显示为二进制数据。

修正方案 #

修改ip_key流的创建语句，明确指定KEY_FORMAT='AVRO'，同时显式声明VALUE_FORMAT='AVRO'以避免歧义。另外需确保Schema Registry服务正常运行，因为AVRO格式依赖它管理Schema。

修正后的创建语句 #

CREATE STREAM ip_key
WITH (
    KAFKA_TOPIC='locator.key',
    PARTITIONS=4,
    REPLICAS=4,
    KEY_FORMAT='AVRO',
    VALUE_FORMAT='AVRO'
) AS
SELECT 
    STRUCT("ip" := ip, "uaid" := uaid) AS KEY,
    AS_VALUE(ip) AS IP,
    AS_VALUE(uaid) AS uaid
FROM ip_locator
PARTITION BY STRUCT("ip" := ip, "uaid" := uaid)
EMIT CHANGES;

验证方法 #

重新创建流后，执行以下命令查看Key格式：

print 'locator.key' FROM BEGINNING LIMIT 1;

此时应能正确识别Key为AVRO格式，输出类似：

Key format: AVRO
Value format: AVRO
rowtime: 2022/04/4 11:13:33.131 Z, key: {"IP": "10.10.10.10", "UAID": "ff6204b2f1bf11eda05b0242ac120003"}, value: {"IP": "10.10.10.10", "UAID": "ff6204b2f1bf11eda05b0242ac120003"}, partition: 3

额外说明 #

• FORMAT是VALUE_FORMAT的简写，指定Key格式时必须显式设置KEY_FORMAT，否则默认使用KAFKA格式。
• 需确认KSQL已配置Schema Registry地址（通过ksql.schema.registry.url参数），否则AVRO格式的Key/Value无法正常序列化和反序列化。

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