使用Kafka Connect在MySQL数据库之间同步数据

Question

我正在尝试使用基于Kafka Connect的Confluent在几个MySQL数据库之间同步数据。我在源连接器配置中使用了“批量”作为模式，因为主键类型是 varchar，所以我无法使用递增模式。它工作正常，但我遇到了两个问题：

1. 似乎它无法同步删除，当源数据库中的数据被删除时，接收器数据库没有任何变化。数据仍存在于接收器数据库中。
2. 同步数据需要相当长的时间。就我而言，同步具有 2~4k 行的表大约需要 4~4 分钟。我可以理解使用批量模式可能会花费更多时间来同步数据，但这不是太长了吗？

这是我的源连接器配置:

name=test-source
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:mysql://xxx.xxx.xxx:3306/xxx?useUnicode=true&characterEncoding=utf8
connection.user=user
connection.password=password
mode=bulk
table.whitelist=a_table

这是我的接收器连接器配置：

name=test-sink
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1 topics=a_table
connection.url=jdbc:mysql://xxx.xxx.xxx.xxx:3306/xxx?useUnicode=true&characterEncoding=utf8
connection.user=user
connection.password=password
insert.mode=upsert
pk.mode=record_value
pk.fields=mypk
auto.evolve=true

任何建议都将受到重视。谢谢你。

Answer 1

>

如果要同步删除操作，则需要使用 CDC，例如 Debezium。JDBC 连接器只能检测存在的记录，而不能检测不存在的记录。

CDC也比批量获取更有效，因为它监视MySQL事务日志中所需表上的任何事务。

您的主键是 VARCHAR？哇。如果您不想使用 CDC，我建议使用基于 INT 的密钥，然后使用 JDBC 连接器进行增量加载。或者向表中添加时间戳列，并将其用于增量。