Kafka JDBC 源连接器插入或更新
Posted
技术标签:
【中文标题】Kafka JDBC 源连接器插入或更新【英文标题】:Kafka JDBC Source connector Insert or Update 【发布时间】:2021-01-17 04:48:47 【问题描述】:我配置了一个 Kafka JDBC 源连接器,以便将来自 PostgreSQL 数据库的记录更改(插入或更新)推送到 Kafka 主题。 我使用“时间戳+递增”模式。似乎工作正常。 我没有配置 JDBC Sink 连接器,因为我使用的是监听该主题的 Kafka Consumer。
关于该主题的消息是 JSON。这是一个例子:
"schema":
"type": "struct",
"fields": [
"type": "int64",
"optional": false,
"field": "id"
,
"type": "int64",
"optional": true,
"name": "org.apache.kafka.connect.data.Timestamp",
"version": 1,
"field": "entity_create_date"
,
"type": "int64",
"optional": true,
"name": "org.apache.kafka.connect.data.Timestamp",
"version": 1,
"field": "entity_modify_date"
,
"type": "int32",
"optional": true,
"field": "entity_version"
,
"type": "string",
"optional": true,
"field": "firstname"
,
"type": "string",
"optional": true,
"field": "lastname"
],
"optional": false,
"name": "author"
,
"payload":
"id": 1,
"entity_create_date": 1600287236682,
"entity_modify_date": 1600287236682,
"entity_version": 1,
"firstname": "George",
"lastname": "Orwell"
如您所见,没有关于源连接器是否由于插入或更新而捕获此更改的信息。 我需要这些信息。怎么解决?
【问题讨论】:
【参考方案1】:您无法使用 JDBC 源连接器获取该信息,除非您在源架构和触发器中执行了定制操作。
这是为什么基于日志的 CDC 通常是从源数据库获取事件的更好方法的原因之一,以及其他原因包括:
捕获删除 捕获操作类型 捕获所有事件,而不仅仅是连接器轮询时的事件。有关此细微差别的更多详细信息,请参阅this blog 或talk based on the same。
【讨论】:
【参考方案2】:使用@Robin Moffatt 建议的基于 CDC 的方法可能是处理您的要求的正确方法。结帐https://debezium.io/
但是,查看表数据时,您可以在消费者中使用“entity_create_date”和“entity_modify_date”来确定消息是插入还是更新。如果 "entity_create_date" = "entity_modify_date" 则为插入,否则为更新。
【讨论】:
以上是关于Kafka JDBC 源连接器插入或更新的主要内容,如果未能解决你的问题,请参考以下文章
无法使用镜头 kudu sink 连接器将数据从 kafka 主题插入或更新到 kudu 表