带有 TimescaleDB 扩展的 Debezium 连接器

Question

我在检测 Postresql 超表（TimescaleDB 扩展）上的更改时遇到问题。

设置：

我安装了带有 TimescaleDB(ver 1.7.1) 扩展的 Postresql(ver 11.10)。 我有 2 个表，我想使用安装在 Kafka Connect 上的 Debezium（ver 1.3.1）连接器来监控它们，用于 CDC（捕获数据更改）。

表是 table1 和 table2hyper，但 table2hyper 是超表。

在 Kafka Connect 中创建 Debezium 连接器后，我可以看到创建了 2 个主题（每个表一个）：

• (A) kconnect.public.table1
• (B) kconnect.public.table2hyper

当使用 kafka-console-consumer 为主题 A 消费消息时，我可以在 table1 中看到行更新后的消息。

但是当使用来自主题 B（table2hyper 表更改）的消息时，例如 table2hyper 表中的行更新之后不会发出任何内容。

最初 Debezium 连接器对 table2hyper 表中的行进行快照并将它们发送到主题 B（使用 kafka-console-consumer 时我可以看到主题 B 中的消息），但是我在初始快照之后所做的更改不会发出.

为什么我无法从 table2hyper 看到后续更改（在初始快照之后）？

连接器创建负载：

{
    "name": "alarm-table-connector7",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "tasks.max": "1",
        "database.hostname": "xxx",
        "database.port": "5432",
        "database.user": "xxx",
        "database.password": "xxx",
        "database.dbname": "xxx",
        "database.server.name": "kconnect",
        "database.whitelist": "public.dev_db",
        "table.include.list": "public.table1, public.table2hyper",
        "plugin.name": "pgoutput",
        "tombstones.on.delete":"true",
        "slot.name": "slot3",
        "transforms": "unwrap",
        "transforms.unwrap.type":"io.debezium.transforms.ExtractNewRecordState",
        "transforms.unwrap.drop.tombstones":"false",
        "transforms.unwrap.delete.handling.mode":"rewrite",
        "transforms.unwrap.add.fields":"table,lsn,op"
    }
}

提前谢谢！

Answer 1

尝试了一段时间后，我没有成功使用Debezium连接器从超表中流式传输数据。我使用的是 1.3.1 版。并升级到最新的 1.4.1。没有帮助。

不过，Confluent JDBC connector 确实成功了。

就我的研究和测试而言，这是结论，如有必要，请随时纠正我：

• Debezium 在 INSERT、UPDATE 和 删除活动
• Confluent 连接器仅捕获 INSERT 事件（除非您结合 一些用于检测变化的列）并适用于普通和 超级（TimescaleDB）表。

Answer 2

我们从未使用 TimescaleDB 测试过 Debezium。我建议您检查逻辑复制槽中是否存在 TimescaleDB 更新。如果是的话，让 Debezium 处理这些事件在技术上应该是可行的。如果没有，那根本不可能。