使用 MongoDB Sink Connector 通过不同的主键更新现有文档答案

【问题标题】：Using MongoDB Sink Connector to update existing documents by a different primary key使用 MongoDB Sink Connector 通过不同的主键更新现有文档
【发布时间】：2021-03-08 12:10:37
【问题描述】：

我正在尝试通过 Confluent Cloud 设置 MongoDB Sink 连接器，以保持 pgsql 和 MongoDB 之间的数据同步。

我希望下面的配置根据id (int) 字段（不是 _id - objectId）更新现有文档，但是它只是在使用时在 MongoDB 中创建一个新文档。来自 pg 的文档不会包含 _id 字段，因此我们需要在我们的 pgsql 主键 (id) 上完成查找。

任何想法为什么这不能像我预期的那样工作？

 {
  "connector.class": "MongoDbAtlasSink",
  "name": "mongodb-sink",
  "kafka.api.key": "mykey",
  "kafka.api.secret": "mysecret",
  "input.data.format": "JSON",
  "topics":"mytopic",
  "connection.host": "myhost",
  "connection.user": "myuser",
  "connection.password": "mypassword",
  "database": "mydatabase",
  "delete.on.null.values": "false",
  "tasks.max": "1",
  "document.id.strategy": "com.mongodb.kafka.connect.sink.processor.id.strategy.PartialValueStrategy",
  "document.id.strategy.partial.value.projection.list":"id",
  "document.id.strategy.partial.value.projection.type":"AllowList",
"writemodel.strategy":"com.mongodb.kafka.connect.sink.writemodel.strategy.ReplaceOneBusinessKeyStrategy"
}

【问题讨论】：

尝试向 id 列添加唯一索引，但没有成功。

标签： mongodb apache-kafka confluent-platform confluent-cloud

【解决方案1】：

key.projection.type 和 value.projection.type 在云版本中显然不受支持，因此不适用于该配置。

Cloud Connector Limitations

【讨论】：