仅当新行使用kafka流从mysql数据库中插入或更新旧行时，kafka如何获取数据

Question

我正在使用融合平台 JDBC 连接器，它是将数据从 mysql 流式传输到 kafka 消费者。通过应用程序将数据插入另一个数据库以进行报告。 这里的问题只是在某个时间间隔内一次又一次地流式传输所有数据。实际上只需要那些新插入的数据或以前记录的任何更新。

根据时间戳不能做，因为表不包含任何时间列。并且根据增量 id 也是不可能的。请分享任何解决方案。

我有示例配置文件。

demo.json
{
    "name":"mysql-connector-demo",
    "config":{
                "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
                "key.converter":"org.apache.kafka.connect.json.JsonConverter",
                "value.converter":"org.apache.kafka.connect.json.JsonConverter",
                "connection.url": "jdbc:mysql://localhost:3306/test",
                "connection.user": "root",
                "connection.password": "1234",
                "topic.prefix": "test",
                "catalog.pattern":"test",
                "mode": "bulk",
                "validate.non.null": false,
                "query": "select * from test ",
                "table.types": "TABLE",
                "topic.prefix": "test-jdbc-",
                "poll.interval.ms": 10000
                "schema.ignore": true,
                "key.converter.schemas.enable": "false",
                "value.converter.schemas.enable": "false"
    }
} 

但是这里新插入的记录和新更新的记录不影响kafka消费者。

Answer 1

只想要那些新插入的数据或以前记录的任何更新

那么您应该使用更改数据捕获 (CDC) 而不是 JDBC 轮询。

Debezium 就是这样一种解决方案