如何用KSQL将同一ENRLMT_ID的数据流聚合为数组？

需求说明 #

作为KSQL新手，需要将流中同一ENRLMT_ID的所有数据分组，聚合为包含数组的JSON负载。

输入数据示例 #

源流ENROLLMENT_STATUS_DATA中的对象示例：

{
  "ENRLMT_ID": "I12345",
  "STUS_CD": "06",      
  "STUS_RSN_CD": "081",
  "STUS_RSN_DESC": "APPROVED FOR REVALIDATION",
  "STUS_DESC": "APPROVED"  
}

{
  "ENRLMT_ID": "I12345",
  "STUS_CD": "13",
  "SRC_ENRLMT_STUS_HSTRY_SK": "OxdP6jOQnr/o+UfE4q0zr5p7lMvK0Fh9N",
  "STUS_RSN_CD": "029",
  "STUS_RSN_DESC": "THE PROVIDER OR SUPPLIER IS VOLUNTARILY WITHDRAWING",
  "STUS_DESC": "DEACTIVATED"  
}

期望输出 #

按ENRLMT_ID聚合后的结果：

{     
  "ENRLMT_ID": "I12345",
  "PAYLOAD": [ 
    {
      "STUS_CD": "06",              
      "STUS_RSN_CD": "081",
      "STUS_RSN_DESC": "APPROVED FOR REVALIDATION",
      "STUS_DESC": "APPROVED"
    },
    {
      "STUS_CD": "13",              
      "STUS_RSN_CD": "029",
      "STUS_RSN_DESC": "THE PROVIDER OR SUPPLIER IS VOLUNTARILY WITHDRAWING",
      "STUS_DESC": "DEACTIVATED"  
    }
  ]
}

当前代码问题 #

你编写的CREATE STREAM语句仅将每条数据转换为单个结构体，未实现分组聚合逻辑——流处理默认逐条处理，不会维护分组状态。

正确实现方案 #

要实现分组聚合，需使用CREATE TABLE（维护状态存储聚合结果）和COLLECT_LIST函数（将分组内的结构体收集为数组），具体语句如下：

CREATE TABLE ENROLLMENT_STATUS_AGG AS
SELECT
  ENRLMT_ID,
  COLLECT_LIST(STRUCT(
    STUS_CD := STUS_CD,
    STUS_RSN_CD := STUS_RSN_CD,
    STUS_RSN_DESC := STUS_RSN_DESC,
    STUS_DESC := STUS_DESC
  )) AS PAYLOAD
FROM ENROLLMENT_STATUS_DATA
GROUP BY ENRLMT_ID;

关键说明 #

• 使用CREATE TABLE：聚合操作是有状态的，表会持续维护每个ENRLMT_ID的最新聚合结果，当源流中有新数据加入时，对应分组的数组会自动更新。
• COLLECT_LIST函数：将同一ENRLMT_ID下的所有目标字段结构体收集为数组，生成期望的PAYLOAD字段。
• 忽略无关字段：源流中未包含在STRUCT中的字段（如SRC_ENRLMT_STUS_HSTRY_SK）会自动被排除，符合输出需求。

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