使用KQL基于NotificationId为列中剩余行追加字符串值

原始数据（Properties列解析后） #

{"State":"Received","NotificationId":"abc"}
{"State":"Queued","NotificationId":"abc"}
{"State":"Published","NotificationId":"abc", "Short": "tim"}

{"State":"Received","NotificationId":"def"}
{"State":"Queued","NotificationId":"def"}
{"State":"Published","NotificationId":"def", "Short": "mit"}

已经用parse_json()解析了Properties列，现在要实现：按NotificationId分组，把每组里State是Published那行的Short值，加到同组其他两行的JSON结构中。

解决方法 #

第一步：提取每个分组对应的Short值 #

先筛选出所有State为Published的行，按NotificationId分组，拿到每个分组对应的Short值：

WITH group_short AS (
  SELECT
    NotificationId,
    Short AS target_short -- 每个分组仅一条Published记录，直接取值即可
  FROM your_table
  WHERE State = 'Published'
)

第二步：关联原表并合并JSON #

将原表与临时表group_short关联，再用对应SQL引擎的JSON工具，把Short值追加到原JSON中：

以Snowflake为例：

SELECT
  *,
  OBJECT_INSERT(parse_json(Properties), 'Short', gs.target_short) AS updated_properties
FROM your_table t
JOIN group_short gs ON t.NotificationId = gs.NotificationId

以Spark SQL为例：

先把解析后的JSON转为结构化数据，追加字段后再转回JSON格式：

WITH parsed_data AS (
  SELECT
    *,
    from_json(Properties, 'State string, NotificationId string') AS parsed_json
  FROM your_table
),
group_short AS (
  SELECT
    NotificationId,
    Short AS target_short
  FROM parsed_data
  WHERE parsed_json.State = 'Published'
)
SELECT
  *,
  to_json(named_struct(
    'State', parsed_json.State,
    'NotificationId', parsed_json.NotificationId,
    'Short', gs.target_short
  )) AS updated_properties
FROM parsed_data t
JOIN group_short gs ON t.parsed_json.NotificationId = gs.NotificationId

注意事项 #

• 确保每个NotificationId分组仅存在一条State='Published'的记录；若存在多条，需补充逻辑指定取值规则（比如取最新生成的记录）；
• 不同SQL引擎的JSON操作函数存在差异，比如PostgreSQL用jsonb_set、MySQL用JSON_SET，需根据实际使用的工具调整函数。

内容的提问来源于stack exchange，提问作者Nikhil Babu Battula