问题标签 [mongodb-kafka-connector]

我正在使用MongoDB Kafka 连接器的 beta 版本从 MongoDB 发布到 Kafka 主题。

消息在 Kafka 中生成，但当它应该是文档 ID 时，它们的键为空：

这是我的连接独立配置：

和mongodb 源属性：

下面是一个消息字符串值的示例：

我尝试使用转换从值中提取 if，特别是从 documentKey 字段中提取：

但有一个例外：

有什么想法可以生成带有文档 ID 的密钥吗？

我正在尝试将 kafka-mongo-connector 添加到 kafka-connect 作为事件的来源。我正在设置这样的连接器：

数据进来了，但关键是奇怪的格式：{"_id": {"_data": "....:}}.

因此，我无法插入 Elasticsearch 接收器连接器。

有人有解决方案吗？

谢谢，

弗朗切斯科

我是 Kafka Connect 和 MongoDB 的新手。我在 Kafka 主题中有一条记录，其值为 ，{ "Id": "foo" }我希望将其作为文档存储在 mongo 的集合中时Id映射到。BAR预期结果为{ "BAR": "foo" }。我应该尝试什么，或者如何配置来做到这一点？

我用这个作为参考：https ://github.com/mongodb/mongo-kafka/blob/master/docs/sink.md

我尝试在此处添加 "field.renamer.mapping": "[{\"oldName\":\"Id\", \"newName\": \"BAR\"}]"和 "field.renamer.mapping": "[{\"oldName\":\"value.Id\", \"newName\": \"BAR\"}]"类似于 mongo sink 创建的配置：https ://github.com/mongodb/mongo-kafka/blob/11bac7636f0d6b0e3313c84445777253d36c2042/docker/run.sh#L108 。请求顺利通过，在 mongodb 中创建记录，但未按预期映射。

我有一个 mongo sink 连接器以及一个模式注册表。

我将 mongo sink 连接器配置为访问类似于：https ://github.com/mongodb/mongo-kafka/blob/master/docs/sink.md#configuration-example-for-avro 的模式注册表

我按照以下方式创建了一个模式：https ://github.com/mongodb/mongo-kafka/blob/master/docs/sink.md#logical-types 。它看起来像这样：

但是，当记录通过时，数据看起来像这样：{ "myTimestampMillisField": 1572035138104 }，而不是类似于 this 的东西{ "myTimestampMillisField": ISODate("2019-10-25T20:28:19.628Z") }。

我检查了模式注册表以确保逻辑类型在那里并且看起来不错。

我不确定我做错了什么，或者是否有更好的方法在 mongo 中设置为 Date 类型。有任何想法吗？

我们可以在 mongodb 中更新/更新记录，但是是否有任何方法或函数可以直接在 mongodb 中更新或更新文档，源系统是 kafka，目标是 mongodb。

我们可以限制我们在 debezium mongodb连接器配置中的连接器属性中检索的数据量吗？由于 debezium 在数据库中查找 cdc，根据我的理解，它适用于整个数据库，我找不到限制少数特定集合的方法。我在我的 debezium 连接器中使用 mongodb atlas 连接链接。

以下是我的连接器配置：

如何将其配置为仅查找我的数据库中的特定集合？

我正在使用具有以下配置的 KafkaConnect - MongoSource：

出于某种原因，当我使用消息时，我得到了一个奇怪的键：

我不知道它是从哪里来的，我的 mongo 文档的 _id 是 UUID，当我使用消息时，我应该在我的消费者密钥上看到 documentKey 字段。

以下是连接器发布到 kafka 的消息示例：

我正在使用 mongo 源来监听 mongo 更改流并将所有事件放入 kafka，但我正在绞尽脑汁地寻找一种从事件中提取“Real”键的方法。我尝试了转换，但它没有用，给了我错误：

在 Mongo 源代码中，我发现了这一行

这基本上意味着它甚至没有一些密钥处理，而是寻找“_id”字段（这不是文档的id，它是一个恢复令牌信息）

相反，我想将主题的键设置为“documentKey”。

以下是连接器获取的事件示例：

我使用了以下配置：

我试过了：

还有 StringConverter （虽然我不认为这可以用字符串来完成）

有没有办法提取密钥？请注意：模式已禁用。

发布第一个插入：

{"Customer_id": 2, "transaction_id": "1", "idd": [999, 1111], "id": 1}

然后是第二个：

{"Customer_id": 2, "transaction_id": "2", "idd": [9, 10], "id": 1}

要求的结果：

{"Customer_id": 2, "transaction_id": "2", "idd": [[9, 10] , [999, 1111]], "id": 1}

我得到了什么：

{"Customer_id": 2, "transaction_id": "2", "idd": [9, 10] , "id": 1}

更新策略此更新数组不追加，但我的预期结果追加数组

配置 ：

我们如何通过 mongo-kafka sink 连接器做到这一点

我正在使用带有 Kafka-connect 的 mongo-source 插件。我检查了源任务状态，它正在运行并监听一个 mongo 集合。

我手动停止了 mongod 服务并等待了大约 1 分钟，然后我重新启动它。

我检查了源任务以查看是否有任何东西可以自行修复，30 分钟后似乎没有任何效果。

只有在重新启动连接器后它才重新开始工作。

由于 mongo-source 没有设置超时时重试 + 回退的选项，我搜索了一个适合简单场景的配置：使用 Kafka-connect 配置在 X 时间后重新启动失败的任务。找不到任何.. :/ 我可以用一个简单的脚本来做到这一点，但是 Kafka-connect 中必须有一些东西可以管理失败的任务。甚至在 mongo-source 中......我不希望它在 1 分钟后失败得这么快......：/