Kafka Connect 和 Debezium（PostgreSql 和 MongoDB）中的增量更新问题答案

【问题标题】：Problem with Incremental updates in Kafka Connect and Debezium (PostgreSql and MongoDB)Kafka Connect 和 Debezium（PostgreSql 和 MongoDB）中的增量更新问题
【发布时间】：2022-02-04 19:03:30
【问题描述】：

我们有一个使用 Kafka 和 Zookeeper 设置的环境。我们需要通过 Kafka Connect 将数据从 PostgreSQL 数据库发送到 MongoDB 数据库。我们能够执行配置，并且在初始化连接器时，数据最初是正确迁移的。

但是，我们需要在连接器配置中添加新列或表时，再次发送受影响的记录以在我们的消费者中进行处理，以便在 MongoDB 数据库中再次对其进行操作和更新。有没有办法做到这一点，而不必再次迁移所有记录？我们需要分阶段进行迁移，因为它需要大量数据。

这是我们的配置示例：

{
"name": "migration_connector",
"config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "host.docker.internal",
    "database.port": "5432",
    "plugin.name": "pgoutput",
    "database.user": "postgres_user",
    "database.password": "postgres_pasword",
    "database.dbname" : "database_source",
    "database.server.name": "server_name",
    "schema.include": "public",
    "column.include.list": "public.customer.id_customer,public.customer.name,public.table_B.id,public.table_c.id ... and so on",
    "table.include.list": "public.customer,public.table_B,public.table_C",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false",
    "heartbeat.interval.ms": 5000,
    "heartbeat.action.query": "INSERT INTO public.heartbeat_kafka_connect (id, ts) VALUES (1, NOW()) ON CONFLICT(id) DO UPDATE SET ts=EXCLUDED.ts;"
}}

例子：

我们需要从“客户”表中添加另一列，我们需要再次接收所有客户注册以进行处理。当我们更新连接器配置时，寄存器不会再次发送。

【问题讨论】：

标签： postgresql apache-kafka apache-kafka-connect debezium

【解决方案1】：

我们能够弄清楚如何让设置正常工作。

我们使用临时快照功能。我们根据下面的脚本（PostgresSql）在源数据库中创建信号表：

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

我们调整了连接器的配置，使其包含指向我们创建的信号表的标签“signal.data.collection”。

"signal.data.collection": "public.debezium_signal"

我们还在标签“table.include.list”中添加了表名，由于我们使用配置来告诉我们希望 kafka 连接迁移哪些列，因此我们必须在标签中添加信号表的列"column.include.list"

"column.include.list": "public.customer.id,(other columns...),public.debezium_signal.id,public.debezium_signal.type,public.debezium_signal.data",
"table.include.list": "public.customer,public.debezium_signal",

之后，我们只是在信号表中添加了一条记录，指向我们希望 kafka 连接的哪个表重新对记录进行快照。

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

【讨论】：