如何在KSQL中创建以结构体（struct）为键的流？

KSQL 5.3及以上版本支持将结构体作为流的键，前提是底层Kafka主题的消息键采用结构化格式（如Avro、JSON），以下是具体实现步骤：

1. 定义结构体类型（可选） #

如果该结构体类型会被多个流/表复用，可先创建自定义类型：

CREATE TYPE USER_KEY AS STRUCT(
    user_id VARCHAR,
    region VARCHAR
);

2. 创建带结构体键的流 #

有两种定义方式，根据需求选择：

方式一：内联结构体定义 #

直接在流定义中声明结构体键的结构，适用于仅当前流使用的场景：

CREATE STREAM user_events_stream (
    -- 声明结构体为键
    key STRUCT<user_id VARCHAR, region VARCHAR> KEY,
    -- 流的其他字段
    event_type VARCHAR,
    event_timestamp BIGINT
) WITH (
    KAFKA_TOPIC='user_events', -- 关联的Kafka主题
    VALUE_FORMAT='AVRO',       -- 值的序列化格式
    KEY_FORMAT='AVRO'          -- 键的序列化格式，必须与主题键格式匹配
);

方式二：使用预定义的结构体类型 #

复用之前创建的USER_KEY类型：

CREATE STREAM user_events_stream (
    key USER_KEY KEY,
    event_type VARCHAR,
    event_timestamp BIGINT
) WITH (
    KAFKA_TOPIC='user_events',
    VALUE_FORMAT='AVRO',
    KEY_FORMAT='AVRO'
);

3. 验证结构体键的有效性 #

执行查询确认键是否被正确解析：

SELECT key->user_id, key->region, event_type FROM user_events_stream LIMIT 10;

关键注意事项 #

• 必须指定KEY_FORMAT：KSQL依赖该参数解析键的格式，常用值为AVRO、JSON，如果使用Protobuf需对应配置
• 字段名匹配：结构体的字段名必须与Kafka主题键中的字段名完全一致，否则无法解析
• 结构体键的使用：结构体中的字段可直接用于分组、关联等操作，例如：

SELECT key->region, COUNT(*) 
FROM user_events_stream 
GROUP BY key->region;

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