KSQLDB：使用CREATE STREAM AS SELECT创建不同KEY SCHEMA的流

问题根源 #

1. 多次调用EXPLODE的问题：原语句中重复使用EXPLODE("responses")会导致KSQLDB无法正确关联数组元素，甚至可能产生笛卡尔积错误，同时也不符合KSQLDB的流处理语法规范。
2. 未明确指定新ROWKEY：KSQLDB要求创建流时必须明确指定键字段，原语句未将构造的新STRUCT标记为流的ROWKEY，导致报错Key missing from projection。

正确实现方式 #

方式1：直接通过CREATE STREAM ... AS SELECT完成转换

使用LATERAL TABLE EXPLODE实现数组行转列，同时显式构造新的ROWKEY并指定为流的键：

CREATE STREAM STREAM_DETAIL WITH (
    KAFKA_TOPIC = 'stream_detail',
    KEY_FORMAT = 'JSON', -- 根据实际需求选择序列化格式，如AVRO
    VALUE_FORMAT = 'JSON'
) AS 
SELECT 
    STRUCT(
        assessment_id := assessment_id,
        student_id := t.response_item->student_id,
        question_id := t.response_item->question_id
    ) AS ROWKEY,
    assessment_id,
    institution_id,
    t.response_item->student_id AS student_id,
    t.response_item->question_id AS question_id,
    t.response_item->response AS response
FROM STREAM_SUMMARY
LATERAL TABLE EXPLODE(responses) AS t(response_item)
EMIT CHANGES;

方式2：先手动定义流结构，再插入数据

如果需要提前明确流的Schema（比如使用AVRO并依赖Schema Registry），可以先创建流，再执行插入：

1. 创建目标流：

CREATE STREAM STREAM_DETAIL (
    ROWKEY STRUCT<assessment_id VARCHAR, student_id INTEGER, question_id INTEGER> KEY,
    assessment_id VARCHAR,
    institution_id INTEGER,
    student_id INTEGER,
    question_id INTEGER,
    response VARCHAR
) WITH (
    KAFKA_TOPIC = 'stream_detail',
    KEY_FORMAT = 'JSON',
    VALUE_FORMAT = 'JSON'
);

2. 插入转换后的数据：

INSERT INTO STREAM_DETAIL
SELECT 
    STRUCT(
        assessment_id := assessment_id,
        student_id := t.response_item->student_id,
        question_id := t.response_item->question_id
    ) AS ROWKEY,
    assessment_id,
    institution_id,
    t.response_item->student_id AS student_id,
    t.response_item->question_id AS question_id,
    t.response_item->response AS response
FROM STREAM_SUMMARY
LATERAL TABLE EXPLODE(responses) AS t(response_item);

关键说明 #

• LATERAL TABLE EXPLODE：将responses数组的每个元素拆分为独立行，确保每个学生-问题-响应组合对应唯一一行，避免重复展开的问题。
• 显式命名ROWKEY：通过AS ROWKEY标记构造的STRUCT为新流的键，替换原流的KEY SCHEMA，满足Kafka Sink Connector的UPSERT需求。
• 序列化格式：根据实际场景选择KEY_FORMAT和VALUE_FORMAT，如果使用AVRO，需确保Schema Registry正常运行，KSQLDB会自动注册新的键结构。

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