KSQL滚动窗口计数写入PostgreSQL遇主键冲突，如何实现更新？

核心问题 #

1. KSQL查询缺失主键依赖字段：你的Sink配置指定pk.fields="REPORT_TIME,WINDOWSTART"，但原KSQL查询并未输出WINDOWSTART字段，导致连接器无法从消息value中获取该主键字段，主键配置完全失效。
2. 主键选择冗余且不合理：REPORT_TIME是WINDOWSTART的字符串格式转换值，二者本质是同一时间标识，组合作为主键完全冗余；窗口统计的唯一标识应该是分组键+窗口起始时间（即X+WINDOWSTART），而非时间的两种格式组合。
3. UPSERT逻辑未正确触发：主键配置错误导致连接器无法识别重复的窗口统计记录，因此执行插入而非更新操作，引发主键冲突。

修复步骤 #

1. 修改KSQL查询，补充必要字段 #

更新查询语句，将WINDOWSTART字段包含在输出结果中，确保Sink连接器能获取到窗口唯一标识：

CREATE TABLE daily_msg_count_tbl WITH (
    KAFKA_TOPIC='daily_msg_count_all',
    PARTITIONS=1,
    REPLICAS=3,
    VALUE_FORMAT='AVRO'
) AS 
SELECT
    'X' AS X,
    COUNT(*) AS DAILY_MSG_COUNT_ALL,
    windowstart AS WINDOWSTART,
    TIMESTAMPTOSTRING(windowstart, 'yyyy-MM-dd HH:mm:ss') AS REPORT_TIME
FROM data-stream 
WINDOW TUMBLING (SIZE 24 HOURS)
GROUP BY 'X'
EMIT CHANGES;

2. 调整Sink连接器配置 #

修正主键配置，使用X+WINDOWSTART作为唯一主键，确保UPSERT逻辑正常触发：

{
    "name":"daily_msg_count_all_sink",
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "connection.url":"db url",
    "connection.user":"",
    "connection.password":"", 
    "topics":"daily_msg_count_all",
    "auto.evolve":"true",
    "key.converter":"org.apache.kafka.connect.storage.StringConverter",
    "auto.create":"true",
    "insert_mode":"upsert",
    "value.converter":"io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url":"schema url",
    "pk.mode":"record_value",
    "pk.fields":"X,WINDOWSTART"
}

• 关键调整：将pk.fields改为X,WINDOWSTART，这两个字段唯一标识一个窗口的统计记录；
• 可选优化：若通过DESCRIBE daily_msg_count_tbl;发现KSQL生成的消息key已包含分组键和窗口信息，可将pk.mode改为record_key，无需手动指定pk.fields，配置更可靠。

3. 调整PostgreSQL表主键（可选） #

如果之前已自动创建了PostgreSQL表，其主键约束可能不符合新配置：

• 若允许删除旧表，直接删除后重启Sink连接器，让连接器自动创建带正确主键的新表；
• 若无法删除表，手动执行SQL修改主键约束：ALTER TABLE daily_msg_count_all ADD PRIMARY KEY(X, WINDOWSTART);

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