KSQL多键表创建后，能否通过Sink连接器将键值写入Postgres？

1. 修正KSQL建表语句 #

你的原语句存在语法问题：GROUP BY子句中重复了mcc，且字段名带多余空格，会导致键定义混乱。修正后的语句：

CREATE TABLE base_mcc 
 WITH (KEY_FORMAT='AVRO',VALUE_FORMAT='AVRO', PARTITIONS=2)
 AS
SELECT mcc, cnt_id, SUM(`amount`) AS `sum_amnt` 
FROM STREAM 
WINDOW TUMBLING (SIZE 90 DAYS)
GROUP BY mcc, cnt_id
EMIT CHANGES;

修正后，Kafka消息的Key会包含mcc和cnt_id两个字段（Avro格式），Value仅包含sum_amnt。

2. 配置Postgres Sink连接器 #

通过连接器的Transform功能，将Key中的字段合并到Value中，确保所有字段能被写入Postgres。示例配置如下：

name=postgres-sink-base-mcc
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
connection.url=jdbc:postgresql://你的Postgres地址:5432/数据库名
connection.user=用户名
connection.password=密码
topics=BASE_MCC  # KSQL生成的Topic名称，默认是表名大写
key.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://你的Schema Registry地址:8081
value.converter=io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url=http://你的Schema Registry地址:8081
insert.mode=upsert  # 聚合表适合用upsert模式，避免重复数据
pk.fields=mcc,cnt_id  # 指定Postgres表的主键为组合键
pk.mode=record_key
auto.create=true  # 自动创建Postgres表（也可手动创建）
auto.evolve=true
# 关键：将Key中的mcc、cnt_id提取到Value中
transforms=extractKey
transforms.extractKey.type=org.apache.kafka.connect.transforms.ValueFromKey
transforms.extractKey.fields=mcc,cnt_id

3. 可选：手动创建Postgres表 #

如果关闭auto.create，可以手动创建匹配的表结构：

CREATE TABLE base_mcc (
    mcc VARCHAR(10) NOT NULL,
    cnt_id VARCHAR(20) NOT NULL,
    sum_amnt NUMERIC NOT NULL,
    PRIMARY KEY (mcc, cnt_id)
);

配置完成后，Sink连接器会自动将Kafka消息中的键（mcc、cnt_id）和值（sum_amnt）一同写入Postgres表。

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