Kafka Mongodb 接收器连接器 - 更新文档

Question

我们一直致力于开发 kafka 生态系统。让我顺其自然

Source(SQLServer) -> Debezium(CDC) -> Kafka Broker -> Kafka Stream（处理、连接等）-> Mongo 连接器 -> Mongo DB

现在我们进入最后一步，我们正在将处理后的数据插入 mongo dB，但现在我们需要更新数据而不是插入。

我们能否从 mongo sink 连接器获得 upsert（插入/更新）功能。至于现在我明白这是不可能的。

Answer 1

请点击提供的链接，它包含有关 kafka mongo 连接器的所有信息。我已经成功实现了 upsert 功能。您只需要仔细阅读此文档即可。

Kafka Connector - Mongodb

Answer 2

实际上这是一个 upsert，如果 ${uniqueFieldToUpdateOn} 不在 mongo 中，我们要插入，或者如果它存在则更新如下。

根据您的用例更新/替换，有两种主要方法可以对集合中的数据更改进行建模，如下所述：

更新

以下配置状态：

1. 更新 ${uniqueFieldToUpdateOn} 为您要建模更新的记录所特有的字段。
2. AllowList（白名单）此字段与PartialValueStrategy 一起使用允许为 id 策略投影自定义值字段。
3. UpdateOneBusinessKeyTimestampStrategy 意味着只有上面声明的唯一字段引用的一个文档将被更新（最新时间戳获胜）。

"document.id.strategy":"com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy", 
"document.id.strategy.partial.value.projection.list":"${uniqueFieldToUpdateOn}",
"document.id.strategy.partial.value.projection.type":"AllowList",
"writemodel.strategy":"com.mongodb.kafka.connect.sink.writemodel.strategy.UpdateOneBusinessKeyTimestampStrategy" 

替换

注意这个模型是替换而不是更新，但可能仍然有用

以下配置状态：

1. 将 ${uniqueFieldToUpdateOn} 替换为您要为其建模的记录所特有的字段。
2. AllowList（白名单）此字段与PartialValueStrategy 一起使用允许为 id 策略投影自定义值字段。
3. ReplaceOneBusinessKeyStrategy 表示仅替换上面声明的唯一字段引用的一个文档。

"document.id.strategy":"com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy", 
"document.id.strategy.partial.value.projection.list":"${uniqueFieldToUpdateOn}",
"document.id.strategy.partial.value.projection.type":"AllowList",
"writemodel.strategy":"com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy"