如何用Flink SQL展开数组关联表后聚合为单个JSON数组？

问题场景 #

现有两个Kafka源表：

• Table1数据示例：

{
"id": "id-1",
"segs": ["seg-id-1", "segs-id-2"]
}

• Table2数据示例：

{
"id": "seg-id-1",
"name": "segs-name-1"
}

通过自定义UDFjson_string_array_to_array解析数组并关联表后，创建视图FINAL_UNESTED_WITH_SEGMENTS，再执行分组聚合插入：

CREATE VIEW IF NOT EXISTS FINAL_UNESTED_WITH_SEGMENTS AS (
SELECT 
       id,
       unnested_segments_id as segs_id,
       sc.name as segs_name
from table1 r CROSS JOIN unnest(json_string_array_to_array(segs)) AS SegmentContentTable (unnested_segments_id)  
         LEFT JOIN table2 sc on (unnested_segments_id = sc.id));

insert into final_table 
select id, json_agg_custom('seg_id', unnested_segments_id, 'seg_name', segs_name) as segs
from FINAL_UNESTED_WITH_SEGMENTS group by id;

当前输出两条Kafka消息：

{
"id": "id-1",
"segs": [{"id":"seg-id-1", "name":"segs-name-1"}]
}

{
"id": "id-1",
"segs": [{"id":"seg-id-2", "name":null}]
}

期望输出单条合并后的消息：

{
  "id": "id-1",
  "segs": [
    {
      "id": "seg-id-1",
      "name": "segs-name-1"
    },
    {
      "id": "seg-id-2",
      "name": null
    }
  ]
}

原因分析 #

Flink流式处理中，默认的分组聚合是增量聚合：每处理一条关联后的数据，就会输出一次当前的聚合结果。当seg-id-1处理完成时输出一次，seg-id-2处理完成时再输出一次，因此得到两条同id的消息。要得到最终合并结果，需要等待该id下所有关联数据处理完成后再输出，或通过更新语义保留最新结果。

解决方案 #

方案1：使用事件时间窗口聚合（适合可接受延迟的场景） #

如果源表包含事件时间字段（比如Table1有event_time字段，类型为TIMESTAMP(3)），可以基于事件时间创建窗口，等待窗口关闭后输出最终聚合结果：

INSERT INTO final_table
SELECT 
    id, 
    json_agg_custom('seg_id', unnested_segments_id, 'seg_name', segs_name) AS segs
FROM FINAL_UNESTED_WITH_SEGMENTS
GROUP BY 
    id,
    TUMBLE(event_time, INTERVAL '5' MINUTE); -- 窗口大小根据业务实际情况调整

窗口关闭后，Flink会输出该窗口内该id对应的完整聚合结果，确保只生成一条包含所有segs的消息。

方案2：使用Upsert Kafka Sink（适合实时更新场景） #

如果下游支持Upsert语义，可将Kafka Sink配置为Upsert模式，让Flink输出更新消息，下游自动覆盖旧数据，最终保留最新的完整聚合结果。

1. 先创建支持Upsert的final_table：

CREATE TABLE final_table (
    id STRING PRIMARY KEY NOT ENFORCED, -- 主键用于匹配更新
    segs STRING
) WITH (
    'connector' = 'upsert-kafka',
    'topic' = 'your_target_topic',
    'properties.bootstrap.servers' = 'your_kafka_broker',
    'key.format' = 'json',
    'value.format' = 'json'
);

2. 执行原有的插入语句即可：

insert into final_table 
select id, json_agg_custom('seg_id', unnested_segments_id, 'seg_name', segs_name) as segs
from FINAL_UNESTED_WITH_SEGMENTS group by id;

Flink会先发送DELETE消息删除旧的聚合结果，再发送INSERT消息写入新的完整结果，下游消费者最终只会保留最新的单条合并消息。

补充：状态管理 #

无论使用哪种方案，都建议配置状态TTL（Time-To-Live），避免Flink状态无限增长：

-- 在视图或表上配置状态TTL
CREATE VIEW IF NOT EXISTS FINAL_UNESTED_WITH_SEGMENTS WITH (
    'state.ttl' = '1 hour' -- 状态1小时后自动清理
) AS (
-- 原视图SQL内容
);

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