KsqlDB中JDBC Sink Connector配置问题：MySQL同步至PostgreSQL

错误根因 #

当启用 delete.enabled=true 和 pk.mode=record_key 时，JDBC Sink 要求 Kafka 消息的 Key 必须是带 Schema 的 Struct 类型或原始类型（如整数、字符串），但 Debezium 默认输出的 Key 是无 Schema 的 HashMap（若使用普通 JSON 转换器），导致 Sink 无法识别主键以执行删除操作。

步骤1：确保 Debezium 源连接器输出带 Schema 的 Key/Value #

修改 Debezium MySQL Source Connector 配置，使用带 Schema 管理的转换器（Avro 或 JSON Schema），确保消息 Key 携带明确的 Schema 信息：

name=MYSQL_SOURCE
connector.class=io.debezium.connector.mysql.MySqlConnector
tasks.max=1
database.hostname=mysql
database.port=3306
database.user=debezium
database.password=dbz
database.server.id=184054
database.server.name=inventory
database.include.list=inventory
table.include.list=inventory.customers
# 配置 Avro 转换器（需 Schema Registry 服务）
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

步骤2：配置正确的 JDBC Sink Connector #

针对 PostgreSQL 配置 Sink，确保主键识别、删除操作正常执行：

name=POSTGRES_SINK
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
# 对应 Debezium 生成的 Topic
topics=inventory.customers
# PostgreSQL 连接信息
connection.url=jdbc:postgresql://postgres:5432/target_db?user=postgres&password=your_pwd
# 自动创建/更新表结构
auto.create=true
auto.evolve=true
# 启用删除操作
delete.enabled=true
# 基于消息 Key 识别主键
pk.mode=record_key
# 指定表的主键字段（替换为 customers 表实际主键，如 id）
pk.fields=id
# 配置与源一致的 Avro 转换器
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
# 启用 Upsert 模式（支持插入/更新）
insert.mode=upsert

备选：使用 JSON Schema 转换器（无需 Avro） #

若不想依赖 Avro，可改用 JSON Schema 转换器：

key.converter=org.apache.kafka.connect.json.JsonSchemaConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter=org.apache.kafka.connect.json.JsonSchemaConverter
value.converter.schema.registry.url=http://schema-registry:8081

1. 事件流处理层（KsqlDB 核心） #

利用 KsqlDB 对多业务域的 CDC 流进行关联、聚合，生成面向查询的物化视图：

• 关联多流生成聚合视图：

  -- 创建 customers 基础流
  CREATE STREAM customers_stream WITH (KAFKA_TOPIC='inventory.customers', VALUE_FORMAT='AVRO');
  
  -- 创建 orders 基础流
  CREATE STREAM orders_stream WITH (KAFKA_TOPIC='inventory.orders', VALUE_FORMAT='AVRO');
  
  -- 生成客户订单聚合表（含订单数量、最近订单时间）
  CREATE TABLE customer_order_summary AS
  SELECT 
    c.id AS customer_id,
    c.name,
    COUNT(o.id) AS total_orders,
    MAX(o.order_time) AS last_order_time
  FROM customers_stream c
  LEFT JOIN orders_stream o 
    ON c.id = o.customer_id
  GROUP BY c.id, c.name
  EMIT CHANGES;
  

• 保证事件顺序：通过 Debezium 自带的 source.timestamp 字段排序，避免乱序导致的聚合错误。

2. Sink 层分层设计 #

• 基础数据层：将原始 CDC 流（如 inventory.customers、inventory.orders）直接同步到 PostgreSQL 的原始表（如 customers_raw、orders_raw），用于数据溯源和灾备。
• 查询视图层：将 KsqlDB 生成的聚合表（如 customer_order_summary）同步到 PostgreSQL 的业务视图表（如 customer_order_summary_view），直接提供给查询服务使用。

3. 一致性与事务保障 #

• 启用 Kafka Connect 事务：在 Sink 配置中添加 transactions.enabled=true（需 Kafka 集群开启事务支持），确保多流同步的原子性。
• PostgreSQL 事务控制：针对聚合视图的写入，使用 PostgreSQL 的 INSERT ... ON CONFLICT DO UPDATE 语法，配合主键实现幂等性更新。

4. 扩展性优化 #

• 任务拆分：将不同业务域的 Sink 任务拆分，避免单任务处理多流导致的性能瓶颈。
• PostgreSQL 分区表：对大流量的聚合表（如订单流水）使用时间分区或主键分区，提升查询和写入性能。

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