如何用ksqlDB展开多列表元素生成3条记录流？

问题背景 #

现有两条Kafka消息：一条是包含1个结构体元素的数组，另一条是包含2个结构体元素的数组。需要通过ksqlDB生成包含3条独立元素的流，使用EXPLODE函数时因原数据无键遇到异常，当前尝试的代码已给出。

核心问题 #

原始流未定义有效键，导致EXPLODE展开后的每条消息继承了原消息的空键，进而引发后续分区处理、聚合操作的异常。ksqlDB要求流消息具备键以保证分区逻辑和状态处理的正确性。

修正后的完整代码 #

-- 创建原始流，指定键格式（原消息无键时ROWKEY为null）
CREATE STREAM ml_stream_raw (
    data ARRAY<STRUCT<source VARCHAR, numericValue DOUBLE, created BIGINT, textValue VARCHAR>>
) WITH (
    KAFKA_TOPIC='testml1withlist',
    VALUE_FORMAT='JSON',
    KEY_FORMAT='KAFKA'
);

-- 展开数组并生成唯一键，确保每条消息有独立分区键
CREATE STREAM ml_stream_raw_exploded AS
    SELECT 
        UUID() AS rowkey,
        EXPLODE(data) AS message
    FROM ml_stream_raw
    PARTITION BY UUID();

-- 提取结构体字段到处理后的流
CREATE STREAM ml_stream_raw_processed AS
    SELECT 
        rowkey,
        message->source AS source,
        message->numericValue AS numericValue,
        message->created AS created,
        message->textValue AS textValue
    FROM ml_stream_raw_exploded;

-- 按1秒滚动窗口聚合消息数量
CREATE TABLE ml_table_messages_per_second_list AS
    SELECT 
        'all_messages' AS message_group,
        WINDOWSTART AS window_start,
        COUNT(*) AS message_count
    FROM ml_stream_raw_processed
    WINDOW TUMBLING (SIZE 1 SECOND)
    GROUP BY 'all_messages';

-- 查询并格式化聚合结果
SELECT 
    TIMESTAMPTOSTRING(window_start, 'dd.MM.yyyy HH:mm:ss') AS window_start_formatted,
    message_count
FROM ml_table_messages_per_second_list
EMIT CHANGES;

关键说明 #

1. 生成唯一键：通过UUID()函数为每条展开后的消息生成独立的rowkey，并以此作为分区键，避免无键导致的分区异常。
2. 替代方案：如果结构体中存在具备业务唯一性的字段（比如source与created的组合），可替换UUID()为该组合键，更贴合业务逻辑。
3. 键格式指定：在创建原始流时显式指定KEY_FORMAT='KAFKA'，明确键的处理规则。

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