使用變更串流讀取即時資料

與 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) 權限。

例如，Datastore 索引管理員 (roles/datastore.indexAdmin) 角色會授予這些權限。

建立變更串流

您必須先建立變更串流，才能開啟對應的變更串流游標。不支援在建立集合或資料庫時，自動啟用變更串流。

如要建立異動串流，請使用 Google Cloud 控制台。

  1. 前往 Google Cloud 控制台的「Databases」(資料庫) 頁面。

    前往「資料庫」

  2. 從清單中選取與 MongoDB 相容的 Firestore 資料庫。「Firestore Studio」面板隨即開啟。
  3. 在「Explorer」面板中，找到「變更串流」 節點，按一下 「More actions」(更多動作)，然後選取「建立變更串流」。
  4. 輸入不重複的異動串流名稱、範圍和保留期限，然後按一下「Save」(儲存)

查看變更串流

您可以在 Google Cloud 控制台查看變更串流的詳細資料。

  1. 前往 Google Cloud 控制台的「Databases」(資料庫) 頁面。

    前往「資料庫」

  2. 從清單中選取與 MongoDB 相容的 Firestore 資料庫。「Firestore Studio」面板隨即開啟。
  3. 在「Explorer」面板中，找出「變更串流」 節點。
  4. 如要開啟或關閉節點，請按一下「 切換節點」。

刪除變更串流

如要刪除變更串流，請使用 Google Cloud 控制台。

  1. 前往 Google Cloud 控制台的「Databases」(資料庫) 頁面。

    前往「資料庫」

  2. 從清單中選取與 MongoDB 相容的 Firestore 資料庫。「Firestore Studio」面板隨即開啟。
  3. 在「Explorer」面板中，找出「變更串流」 節點。
  4. 如要開啟或關閉節點，請按一下「 切換節點」。
  5. 在「Explorer」中，找出要刪除的變更串流。
  6. 按一下「More actions」(更多動作) 圖示 ，然後選取「Delete change stream」(刪除變更串流)。
  7. 在對話方塊中輸入變更串流名稱，確認刪除，然後按一下「Delete」(刪除)。

開啟或繼續變更串流游標

以下範例說明如何建立、繼續及設定變更串流游標。

建立變更串流游標前，您必須明確為資料庫或集合建立變更串流。

建立變更串流游標

如要建立新的變更串流游標，請使用 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 選項，將變更串流查詢分散至多個 worker。每個 worker 負責提供一組不同文件的變更。您必須透過 runCommand 或 aggregate 查詢建立平行游標。

舉例來說，您可以將變更串流分配給 3 個 worker，如下所示：

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 }}
  }]);

變更串流和備份

備份還原作業不提供變更串流設定和變更串流資料。如果還原的資料庫含有變更串流，您必須在目的地資料庫中重新建立這些變更串流，才能開啟該資料庫的游標。

帳單

行為差異

以下各節說明與 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: []
}

後續步驟