KSQL创建带STRUCT类型KEY的STREAM时遇格式不支持问题

核心问题 #

你遇到的情况是：

1. 初始创建带STRUCT类型KEY的流时触发错误，因为默认KEY_FORMAT='KAFKA'不支持STRUCT类型；
2. 指定KEY_FORMAT='JSON'后流创建成功，但无消息流入——根源是实际Kafka消息的KEY格式和JSON不匹配：Lenses MQTT Source Connector用ByteArrayConverter输出的KEY是字符串化的Struct（如Struct{topic=m/bgmdev/af1/t5/test,id=0}），并非标准JSON格式，导致KSQL的JSON解析器无法识别，消息被静默丢弃。

解决方案 #

方法一：修改MQTT连接器配置，输出标准JSON格式KEY #

将连接器的KEY转换器改为JsonConverter，让它把Struct类型的KEY序列化为标准JSON字符串，这样KSQL就能用KEY_FORMAT='JSON'正确解析。

连接器配置示例：

name=mqtt-source-t10
connector.class=io.lenses.streamreactor.connect.mqtt.source.MqttSourceConnector
tasks.max=1
mqtt.hosts=tcp://mqtt-broker:1883
mqtt.topics=m/bgmdev/af1/t5/test
kafka.topic=t10
# 替换ByteArrayConverter为JsonConverter
key.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

修改后重新创建KSQL流（和你之前成功创建的语句一致）：

CREATE STREAM TEST51 (
  `topic` STRUCT<`topic` VARCHAR, id INT> KEY,
  val INT
) WITH (
  KAFKA_TOPIC='t10',
  PARTITIONS=1,
  KEY_FORMAT='JSON',
  VALUE_FORMAT='JSON'
);

执行SET auto.offset.reset = 'earliest';让KSQL重新消费历史消息，即可看到数据流入。

方法二：无法修改连接器时，先以STRING接收KEY再手动解析 #

如果不能调整连接器配置，可以先创建一个以STRING类型接收KEY的流，再用正则表达式提取字段并转换为STRUCT：

1. 创建基础流接收原始KEY：

CREATE STREAM TEST50_RAW (
  `key_str` STRING KEY,
  val INT
) WITH (
  KAFKA_TOPIC='t10',
  PARTITIONS=1,
  KEY_FORMAT='KAFKA',  -- 以RAW字节格式接收KEY，转成STRING
  VALUE_FORMAT='JSON'
);

2. 解析STRING KEY为STRUCT：

CREATE STREAM TEST51 AS
SELECT
  STRUCT(
    `topic` => REGEXP_EXTRACT(key_str, 'topic=(.*?),id=', 1),
    id => CAST(REGEXP_EXTRACT(key_str, 'id=(\\d+)', 1) AS INT)
  ) AS `topic`,
  val
FROM TEST50_RAW
EMIT CHANGES;

辅助排查配置 #

如果仍有问题，可调整KSQL配置排查：

• SET auto.offset.reset = 'earliest';：让KSQL重新消费历史消息
• SET log.consumer.record.processing.errors = 'WARN';：开启解析错误的警告日志，方便定位问题

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