KSQLDB如何封装结构体数组实现跨Topic的流数据格式转换

原查询无法执行并非只有数组结构定义的问题，一共存在4处语法/逻辑错配：

• 流引用错位：前置定义的源读流是GENERAL_STREAM，但SELECT语句中引用了不存在的EARTH_DELIVERY_COURIER_PUDOPACKAGESTATUSUPDATED_0，同时过滤条件访问了不存在的嵌套字段payload->status，和源Topic的平铺JSON结构完全不匹配
• 嵌套结构构造逻辑缺失：原代码段仅写了空的类型占位，既没有为STRUCT内部字段定义数据类型，也没有将源字段的实际值映射到结构中，连基本的语法分隔都缺失
• 字段名不匹配：目标输出的数组元素字段为ordernumber、barcodenumber，原STRUCT定义使用的是源侧驼峰字段名，和预期输出格式不一致
• 函数误用：AS_VALUE(requestId)是用于将Kafka消息键反序列化为结构化值的函数，这里requestId是流值里的普通字段，直接取值即可，该写法属于多余且错误的调用

KSQL中如果要在查询中动态构造数组+嵌套结构，不能直接写DDL场景下的类型占位，必须通过内置函数完成值构造：

1. 先用STRUCT()函数构造单个嵌套对象，按照目标输出的字段名做值映射
2. 再用ARRAY()函数把构造好的嵌套对象包裹为数组类型
3. 修正FROM和WHERE的错误引用，回到和源Topic匹配的平铺字段判断逻辑

CREATE STREAM REDIRECTION_STREAM
WITH (
    partitions = 6,
    replicas = 3,
    kafka_topic = 'write-topic',
    value_format = 'json'
) AS
SELECT
    requestId,
    ARRAY(
        STRUCT(
            `ordernumber` := orderNumber,
            `barcodenumber` := deliveryBarcode
        )
    ) AS `packages`
FROM GENERAL_STREAM
WHERE status = 'APPROVED_BY_SUPERVISOR'
EMIT CHANGES;

兼容提示：如果使用的KSQL版本不支持STRUCT构造中的:=映射语法，可以替换为AS关键字实现重命名，写法为STRUCT(orderNumber AS ordernumber, deliveryBarcode AS barcodenumber)，最终输出效果完全一致。

执行修正后的查询后，写入write-topic的消息格式完全符合预期：

{
  "requestId": "1616516663000",
  "packages": [
    {
      "ordernumber": "01235656",
      "barcodenumber": "733998877"
    }
  ]
}

内容的提问来源于stack exchange，提问作者Sinan Türemiş