使用变更数据流读取实时数据

借助与 MongoDB 兼容的 Firestore 的变更数据流，应用可以访问对集合或整个数据库所做的实时更改（插入、更新和删除）。变更数据流会按修改时间对更新进行排序。

您可以通过与 MongoDB 兼容的 API 和传统 MongoDB 驱动程序访问变更数据流。与 MongoDB 兼容的 Firestore 变更数据流实现可以通过独特的写入自动分区和读取并行实现来处理任何吞吐量的写入和读取。这样，您就可以构建高吞吐量工作负载。 您还可以改进 Cloud Firestore 与其他存储解决方案之间的迁移和数据同步基础架构。

除了与 MongoDB 驱动程序兼容之外，您还可以使用 Cloud Firestore 并行读取变更数据流。这样，您就可以构建并行的高吞吐量读取工作负载。每个流都代表一个分布良好的结果分区。

变更数据流支持以下功能：

• 可配置的变更数据流，范围为数据库或集合。
• 在创建时指定的变更数据流保留时长。 默认保留期限为 7 天，最短保留期限为 1 天。保留时间必须是 1 天的倍数，最长为 7 天。保留时长一经创建便无法更改。如需更改保留期限，您必须删除并重新创建变更数据流。
• delete、insert、update 和 drop 更改事件，可使用 db.collection.watch() 和 db.watch() 进行观察。
• updateDescription.updatedFields 包含更新差异。
• 所有 fullDocument 和 fullDocumentBeforeChange 选项。
  • 正在查找完整文档以获取更新。
  • 文档在被替换、更新或删除之前的预映像。
  • 文档被替换或更新后的后映像。
  • 对于超过一小时的前后映像，需要启用时间点恢复 (PITR)。
• 所有恢复选项，包括 resumeAfter 和 startAfter。
• 使用 watch() 观察更改时，您可以链接聚合阶段，例如 $addFields、$match、$project、$replaceRoot、$replaceWith、$set 和 $unset。

配置变更数据流

如需为数据库创建、删除或查看现有变更数据流，请使用 Google Cloud 控制台。

角色与权限

如需创建、删除和列出变更数据流，主账号分别需要 datastore.schemas.create、datastore.schemas.delete 和 datastore.schemas.list Identity and Access Management (IAM) 权限。

例如，数据存储区索引 Admin (roles/datastore.indexAdmin) 角色可授予这些权限。

创建变更数据流

您必须先创建变更数据流，然后才能打开相应的变更数据流游标。不支持在创建集合或数据库时自动启用变更数据流。

如需创建变更数据流，请使用 Google Cloud 控制台。

1. 在 Google Cloud 控制台中，转到数据库页面。

   前往“数据库”

2. 从列表中，选择一个与 MongoDB 兼容的 Firestore 数据库。Firestore Studio 面板随即打开。
3. 在探索器面板中，找到变更数据流节点，点击 more_vert 更多操作，然后选择创建变更数据流。
4. 输入唯一的变更数据流名称、范围和保留期限，然后点击保存。

查看变更数据流

您可以在 Google Cloud 控制台中查看有关变更数据流的详细信息。

1. 在 Google Cloud 控制台中，转到数据库页面。

   前往“数据库”

2. 从列表中，选择一个与 MongoDB 兼容的 Firestore 数据库。Firestore Studio 面板随即打开。
3. 在探索器面板中，找到变更数据流节点。
4. 如需打开或关闭节点，请点击 arrow_drop_down 切换节点。

删除变更数据流

如需删除变更数据流，请使用 Google Cloud 控制台。

1. 在 Google Cloud 控制台中，转到数据库页面。

   前往“数据库”

2. 从列表中，选择一个与 MongoDB 兼容的 Firestore 数据库。Firestore Studio 面板随即打开。
3. 在探索器面板中，找到变更数据流节点。
4. 如需打开或关闭节点，请点击 arrow_drop_down 切换节点。
5. 在探索器中，找到要删除的变更数据流。
6. 点击 more_vert 更多操作，然后选择删除变更数据流。
7. 在对话框中，输入变更数据流名称以确认删除，然后点击删除。

打开或恢复变更数据流游标

以下示例演示了如何创建、恢复和配置变更数据流游标。

在创建变更数据流游标之前，您必须为数据库或集合显式创建变更数据流。

创建变更数据流游标

如需创建新的变更数据流游标，请使用 MongoDB 驱动程序中的 watch 方法。如需监听数据库中的所有变更，请创建数据库级变更数据流，并在 db 对象上调用 watch 方法。

let cursor = db.watch()

如需创建限定于某个集合的游标，您必须先为该集合创建变更数据流。然后，对相应集合调用 watch 方法。

let cursor = db.my_collection.watch()

现在，您已创建变更数据流游标，可以开始流式传输了。 例如，如果您插入文档并在游标上调用 tryNext，则会看到更改出现在变更数据流中。

let doc = db.my_collection.insertOne({value: "hello world"})
console.log(cursor.tryNext())

如果您更新和删除文档，则会看到这些更改显示在变更数据流中：

db.my_collection.updateOne({"_id": doc.insertedId}, {$set: {value: "hello world!"}})
db.my_collection.deleteOne({"_id": doc.insertedId}})

// Prints the update event
console.log(cursor.tryNext())

// Prints the delete event
console.log(cursor.tryNext())

恢复变更数据流

如需恢复变更数据流，请使用 resumeAfter 或 startAfter 选项。如需确定从 resumeAfter 和 startAfter 的更新日志中的哪个位置继续，请使用续传令牌。

// Create a cursor and add one event to the change stream.
let cursor = db.my_collection.watch();
db.my_collection.insertOne({value: "hello world"});
let event = cursor.tryNext();

// Get the resume token from the event.
let resumeToken = event._id;

// Add a new event to the change stream.
db.my_collection.insertOne({value: "foobar"});

// Create a new cursor by using the resume token as a starting point.
let newCursor = db.my_collection.watch({resumeAfter: resumeToken})

// Log the change event containing the "foobar" value.
console.log(newCursor.tryNext())

使用 startAfter：

// Start after the resume token.
let startAfterCursor = db.my_collection.watch({startAfter: resumeToken})

在更新和删除操作中包含前置和后置映像

如果需要，您可以在更新和删除更改事件中包含文档的前后映像。映像可用性取决于时间点恢复 (PITR) 窗口，如需读取超过一小时的文档映像，您必须启用 PITR。

变更数据流利用 PITR 窗口提供给定变更事件前后的文档视图。默认情况下，更新事件包含一个 updateDescription 字段，该字段是更新操作修改的字段的增量。

如需在变更事件中包含前置和后置映像，您必须在变更数据流查询中指定 fullDocumentBeforeChange 和 fullDocument 选项。

let cursor = db.my_collection.watch({
  "fullDocument": "required",
  "fullDocumentBeforeChange": "required"
})

如果查询尝试读取 PITR 保留窗口之外的文档，或者未启用 PITR，则 required 值会抛出服务器端出错提示。

除了抛出错误之外，您还可以使用 whenAvailable 值在图片不再可用时返回 null 值。

let cursor = db.my_collection.watch({
  "fullDocument": "whenAvailable",
  "fullDocumentBeforeChange": "whenAvailable"
})

在更新中包含当前图片

默认情况下，更新事件包含一个 updateDescription 字段，该字段是更新操作修改的字段的增量。如需改为查找整个文档的最新版本，请在 fullDocument 选项中使用 updateLookup 值。

此功能不需要 PITR，而是对文档执行查找。

let cursor = db.my_collection.watch({
  "fullDocument": "updateLookup",
})

并行读取

如需提高吞吐量，您可以使用 firestoreWorkerConfig 选项在多个工作器之间拆分变更数据流查询。每个工作器负责提供一组不同文档的更改。您必须通过 runCommand 或 aggregate 查询创建并行游标。

例如，您可以将变更数据流分布到 3 个工作器，如下所示：

let cursor1 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 0 }}
  }]);

let cursor2 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 1 }}
  }]);

let cursor3 = db.my_collection.aggregate([{
    "$changeStream": {
        "firestoreWorkerConfig": {numWorkers: 3, workerId: 2 }}
  }]);

变更数据流和备份

备份恢复操作中不提供变更数据流配置和变更数据流数据。如果您恢复具有变更数据流的数据库，则必须在目标数据库中重新创建这些变更数据流，才能打开指向该数据库的游标。

结算

• 变更数据流会产生读取单位和存储费用。请参阅变更数据流价格。
• 如需在读取请求时包含超过 1 小时的前映像和后映像，您必须启用 PITR，这会产生 PITR 费用。

行为差异

以下部分介绍了与 MongoDB 兼容的 Firestore 和 MongoDB 之间的变更数据流差异。

updateDescription

updateDescription 是 update 事件中的一个文档，用于描述更新操作所更新或移除的字段。在 Cloud Firestore 中，显著差异如下：

• 在 updateDescription 中，字段 truncatedArrays 和 disambiguatedPaths 未填充。
• updateDescription.updatedFields 表示应用变更前后文档的前后映像之间的规范差异。

假设文档的初始状态如下：

db.my_collection.insertOne({
  _id: 1,
  root: {
    array: [{a: 1}, {b: 2}, {c: 3}]
  }
})

场景 1：仅更改数组的第一个元素。

在这种情况下，Cloud Firestore 行为与 MongoDB 相同。

db.my_collection.updateOne(
  {_id: 1},
  {'$set': {"root.array.0.a": 100}}
)

{
  updatedFields: {"root.array.0.a": 100},
  removedFields: []
}

场景 2：用整个数组覆盖

在这种情况下，该操作仅更新第一个数组字段，但会覆盖整个数组。

Cloud Firestore 更新差异不会区分这两种情况，并会为这两种情况返回相同的 updateDescription.updatedFields：

db.my_collection.updateOne(
  {_id: 1},
  {'$set': {"root.array": [{a: 100}, {b: 2}, {c: 3}]}}
)

// In other implementations, updatedFields reflects the mutation itself
{
  updatedFields: {
    "root.array": [{a: 100}, {b: 2}, {c: 3}]
  },
  removedFields: []
}

// Firestore updatedFields is the diff between the before and after versions of the document
{
  updatedFields: {"root.array.0.a": 100},
  removedFields: []
}

后续步骤

• 理解大规模读写操作
• 优化查询执行