使用表白名单选项更新Debezium MySQL连接器

Question

我正在使用Debezium (0.7.5) MySQL连接器，如果我想用选项table.whitelist更新这个配置，我想了解最好的方法是什么。

假设我创建了一个连接器，如下所示：

curl -i -X POST -H "Accept:application/json" -H  "Content-Type:application/json" http://debezium-host/connectors/ -d '
{
  "name": "MyConnector",
  "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "connect.timeout.ms": "60000",
      "tasks.max": "1",
      "database.hostname": "myhost",
      "database.port": "3306",
      "database.user": "***",
      "database.password": "***",
      "database.server.id": "3227197",
      "database.server.name": "MyServer",
      "database.whitelist": "myDb",
      "table.whitelist": "myDb.table1,myDb.table2",
      "database.history.kafka.bootstrap.servers": "kb0:9092,kb1:9092,kb2:9092",
      "database.history.kafka.topic": "MyConnectorHistoryTopic",
      "max.batch.size": "1024",
      "snapshot.mode": "initial",
      "decimal.handling.mode": "double"
    }
}'

一段时间后(2周)，我需要在这个table.whitelist选项中添加一个新表(table.whitelist选项)(这个表是旧的，它是在连接器之前创建的)

我试过的是：

• 暂停连接器。
• 删除了历史主题(也许这就是问题所在)。
• 通过API更新配置端点更新配置。
• 恢复连接器。

通过API更新命令：

curl -i -X PUT -H "Accept:application/json" -H  "Content-Type:application/json" https://kafka-connect-host/connectors/MyConnector/config/ -d '
{
  "connector.class": "io.debezium.connector.mysql.MySqlConnector",
  "connect.timeout.ms": "60000",
  "tasks.max": "1",
  "database.hostname": "myhost",
  "database.port": "3306",
  "database.user": "***",
  "database.password": "***",
  "database.server.id": "3227197",
  "database.server.name": "MyServer",
  "database.whitelist": "myDb",
  "table.whitelist": "myDb.table1,myDb.table2,myDb.table3",
  "database.history.kafka.bootstrap.servers": "kb0:9092,kb1:9092,kb2:9092",
  "database.history.kafka.topic": "MyConnectorHistoryTopic",
  "max.batch.size": "1024",
  "snapshot.mode": "schema_only",
  "decimal.handling.mode": "double"
}'

但是它没有起作用，也许这根本不是最好的方法。在其他连接器中，我没有使用table.whitelist选项，所以当我需要侦听一个新表时，我没有这个问题。

我的最后一个选项，我认为应该删除这个连接器，并创建另一个具有这个新配置的连接器，它也侦听新表(myDb.table3)。问题是，如果我想要从myDb.table3获得初始数据，就必须使用快照initial创建这些数据，但我不想从其他表myDb.table1,myDb.table2中生成所有的快照消息。

Answer 1

此时还不支持对白名单/黑名单配置的更改。目前正在进行这方面的工作(请参阅DBZ-175)，我们希望在下一个版本中对此提供预览支持。这里有一个待定公关，但是它需要更多的工作。

在实现这一点之前，最好的选择是设置一个连接器的新实例，它只捕获您感兴趣的其他表。这是以运行两个连接器(这两个连接器都将维护一个binlog读取器会话)为代价的，但只要您不需要频繁地更改筛选器配置，它就能做到这一点。

Answer 2

Debezium Server的最新版本，您可以添加以下配置

debezium.snapshot.new.tables=parallel

如果您正在使用Debezium，您可以尝试这个配置值。

snapshot.new.tables=parallel

注意: Debeziyum是支持Kinesis、Google和Apache的服务器。我正在使用它，它的配置有点不同。我必须在每个项目之前加上"debezium“。

一旦添加了此配置，任何添加到tables.whitelist中的附加表Debezium都会创建快照。

我不能向您指出文档，但是我在GitHub中查看了它们的代码，并且我也实际尝试了它，这对我来说是有效的。下面是指向MySqlConnector代码的链接

https://github.com/debezium/debezium/blob/master/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/MySqlConnectorConfig.java

搜索Field.create("snapshot.new.tables")

就我个人而言，我觉得Debezium有很多东西，但是文档是分散的。

Answer 3

我也有同样的问题，用信号表来解决。它的工作方式是，您必须在datatable中创建一个要发送到debezium命令的表。

CREATE TABLE public.debezium_signal (id VARCHAR(42) PRIMARY KEY, type VARCHAR(32)  NULL, data VARCHAR(2048)  NULL);

并在您的配置中设置一个标记"signal.data.collection": "public.debezium_signal"

之后，您可以在该表中发送带有insert的命令：

INSERT INTO debezium_signal (id, type, data)
VALUES(gen_random_uuid(),'execute-snapshot','{"data-collections": "myDb.table3"]}');

在我的例子中，我必须在table.include.list中添加de信号，在column.include.list中也添加列。

https://debezium.io/documentation/reference/stable/configuration/signalling.html