Kafka 连接 JDBC 接收器连接器 - 没有主键的表答案

【问题标题】：Kafka connect JDBC sink connector - tables without Primary KeyKafka 连接 JDBC 接收器连接器 - 没有主键的表
【发布时间】：2020-08-08 00:26:06
【问题描述】：

我想将 100 多个表从旧 MySQL 服务器复制到 PostgreSQL。我已经设置了运行良好的 debezium。现在我想为 PostgreSQL 设置一个 JDBC 接收器连接器。我还需要启用 DELETE。

在这种情况下，如何为没有主键的表配置接收器连接器？

它应该复制插入、更新和删除。

【问题讨论】：

哪里说 debezium 需要主键？
Debezium 方面一切正常。一旦 Kafka 代理中的数据，从那里我将使用 JDBC SINK ，那个 SINK 连接器需要 PK
好的。那么，您必须使用 Connect 吗？
我什么都可以，只是想将 MySQL 与 postgresql 同步（插入、更新、删除）
只要您的源表上有 any 唯一键，Debezium 将在没有真正的 PK 列的情况下自动将其用作消息键，或者如果未明确定义，您可以配置应形成消息键的列。我想在接收端需要它来确保 upsert 语义。

标签： apache-kafka apache-kafka-connect confluent-platform debezium

【解决方案1】：

我使用过 Debezium 和 Kafka Connect 2 Postgres。我指定了键列pk.mode=record_key和pk.fields=name_of_pk_column，因为Kafka Connect需要，所以可以删除（enable.delete =true) 并更新。你也可以设置 auto.evolve=true。

如果将 enable.delete 设置为 true，则必须指定 PK。源表和目标表中的列键必须相同。

但如果您需要传输多个表，这并不方便。

我没有尝试过，但我认为单个消息转换脚本可能会从 kafka 消息中获取密钥，对其进行转换（可能重命名）并在 kafka_connect_postgres 设置中使用它。

我实际上两边都有 1:1 的数据库，所以获取 pk 列名我只需要担心。

如果您愿意，我可以提供我的 Kafka connect 设置、Kafka pg connect...

connector-pg.properties
name=sink-postgres
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
#tasks.max=2
value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=true
topics=pokladna.public.books #my debezium topic with CDC from source DB
connection.url=jdbc:postgresql://localhost:5434/postgres
connection.user=postgres
connection.password=postgres
dialect.name=PostgreSqlDatabaseDialect
table.name.format=books_kafka #targe table in same db, in thus case given by url
transforms=unwrap
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones=false
auto.create=true
auto.evolve=true
insert.mode=upsert #insert does not work for updates
delete.enabled=true
pk.fields=id #pk column name same in target and resource db
pk.mode=record_key

尝试使用SMT extract field，但对我来说还没有用。我仍然不知道如何自动从 kafka 消息中提取主键。也许我必须编写自己的 SMT...

实际上我要开始一个关于它的新 SO 主题。哈哈

【讨论】：

好的，我会在单独的答案中发布。