问题标签 [changestream]

我创建了一个应用程序，它读取 mongo 更改流以进行更新和插入，然后我们对更改的数据采取行动。下面是我的代码片段

这工作正常。我面临的问题是，当我启动服务器时，更改流开始读取旧提交的更改。我不想要。我希望在部署之后只选择新的更新。

任何人都可以建议如何做到这一点？

在更改流中，我可以提供一个 resumeAfter 令牌，我可以通过访问change._id.

我可以生成自己的基于时间戳的令牌吗？

是否可以使用 Apache Spark 处理 MongoDB 变更流？

我搜索并发现有用于读取和写入 mongoDB 的连接器，但不幸的是我找不到任何关于更改流的提及。

在我的用例中，我必须同时处理来自多个文档的变更流，请建议我是否可以使用 Apache Spark 或其他东西。

由于 MongoDB 变更流上可用的文档有限，他们从 3.6 版开始引入，我无法断定 MongoDB 变更流是否让我们监听系统集合更改。

目前，我对 system.profile 集合感兴趣。

正如文档（https://docs.mongodb.com/manual/changeStreams/）所说，我们可以使用 resumeToken 恢复从 MongoDB 获取事件。

我会将最近的 resumeToken 存储到公共存储中，但我不确定这个恢复令牌是否在全球范围内是唯一的。我想知道它对于整个系列或每个系列都是独一无二的。

它似乎使用 UUID ( https://github.com/mongodb/mongo/blob/master/src/mongo/db/pipeline/resume_token.h#L61 )..

谢谢

我的 Node 应用程序使用 Mongo 更改流，并且该应用程序在生产中运行 3 个以上的实例（最终，随着它的增长，这将成为一个更大的问题）。因此，当更改流中发生更改时，功能运行的次数与进程的次数一样多。

如何设置以使更改流仅运行一次？

这是我所拥有的：

我正在尝试通过设置更改流并（使用ws websocket 库）将这些更改转发到连接的客户端来编写一些与当前正在运行的 MongoDB 数据库一起使用的 nodejs 代码。

我按照本指南创建了初始的 mongo changestream 代码，该代码正在运行：

然后我用 mongo 脚本运行这个文件：

~/Mongo/bin/mongo change-stream-socket-server.js

所以，我基本上想使用上面的这段代码，但添加 websockets 并调用，ws.send()而不是仅仅将更改记录到控制台。

不过，问题是，我似乎无法在 mongo 脚本执行的 javascript 文件中使用“require”。有没有办法让它添加ws依赖或者能够使用require？

当我用 mongo 运行它时，我得到一个错误的输出，“require is not defined”：

MongoDB shell 版本 v3.6.4 连接到：mongodb://127.0.0.1:27017 MongoDB 服务器版本：3.6.4 2018-10-28T21:23:53.515-0400 E QUERY
[thread1] ReferenceError: require is not defined : @change -stream-socket-server.js:2:7 加载失败：change-stream-socket-server.js

如果我尝试仅使用“节点”和文件运行它，则会收到未定义 Mongo 的错误。我尝试使用下面的代码使用“MongoClient”连接到它，但这对我也不起作用。

上面的代码给了我一个错误的输出，“cursor.isExhausted 不是一个函数”：

(node:4590) DeprecationWarning：当前的 URL 字符串解析器已被弃用，并将在未来的版本中删除。要使用新的解析器，请将选项 { useNewUrlParser: true } 传递给 MongoClient.connect。成功连接到服务器 /Users/jim/Git-Projects/MongoDB-Change-Streams-Example/node_modules/mongodb/lib/operations/mongo_client_ops.js:466 throw err; ^

TypeError：cursor.isExhausted 不是 /Users/jim/Git 的 pollStream (/Users/jim/Git-Projects/MongoDB-Change-Streams-Example/basic-mongo-changesream-watcher.js:68:17) 的函数-Projects/MongoDB-Change-Streams-Example/basic-mongo-changesream-watcher.js:47:2 结果（/Users/jim/Git-Projects/MongoDB-Change-Streams-Example/node_modules/mongodb/lib/ utils.js:414:17) 在 executeCallback (/Users/jim/Git-Projects/MongoDB-Change-Streams-Example/node_modules/mongodb/lib/utils.js:406:9) 在错误 (/Users/jim/ Git-Projects/MongoDB-Change-Streams-Example/node_modules/mongodb/lib/operations/mongo_client_ops.js:286:5) 在 connectCallback (/Users/jim/Git-Projects/MongoDB-Change-Streams-Example/node_modules/ mongodb/lib/operations/mongo_client_ops.js:241:5) 在进程中。nextTick (/Users/jim/Git-Projects/MongoDB-Change-Streams-Example/node_modules/mongodb/lib/operations/mongo_client_ops.js:463:7) 在 _combinedTickCallback (internal/process/next_tick.js:131:7)在 process._tickCallback (internal/process/next_tick.js:180:9)

我有什么办法可以同时使用 Mongo 变更流和 ws websocket 库吗？谢谢！

我们在生产中使用 MongoDB，对于某些用例，我们必须在每次更新操作后将更新的文档从应用程序推送到 Kafka。在 MongoDB 3.6 中引入了名为Change Stream的新功能，目前，我们正在尝试使用更改流功能，而不是从应用程序推送更新的文档。

但是更改流仅在更新操作期间返回字段的增量，并且还使用full_document='updateLookup'但这会返回最新的多数提交版本文档，而不是针对该特定事件。

是否有任何方法（或替代方法）来获取特定事件的更新文档，例如 AWS DynamoDB（对于每个更新操作，DynamoDB 流都提供 NewImage）

对于 AWS Dynamodb 表中的每个更新事件，事件包含 OldImage（DynamoDB 表中的项目在修改之前出现的样子）和NewImage（DynamoDB 表中的项目在修改后出现的样子）

但是在 MongoDB Change Stream 的情况下，full_document='updateLookup' 作为选项传递时将返回更新文档的最新多数提交版本

请参阅https://docs.mongodb.com/manual/changeStreams/#lookup-full-document-for-update-operations

我已经在我的 Spring Boot 应用程序中添加了 mongodb 依赖项，但是在“where”方法上出现未定义的错误：

POM 配置：

请给我建议

我是 Node.js 的新手，并尝试使用 Node.js Mongodb 驱动 API changeStream ( https://mongodb.github.io/node-mongodb-native/3.0/api/ChangeStream.html#hasNext )。我不知道如何处理 Nodejs 返回的 Promise。我的代码如下。

输出是：承诺 { }

有人可以帮忙吗？谢谢！